massiv bugfixes around getter methods and ThermalCap outpost handling

libs/utils/general.py

@@ -0,0 +1,4 @@
+console = get_component("console")
+
+def default():
+    console.clear()

libs/utils/item.py

@@ -0,0 +1,59 @@
+from __builtins__ import OutpostComponent, OutpostRef, LargeWarehouse, Warehouse, StorageBin
+
+class Item:
+    @staticmethod
+    def get_storages(
+            item: str, outpost: OutpostComponent | OutpostRef | str = None, 
+            sbin: bool = True, whouse: bool = True,
+            nearest: tuple[int, int] | None = None
+        ) -> list[StorageBin | Warehouse | LargeWarehouse]:
+        """get all storages (without inventory) for with a given item
+        
+        Args:
+            item (str): item id
+            outpost (OutpostComponent | OutpostRef | str): outpost to search for storages
+            sbin (bool): control StorageBin search. default True
+            whouse (bool): control Warehouese & LargeWarehouse search. default True
+            TODO: nearest (tuple[int, int] | None): on True get only the nearest storage for position (x, y). None = get all. default None (planed!)
+        Return:
+            list[StorageBin | Warehouse | LargeWarehouse]
+        """
+        # check type of item
+        if not isinstance(item, str):
+            return list()
+
+        # check type of outpost
+        if outpost is None:
+            # get all outposts
+            for op in get_component("outpost_network").outposts():
+                outposts = get_component("outpost_network").outposts()
+        else:
+            # ensure outpost is a OutpostComponent
+            if isinstance(outpost, str): outpost = get_component(outpost)
+            if isinstance(outpost, OutpostRef): outpost = get_component(outpost.id)
+            if not isinstance(outpost, OutpostComponent):
+                return list()
+            outposts = [outpost]
+
+        storages = list()
+        for outpost in outposts:
+            # search for StorgageBins
+            if sbin:
+                for b in outpost.buildings("storage_bin"):
+                    storage = get_component(b.id)
+                    if item == storage.get_material(): storages.append(storage)
+
+            # search for Warehouse and LargeWarehouse
+            if whouse:
+                for w in outpost.buildings("warehouse"):
+                    storage = get_component(w.id)
+                    if item in storage.materials(): storages.append(storage)
+
+        # check if nearest is a tuple and has 2 int insert
+        if isinstance(nearest, tuple) and len(nearest) == 2 and all(isinstance(v, int) for v in nearest):
+            pass # TODO
+        return storages
+
+    def count(item, outpost: str = None):
+        pass
+

libs/utils/map.py

