from math import floor, ceil
import numpy as np
from wolfhece.wolf_array import header_wolf
from wolfgpu.utils import count_infiltration_cells
from wolfgpu.injector import SimulationInjector
from wolfgpu.sampled_timer import TimeSampler
from wolfgpu.simplesimu.infiltrations import InfiltrationChronology
[docs]
class InjectorWrapper:
"""
An injector wrapper allows to add information on top of a user defined
injector. It's useful because while being executed by the simulator, it's
convenient to maintain some state around an injector. It also allows to
leave the user's injector code free of information he's not concerned about.
"""
def __init__(self, injector: SimulationInjector,
sim: "wolfgpu.simple_simulation.SimpleSimulation",
active_zone: header_wolf,
identifier:str,
time_to_first_injection: "wolfgpu.simple_simulation.SimulationDuration"):
"""
:param name: A name used to reference the injector. Must be unique in a simulation.
:param active_zone: The geographic area the injector will use
"""
from wolfgpu.simple_simulation import SimpleSimulation, SimulationDuration
assert isinstance(active_zone, header_wolf), f"Expected wolf_header, got {active_zone}"
assert isinstance(sim, SimpleSimulation)
assert isinstance(time_to_first_injection, SimulationDuration)
[docs]
self._injector = injector
# Figure out the time to first injection
[docs]
self._injector_time_sampler = TimeSampler.make_one_shot_timer(
time_to_first_injection,
name="DataInjection")
[docs]
self._identifier = identifier
[docs]
self._time_to_first_injection = time_to_first_injection
[docs]
self._active_zone: header_wolf = active_zone
[docs]
self._infiltration_zones_injection_zone = None
[docs]
self._non_updated_infiltration_zones_count_active_zone = None
[docs]
self._sim: SimpleSimulation = sim
[docs]
def set_infiltration_chronology(self, infiltration_chronolgy: InfiltrationChronology):
# Compute some data over the infiltrations over the active zone of this
# injector.
self._infiltration_zones_injection_zone = self._sim.infiltration_zones[self.active_zone_as_slices()]
nb_zones = infiltration_chronolgy.nb_zones
self._non_updated_infiltration_zones_count_active_zone = count_infiltration_cells(
self._infiltration_zones_injection_zone, nb_zones)
[docs]
def active_zone_as_slices(self) -> tuple[slice, slice]:
"""
Active zone of the injector is defined from (0,0), lambert coordinates;
its origin is measured in that frame of reference. So is the simulation
domain.
"""
inter = self._sim.header_wolf().find_intersection(self._active_zone, ij=True)
if inter is None:
return (slice(0,0), slice(0,0))
elif len(inter) == 2:
inter_sim, inter_inj = inter
return (slice(int(inter_sim[0][0]), int(inter_sim[0][1])),
slice(int(inter_sim[1][0]), int(inter_sim[1][1])))
else:
raise ValueError("Unexpected intersection result")
@property
[docs]
def injector(self):
return self._injector
@property
[docs]
def time_sampler(self):
return self._injector_time_sampler
@property
[docs]
def identifier(self):
return self._identifier
@property
[docs]
def active_zone(self) -> header_wolf:
return self._active_zone
@property
[docs]
def time_to_first_injection(self):
return self._time_to_first_injection