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.98, 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.98. 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 notify_timeout = 100 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 self.__notify_timeout = 0 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.__notify_timeout: self.__notify_timeout -= 1 return 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) return if self.turbine.power_output() == 108 and get_component("power_control").grid(self.turbine.id).net < self.notify_threshold * -1: notify(f"{self.turbine.id}: insufficient energy supply", "error", 30) return if self.steam_in.level() < self.notify_threshold: notify(f"{self.turbine.id}: steam level under {self.notify_threshold}", "warn", 30) return self.notify_timeout = 100 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()