@@ -0,0 +1,120 @@
+from __builtins__ import MiningSite, PointOfInterest
+import json
+
+
+
+class Map:
+    """Utility class for map interactions"""
+    journal = get_component("journal")
+    planet = get_component("nocturna")
+    marks = get_component("markers")
+    outposts = get_component("outpost_network")
+    note = get_component("notebook")
+    construct = get_component("construction_blueprint")
+        
+    ## point of intrest methods
+    @staticmethod
+    def get_unknown_pois() -> list[PointOfInterest]:
+        """return all unscanned points of intrests"""
+        return [poi for poi in Map.planet.points_of_interest() if not poi.scanned]
+
+    @staticmethod
+    def get_known_pois() -> list[PointOfInterest]:
+        """return all scanned points of intrests"""
+        return [poi for poi in Map.planet.points_of_interest() if poi.scanned]
+
+    @staticmethod
+    def filter_blocklist_pois(pois: list[PointOfInterest], sonar: str) -> list[PointOfInterest]:
+        if not isinstance(pois, list):
+            return list()
+        blocklist = Map.get_blocklist_pois(sonar)
+        if not isinstance(blocklist, list) or len(blocklist) == 0:
+            return pois
+        filterd_pois = list()
+        for poi in pois:
+            if not isinstance(poi, PointOfInterest):
+                continue
+            if not f"{poi.x}{poi.y}" in blocklist:
+                filterd_pois.append(poi)
+        return filterd_pois
+    
+    @staticmethod
+    def get_blocklist_pois(sonar: str) -> list[str]:
+        """get blocklisted point of interests for sonar level
+        
+        Args:
+            sonar (int): sonar tier. get via vehicle.sonar.tier()
+        Return:
+            list[str]: list of coords from blocklist, point of intersts like str(x) + str(y)
+        """
+        if not sonar in ["basic", "wide", "deep", "seismic"]:
+            return list()
+        pointer = f"map.scan.blocked.{sonar}"
+        if not Map.note.has(pointer):
+            return list()
+        response = json.loads( Map.note.get(pointer) )
+        return response
+
+    @staticmethod
+    def add_blocklist_poi(poi: PointOfInterest, sonar: str) -> bool:
+        """add a new poi to blocklist for sonar level
+        
+        Args:
+            poi (PointOfInterest): poi to blacklist
+            sonar (int): sonar tier. get via vehicle.sonar.tier()
+        Return:
+            bool: True if saved successfully
+        """
+        if not isinstance(poi, PointOfInterest):
+            return False
+        if not sonar in ["basic", "wide", "deep", "seismic"]:
+            return False
+        blocklist = Map.get_blocklist_pois(sonar)
+        blocklist.append(f"{poi.x}{poi.y}")
+        blocklist = json.dumps( blocklist )
+        response = Map.note.set( f"map.scan.blocked.{sonar}", blocklist )
+        if response == "ok": return True
+        return False
+
+    ## mining methods
+    @staticmethod
+    def get_mines_by_ressource(ressource: str) -> list[MiningSite]:
+        """collect all known mines with ressource type
+        [[helper]]
+
+        Args:  
+            ressources (str):  ressource id name like `iron_ore` or `silicon`
+        Return:
+            `list[MiningSite]`: known mines of type `ressource`   
+        """
+        mines_by_item = dict()
+
+        # collect all MiningSites and sort by type
+        for site in Map.journal.discovered_sites("nocturna"):
+            if not isinstance(site, MiningSite):
+                continue
+            if site.item_id not in mines_by_item:
+                mines_by_item[site.item_id] = list()
+
+            mines_by_item[site.item_id].append(site)
+
+        # check if ressource in mines and return a list
+        if ressource in mines_by_item.keys():
+            return mines_by_item[ressource]
+        else: return list()
+
+    @staticmethod
+    def get_mines_iron() -> list[MiningSite]:
+        """short method for `get_mines_by_ressources` directly for ìron_ore`"""
+        return Map.get_mines_by_ressource("iron_ore")
+
+    @staticmethod
+    def get_mines_silicon() -> list[MiningSite]:
+        """short method for `get_mines_by_ressources` directly for `silicon`"""
+        return Map.get_mines_by_ressource("silicon")
+
+    @staticmethod
+    def get_mines_titan() -> list[MiningSite]:
+        """short method for `get_mines_by_ressources` directly for `titanium`"""
+        return Map.get_mines_by_ressource("titanium")
+  

libs/wrapper/power.py

