bsm_base ¶
This represents a base model of a BSM model.
It models basic input and output data, and deals with model-independent parameters.
BSMBase ¶
BSMBase(data_in=None, timestep=None, endtime=None, evaltime=None, data_out=None)
Creates a BSMBase object. It is a skeleton class and resembles basic Input and output data of the BSM model. It is not meant to be used directly.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
data_in | ndarray(n, 22) | str(optional) | Influent data. Has to be a 2D array. [SI, SS, XI, XS, XBH, XBA, XP, SO, SNO, SNH, SND, XND, SALK, TSS, Q, TEMP, SD1, SD2, SD3, XD4, XD5] | None |
timestep | float(optional) | Timestep for the simulation [d]. | None |
endtime | float(optional) | Endtime for the simulation [d]. | None |
evaltime | int | ndarray(2) | list[float] | tuple[float](optional) | Evaluation time for the simulation [d]. [starttime, self.simtime[-1]] | None |
data_out | str(optional) | Path to the output data file. | None |
Source code in src/bsm2_python/bsm_base.py
def __init__(
self,
data_in: np.ndarray | str | None = None,
timestep: float | None = None,
endtime: float | None = None,
evaltime: int | np.ndarray | list[float] | tuple[float] | None = None,
data_out: str | None = None,
):
if data_in is None:
# dyninfluent from BSM2:
with open(path_name + '/data/dyninfluent_bsm2.csv', encoding='utf-8-sig') as f:
self.data_in = np.array(list(csv.reader(f, delimiter=','))).astype(float)
elif isinstance(data_in, str):
with open(data_in, encoding='utf-8-sig') as f:
self.data_in = np.array(list(csv.reader(f, delimiter=','))).astype(float)
elif isinstance(data_in, np.ndarray):
self.data_in = data_in.astype(float)
else:
raise ValueError('data_in must be a numpy array, a string or None.')
if isinstance(timestep, int | float):
self.simtime = np.arange(0, self.data_in[-1, 0], timestep, dtype=float)
self.timesteps = timestep * np.ones(len(self.simtime) - 1)
elif timestep is None:
# calculate difference between each time step in data_in
self.simtime = self.data_in[:, 0].flatten()
self.timesteps = np.diff(self.data_in[:, 0], append=(2 * self.data_in[-1, 0] - self.data_in[-2, 0]))
else:
err = 'Timestep must be a number or None.\n \
Please provide a valid timestep.\n'
raise ValueError(err)
if endtime is None:
self.endtime = self.data_in[-1, 0]
elif isinstance(endtime, int | float):
if endtime > self.data_in[-1, 0]:
err = 'Endtime is larger than the last time step in data_in.\n \
Please provide a valid endtime.\n Endtime should be given in days.'
raise ValueError(err)
self.endtime = endtime
self.simtime = self.simtime[self.simtime <= endtime].flatten()
else:
err = 'Endtime must be a number or None.\n \
Please provide a valid endtime.\n'
raise ValueError(err)
self.data_time = self.data_in[:, 0]
if isinstance(evaltime, np.ndarray | list | tuple):
# allow a 1D numpy array or a list/tuple with two float values
dim_eval = 2
if len(evaltime) != dim_eval:
raise ValueError('evaltime must be a 1D numpy array or list with two values.')
try:
start, end = map(float, evaltime)
except Exception as excpt:
raise ValueError('evaltime values must be numbers.') from excpt
if start < self.simtime[0] or end > self.simtime[-1]:
raise ValueError('evaltime must be within the simulation time.')
self.evaltime = np.array([start, end])
elif isinstance(evaltime, int):
# evaluate the last evaltime days of the simulation
if evaltime > self.simtime[-1]:
err = 'evaltime is larger than the simulation time.\n \
The evaluation time must be within the simulation time.'
raise ValueError(err)
if evaltime < 0:
err = 'evaltime must be a positive number.\n \
The evaluation time must be within the simulation time.'
raise ValueError(err)
starttime = self.simtime[np.argmin(np.abs(self.simtime - (self.simtime[-1] - evaltime)))]
self.evaltime = np.array([starttime, self.simtime[-1]])
self.evaltime = np.array([evaltime, self.simtime[-1]])
elif evaltime is None:
# evaluate the last 5 days of the simulation
last_days = 5
starttime = self.simtime[np.argmin(np.abs(self.simtime - (self.simtime[-1] - last_days)))]
self.evaltime = np.array([starttime, self.simtime[-1]])
else:
err = 'evaltime must be a number, a 1D numpy array with two values or None.\n \
Please provide a valid evaltime.\n'
raise ValueError(err)
self.eval_idx = np.array(
[np.where(self.simtime <= self.evaltime[0])[0][-1], np.where(self.simtime <= self.evaltime[1])[0][-1]]
)
self.stabilized = False
self.evaluator = Evaluation(data_out)
step ¶
step(*args, **kwargs)
Abstract method to perform a simulation step.
This method must be implemented by all child classes. It defines how the model advances by one timestep.
Raises:
| Type | Description |
|---|---|
NotImplementedError | If the child class does not implement this method. |
Source code in src/bsm2_python/bsm_base.py
def step(self, *args, **kwargs):
"""Abstract method to perform a simulation step.
This method must be implemented by all child classes.
It defines how the model advances by one timestep.
Raises
------
NotImplementedError
If the child class does not implement this method.
"""
raise NotImplementedError('Child classes must implement the step() method.')
simulate ¶
simulate(*args, **kwargs)
Abstract method to perform a simulation.
This method must be implemented by all child classes. It offers a convenient way to run the simulation for the entire time period. It should call the step() method for each timestep.
Raises:
| Type | Description |
|---|---|
NotImplementedError | If the child class does not implement this method. |
Source code in src/bsm2_python/bsm_base.py
def simulate(self, *args, **kwargs):
"""Abstract method to perform a simulation.
This method must be implemented by all child classes.
It offers a convenient way to run the simulation for the entire time period.
It should call the step() method for each timestep.
Raises
------
NotImplementedError
If the child class does not implement this method.
"""
raise NotImplementedError('Child classes must implement the simulate() method.')