281 lines
12 KiB
Python
281 lines
12 KiB
Python
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()
|