@@ -0,0 +1,302 @@
+from __builtins__ import SolarGenerator, ThermalCap, SteamTurbine, GasTank, OutpostComponent, OutpostRef
+
+class SolarGeneratorWrapper:
+    """wrapper class for SolarGenerator"""
+    clock = get_component("clock")
+    target_tilt = 90
+    
+    def __init__(self, solar: SolarGenerator | str):
+        """inital wrapper for `SolarGenerator`
+        
+        Args:
+            solar (`SolarGenerator` | `str`): a `SolarGenerator` to adjust. component or id. usally self.
+        """
+        self.solar = solar
+    
+    def run(self) -> None:
+        """run loop for `SolarGenerator`"""
+        while True:
+            self.adjust()
+
+    def adjust(self) -> None:
+        """adjust SolarGenerator to sun elevation"""
+        debug(f"adjust SolarGenerator to {self.clock.get_elevation()=}")
+        self.solar.set_tilt( self.target_tilt - self.clock.get_elevation() )
+
+    @property
+    def solar(self) -> SolarGenerator:
+        if isinstance(self.__solar, SolarGenerator): return self.__solar
+        raise TypeError(f"f{self.__solar=} ({type(self.__solar)=}) is not a SolarGenerator")
+
+    @solar.setter
+    def solar(self, value: SolarGenerator | str):
+        if isinstance(value, str): value = get_component(value)
+        if not isinstance(value, SolarGenerator): raise TypeError(f"{value=} ({type(value)=}) is not a SolarGenerator")
+        self.__solar = value
+        
+
+# in power ?
+class ThermalCapWrapper:
+    """wrapper class for ThermalCap"""
+    clock = get_component("clock")
+    default_threshold = 0.98
+
+    def __init__(self, cap: ThermalCap | str, threshold: float = 0.99, outposts: list[OutpostComponent] | OutpostComponent | OutpostRef | str | None = None):
+        """init wrapper for `ThermalCap`
+        
+        Args:
+            cap (`ThermalCap`): thermal cap component. usally self.
+            threshold (`float`, optional): percent level what thermal cap hold this pressure level. (0-1). default 0.99.
+            outpost (`OutpostComposnent` | `OutpostRef` | `str` | `None`, optional): outpost for steam out. None = nearest. default None.
+        """
+        self.cap = cap
+        self.threshold = threshold
+        self.outposts = outposts
+        self.steam_out = cap.steam_out
+
+    def run(self):
+        """run loop for `ThermalCap`"""
+        while True:
+            match self.cap.phase():
+                case "dormant":
+                    self.phase_dormant()
+                case "active":
+                    self.phase_active()
+                    self.check_storage()
+
+    def phase_dormant(self) -> None:
+        """run logik for `dormant` phase"""
+        self.cap.set_relief(0)
+        sleep_time = (self.cap.next_phase_in() / 60) * self.clock.real_seconds_per_hour()
+        sleep(sleep_time)
+
+    def phase_active(self) -> None:
+        """run logik for `active` phase"""
+        if self.cap.pressure() > self.threshold:
+            # ntk: fix 2k output rate because it seams fixed and no function give it back. 
+            # hint: It can be found by set_relief(1) and read from relief_rate()
+            self.cap.set_relief( (self.cap.capture_rate() - self.steam_out.flow_rate()) / 2000) 
+        else:
+            self.cap.set_relief(0)
+        
+    def fill_storage(self, storage: GasTank) -> None:
+        """fills the given `GasTank` until full"""
+        if not isinstance(storage, GasTank): return
+        if storage.is_full():
+            debug(f"{storage.id=} if full, disconnect.")
+            self.cap.set_throttle(0)
+            self.steam_out.disconnect()
+        else:
+            debug(f"start filling {storage.id=}")
+            self.steam_out.connect(storage.id)
+            self.cap.set_throttle(1.0)
+            self.cap.set_relief(0)
+
+        while not storage.is_full() and self.cap.phase() == "active":
+            debug(f"wait until {storage.id=} is full")
+            sleep(1)
+        debug(f"{storage.id=} is full, disconnect.")
+        self.steam_out.disconnect()
+
+    def check_storage(self) -> None:
+        """search a `GasTank` to fill"""
+        if self.cap.pressure() < 0.1: return
+        gas_tanks = list()
+        for outpost in self.outposts:
+            gas_tanks.extend([ get_component(gt.id) for gt in outpost.buildings('gas_tank') ])
+
+        for tank in gas_tanks:
+            # TODO: atm get only prefilled steam gas tanks
+            # TODO: if check on none, all new will be filled with steam
+            if not tank.fluid() == "steam": continue 
+            if tank.fill_pct() < 0.8:
+                self.fill_storage(tank)
+                break
+
+    def get_nearest_outpost(self):
+        """get nearest outpost to `ThermalCap` vent"""
+        vent = self.cap.vent()
+        outpost = get_component(get_component("outpost_network").nearest(vent.x, vent.y).id)
+        if isinstance(outpost, OutpostComponent):
+            debug(f"nearest {outpost=}")
+            return outpost
+        else: return None
+        
+    @property
+    def cap(self) -> ThermalCap:
+        if isinstance(self.__cap, ThermalCap): return self.__cap
+        raise TypeError(f"{self.__cap=} ({type(self.__cap)=} is not a ThermalCap)")
+
+    @cap.setter
+    def cap(self, value: ThermalCap | str):
+        if isinstance(value, str): value = get_component(value)
+        if not isinstance(value, ThermalCap): raise TypeError(f"{value=} ({type(value)=}) is not a ThermalCap")
+        self.__cap = value
+
+    @property
+    def threshold(self) -> float:
+        try: raw = self.__threshold
+        except AttributeError: return self.default_threshold
+    
+        if not isinstance(raw, (float, int)) or isinstance(raw, bool):
+            return self.default_threshold
+        if not 0 < raw < 1: return self.default_threshold
+        return raw
+
+    @threshold.setter
+    def threshold(self, value: float):
+        if not isinstance(value, float) and not 0 < value < 1: 
+            warn(f"set threshold: {value=} ({type(value)=}) is not a float, use {self.default_threshold=}")
+            self.__threshold = self.default_threshold
+            return
+        self.__threshold = value
+
+    @property
+    def outposts(self) -> list[OutpostComponent]:
+        try: raw = self.__outposts
+        except AttributeError: return [self.get_nearest_outpost()]
+    
+        if not isinstance(raw, list): return [self.get_nearest_outpost()]
+        return [o for o in raw if isinstance(o, OutpostComponent)]
+
+    @outposts.setter
+    def outposts(self, value: list[OutpostComponent] | OutpostComponent | OutpostRef | str):
+        if isinstance(value, list):
+            debug(f"apply all OutpostComponents to self.__outposts: {value=}")
+            self.__outposts = [ o for o in value if isinstance(o, OutpostComponent) ]
+            return
+        if isinstance(value, str): value = get_component(value)
+        if isinstance(value, OutpostRef): value = get_component(value.id)
+        if not isinstance(value, OutpostComponent):
+            warn(f"no outpost for steam out found ({value=}, use default nearest")
+            self.__outposts = [ self.get_nearest_outpost() ]
+            return
+        else: self.__outposts = [ value ]
+
+class SteamTurbineWrapper:
+    """wrapper class wor `SteamTurbine`"""
+    default_threshold = 25
+    notify_threshold = 10
+    
+    def __init__(self, turbine: SteamTurbine, threshold: int = 25, storages: list[GasTank] = None):
+        """init wrapper for `SteamTurbine`
+        
+        Args:
+            turbine (`SteamTurbine`): `SteamTurbine` component. usally self.
+            threshold (`int`, optinoal): under this threshold the `SteamTurbine` try to get new steam. default 25.
+            storages (`list[GasTank]`, optional): `GasTanks` from where the `SteamTurbine` can get new steam. default all local prefiled `GasTank`s
+        """
+        self.turbine = turbine
+        self.threshold = threshold
+        self.storages = storages
+        self.pcontrol = get_component("power_control")
+        self.steam_in = turbine.steam_in
+
+    def run(self):
+        """run loop for `SteamTurbine`"""
+        while True:
+            self.check_health()
+            self.fill_steam()
+            self.set_energie()
+
+    def check_health(self):
+        """check own function and notify if not in expected parameters"""
+        if self.steam_in.level() < self.notify_threshold:
+            notify(f"{self.turbine.id}: steam level under {self.notify_threshold}", "warn", 30)
+        if self.turbine.is_stalled() or self.steam_in.level() == 0:
+            notify(f"{self.turbine.id}: stopped working or has low efficency ({self.turbine.efficiency()*100}%)", "error", 30)
+            sleep(30)
+    
+    def fill_steam(self):
+        """fills steam if own steam under threshold"""
+        if self.steam_in.level() < self.threshold:
+            for storage in self.storages:
+                if storage.level() > self.steam_in.capacity():
+                    self.steam_in.connect(storage.id)
+                    break
+        elif self.steam_in.level() > 99:
+            self.steam_in.disconnect()
+
+    def set_energie(self):
+        """set energie output relative to all online `SteamTurbines` in same power network"""
+        power = self.pcontrol.grid(self.turbine.id).net
+        if power < 0:
+            power_part = abs(( power / max( len(self.get_net_turbines()),1 ) ) - self.turbine.power_output())
+            self.turbine.set_throttle( min(power_part / 108, 1) )
+        elif power > 10:
+            self.turbine.set_throttle( 0 )
+
+    def get_net_turbines(self) -> list[SteamTurbine]:
+        """collect all turbines with poweroutut
+        
+        Return:
+            `list[SteamTurbine]`: all turbines in same power network and power output
+        """
+        # TODO: Check is have steam and is online
+        turbines = [ get_component(t) for t in get_component("power_control").grid(self.turbine.id).machine_ids if t.startswith("turbine_") ]
+        ## TEST: ggf. nicht gut genug als indikator z.B. wenn !!effizienz nicht 100%!! ist oder grade in den übergangsphasen kann es zu flapping kommen.
+        # zusatz prüfung auf effizienz und steamout.level() ? 
+        return [ t for t in turbines if t.steam_in.level() > self.notify_threshold ]
+        return turbines
+    
+    def get_local_storages(self, prefiled: bool = True) -> list[GasTank]:
+        """get all `GasTanks` in same outpost as the `SteamTrubine`
+        
+        Args:
+            prefiled (`bool`): True to get only steam prefilled `GasTanks`
+        Return:
+            `list[GasTank]`
+        """
+        gastanks = [ get_component(s.id) for s in get_component(self.turbine.outpost.id).buildings("gas_tank") ]
+        if prefiled:
+            return [ s for s in gastanks if s.fluid() == "steam" ]
+        else: return gastanks
+
+    @property
+    def turbine(self) -> SteamTurbine:
+        if isinstance(self.__turbine, SteamTurbine): return self.__turbine
+        raise TypeError(f"{self.__turbine=} ({type(self.__turbine)=} is not a SteamTurbine)")
+
+    @turbine.setter
+    def turbine(self, value: SteamTurbine | str):
+        if isinstance(value, str): value = get_component(value)
+        if not isinstance(value, SteamTurbine): raise TypeError(f"{self.__turbine=} ({type(self.__turbine)=} is not a SteamTurbine)")
+        self.__turbine = value
+
+    @property
+    def threshold(self) -> int:
+        try: raw = self.__threshold
+        except AttributeError: return self.default_threshold
+    
+        if not isinstance(raw, int) or isinstance(raw, bool):
+            return self.default_threshold
+        if not 0 < raw < 100: return self.default_threshold
+        return raw
+
+    @threshold.setter
+    def threshold(self, value: int):
+        if not isinstance(value, int) or not 0 < value < 100:
+            warn(f"set threshold: {value=} ({type(value)=} is not a int, use {self.default_threshold=}")
+            self.__threshold = self.default_threshold
+            return
+        self.__threshold = value            
+
+    @property
+    def storages(self) -> list[GasTank]:
+        try: raw = self.__storages
+        except AttributeError: return self.get_local_storages()
+        
+        if not isinstance(raw, list): return self.get_local_storages()
+        return [s for s in raw if isinstance(s, GasTank)]
+
+    @storages.setter
+    def storages(self, value: list[GasTank] | GasTank | str | None):
+        if isinstance(value, str): value = get_component(value)
+        if isinstance(value, GasTank): 
+            self.__storages = [ value ]
+            return
+        if isinstance(value, list): 
+            self.__storages = [ s for s in value if isinstance(s, GasTank) ]
+            return
+        warn(f"set storages: given {value=} ({type(value)=} has incorrect type, use default.")
+        self.__storages = self.get_local_storages()