From e90b4d2097bfe439714d347c95419f2ec1feba6f Mon Sep 17 00:00:00 2001 From: anima Date: Fri, 13 Nov 2026 01:00:02 +0100 Subject: [PATCH] steamlined power wrapper --- libs/wrapper/power.py | 280 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 280 insertions(+) create mode 100644 libs/wrapper/power.py diff --git a/libs/wrapper/power.py b/libs/wrapper/power.py new file mode 100644 index 0000000..ebdb04b --- /dev/null +++ b/libs/wrapper/power.py @@ -0,0 +1,280 @@ +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, outpost: 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.outpost = outpost + 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(): + self.cap.set_throttle(0) + self.steam_out.disconnect() + else: + 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": + sleep(1) + self.steam_out.disconnect() + + def check_storage(self) -> None: + """search a `GasTank` to fill""" + if self.cap.pressure() < 10: 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 + + @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: + if "__threshold" not in self.__dict__: return self.default_threshold + if not isinstance(self.__threshold, (float, int)): return self.default_threshold + if not 0 < self.__threshold < 1: return self.default_threshold + return self.__threshold + + @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]: + if "__outpost" not in self.__dict__: return None + if isinstance(self.__outposts, list): return [ o for o in self.__outposts if isinstance(o, OutpostComponent) ] + # TODO: default + return None + + @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): + # TODO: find deafult. powernet ? or nearest ? + warn(f"no outpost for steam out found ({value=}") + self.__outposts = list() + 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(): + notify(f"{self.turbine.id}: stopped working or has low efficency ({self.turbine.efficiency()*100}%)", "error", 10) + sleep(10) + + 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: + if not "__threshold" in self.__dict__: return self.default_threshold + if not isinstance(self.__threshold, int): return self.default_threshold + if not 0 < self.__threshold < 100: return self.default_threshold + return self.threshold + + @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]: + if not "__storages" in self.__dict__: return self.get_local_storages() + if isinstance(self.__storages, list): return [ s for s in self.__storages if isinstance(s, GasTank) ] + return self.get_local_storages() + + @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()