agentlib.modules.utils package

Submodules

agentlib.modules.utils.agent_logger module

This module contains a custom Module to log all variables inside an agent’s data_broker.

class agentlib.modules.utils.agent_logger.AgentLogger(*, config: dict, agent: Agent)

Bases: BaseModule

A custom logger for Agents to write variables which are updated in data_broker into a file.

cleanup_results()

Deletes the log if wanted.

property filename

Return the filename where to log.

get_results() → DataFrame

Load the own filename

classmethod load_from_file(filename: str, values_only: bool = True, merge_sources: bool = True) → DataFrame

Loads the log file and consolidates it as a pandas DataFrame.

Parameters:

• filename – The file to load

• values_only – If true, loads a file that only has values saved (default True)

• merge_sources – When there are variables with the same alias from multiple sources, they are saved in different columns. For backwards compatibility, they are merged into a single column. However, if you specify False for this parameter, you can view them separately, resulting in a multi-indexed return column index

process()

Calls the logger every other t_sample is used.

register_callbacks()

Callbacks trigger the log_cache function

terminate()

Terminate all relevant processes of the module. This is necessary to correctly terminate an agent at runtime. Not all modules may need this, hence it is not an abstract method.

pydantic model agentlib.modules.utils.agent_logger.AgentLoggerConfig

Bases: BaseModuleConfig

Define parameters for the AgentLogger

Config:

• arbitrary_types_allowed: bool = True

• validate_assignment: bool = True

• extra: str = forbid

• frozen: bool = True

Fields:

• clean_up (bool)

• filename (str)

• overwrite_log (bool)

• t_sample (float | int)

• values_only (bool)

Validators:

• check_existence_of_file » filename

field clean_up: bool = True

If True, file is deleted once load_log is called.

field filename: str [Required]

The filename where the log is stored.

Validated by:

• check_existence_of_file

field overwrite_log: bool = False

If true, old logs are auto deleted when a new log should be written with that name.

field t_sample: float | int = 300

The log is saved every other t_sample seconds.

field values_only: bool = True

If True, only the values are logged. Else, allfields in the AgentVariable are logged.

validator check_existence_of_file  »  filename

Checks whether the file already exists.

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialise private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self – The BaseModel instance.

• context – The context.

agentlib.modules.utils.try_sensor module

This module contains a SensorModule which reads a .dat file from the Deutsche Wetterdienst (DWD).

class agentlib.modules.utils.try_sensor.TRYSensor(*, config: dict, agent: Agent)

Bases: BaseModule

A module that emulates a sensor using the TRY data of the DWD. The module casts the following AgentVariables into the data_broker:

• T_oda: Air temperature 2m over ground [K]

• pressure: Air pressure in standard height [hPa]

• wind_direction: Wind direction 10 m above gorund [Grad] {0..360;999}

• wind_speed: Wind speed 10 m above ground [m/s]

• coverage: [eighth] {0..8;9}

• absolute_humidity: [g/kg]

• relative_humidity: Relative humidity 2 m above ground [%] {1..100}

• beam_direct: Direct beam of sun (hor. plane) [W/m^2] downwards: positive

• beam_diffuse: Diffuse beam of sun (hor. plane) [W/m^2] downwards: positive

• beam_atm: Beam of atmospheric heat (hor. plane) [W/m^2] downwards: positive

• beam_terr: Beam of terrestrial heat [W/m^2] upwards: negative

property filename

Return the filename.

get_data_now()

Get the data at the current env time. Reiterate the year if necessary.

process()

Write the current TRY values into data_broker every other t_sample

register_callbacks()

Dont do anything as this BaseModule is not event-triggered

property t_sample

Return the sample rate.

pydantic model agentlib.modules.utils.try_sensor.TRYSensorConfig

Bases: BaseModuleConfig

Define parameters for the TRYSensor

Config:

• arbitrary_types_allowed: bool = True

• validate_assignment: bool = True

• extra: str = forbid

• frozen: bool = True

Fields:

• filename (pathlib.Path)

• outputs (List[agentlib.core.datamodels.AgentVariable])

• shared_variable_fields (List[str])

• t_sample (float | int)

Validators:

• check_valid_fields » shared_variable_fields

field filename: Path [Required]

The filepath to the data.

Constraints:

• path_type = file

field outputs: List[AgentVariable] = [AgentVariable(name='T_oda', type=None, timestamp=None, unit='K', description='Air temperature 2m over ground [K]', ub=inf, lb=-inf, clip=False, allowed_values=[], value=None, alias='T_oda', source=Source(agent_id=None, module_id=None), shared=None, rdf_class=None), AgentVariable(name='pressure', type=None, timestamp=None, unit='hPa', description='Air pressure in standard height [hPa]', ub=inf, lb=-inf, clip=False, allowed_values=[], value=None, alias='pressure', source=Source(agent_id=None, module_id=None), shared=None, rdf_class=None), AgentVariable(name='wind_direction', type=None, timestamp=None, unit='°', description='Wind direction 10 m above gorund [Grad] {0..360;999}', ub=inf, lb=-inf, clip=False, allowed_values=[], value=None, alias='wind_direction', source=Source(agent_id=None, module_id=None), shared=None, rdf_class=None), AgentVariable(name='wind_speed', type=None, timestamp=None, unit='m/s', description='Wind speed 10 m above ground [m/s]', ub=inf, lb=-inf, clip=False, allowed_values=[], value=None, alias='wind_speed', source=Source(agent_id=None, module_id=None), shared=None, rdf_class=None), AgentVariable(name='coverage', type=None, timestamp=None, unit='eighth', description='[eighth]  {0..8;9}', ub=inf, lb=-inf, clip=False, allowed_values=[], value=None, alias='coverage', source=Source(agent_id=None, module_id=None), shared=None, rdf_class=None), AgentVariable(name='absolute_humidity', type=None, timestamp=None, unit='g/kg', description='[g/kg]', ub=inf, lb=-inf, clip=False, allowed_values=[], value=None, alias='absolute_humidity', source=Source(agent_id=None, module_id=None), shared=None, rdf_class=None), AgentVariable(name='relative_humidity', type=None, timestamp=None, unit='%', description='Relative humidity 2 m above ground [%] {1..100}', ub=inf, lb=-inf, clip=False, allowed_values=[], value=None, alias='relative_humidity', source=Source(agent_id=None, module_id=None), shared=None, rdf_class=None), AgentVariable(name='beam_direct', type=None, timestamp=None, unit='W/m^2', description='Direct beam of sun (hor. plane) [W/m^2] downwards: positive', ub=inf, lb=-inf, clip=False, allowed_values=[], value=None, alias='beam_direct', source=Source(agent_id=None, module_id=None), shared=None, rdf_class=None), AgentVariable(name='beam_diffuse', type=None, timestamp=None, unit='/m^2', description='Diffuse beam of sun (hor. plane) [W/m^2] downwards: positive', ub=inf, lb=-inf, clip=False, allowed_values=[], value=None, alias='beam_diffuse', source=Source(agent_id=None, module_id=None), shared=None, rdf_class=None), AgentVariable(name='beam_atm', type=None, timestamp=None, unit='/m^2', description='Beam of atmospheric heat (hor. plane) [W/m^2] downwards: positive', ub=inf, lb=-inf, clip=False, allowed_values=[], value=None, alias='beam_atm', source=Source(agent_id=None, module_id=None), shared=None, rdf_class=None), AgentVariable(name='beam_terr', type=None, timestamp=None, unit='/m^2', description='Beam of terrestrial heat [W/m^2] upwards: negative', ub=inf, lb=-inf, clip=False, allowed_values=[], value=None, alias='beam_terr', source=Source(agent_id=None, module_id=None), shared=None, rdf_class=None)]

field shared_variable_fields: List[str] = ['outputs']

Validated by:

• check_valid_fields

field t_sample: float | int = 1

Sample time of sensor. As TRY are hourly, default is 1 h

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialise private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self – The BaseModel instance.

• context – The context.

agentlib.modules.utils.try_sensor.read_dat_file(dat_file)

Read a .dat file from the DWD’s TRY data from the given .dat-file

Parameters:

dat_file (str) – The file to read

Returns:

Module contents

Package containing utility modules.