agentlib.modules.utils package

Package containing utility modules.

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)

• log_level (Optional[str])

• module_id (str)

• overwrite_log (bool)

• shared_variable_fields (List[str])

• t_sample (float | int)

• type (Union[str, Dict[str, str]])

• validate_incoming_values (Optional[bool])

• values_only (bool)

Validators:

• check_existence_of_file » filename

• check_valid_fields » shared_variable_fields

• check_valid_level » log_level

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 log_level: str | None = None

The log level for this Module. Default uses the root-loggers level.Options: DEBUG; INFO; WARNING; ERROR; CRITICAL

Validated by:

• check_valid_level

field module_id: str [Required]

The unqiue id of the module within an agent, used only to communicate withing the agent.

field overwrite_log: bool = False

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

field shared_variable_fields: List[str] = []

A list of strings with each string being a field of the Modules configs. The field must be or contain an AgentVariable. If the field is added to this list, all shared attributes of the AgentVariables will be set to True.

Validated by:

• check_valid_fields

field t_sample: float | int = 300

The log is saved every other t_sample seconds.

field type: str | Dict[str, str] [Required]

The type of the Module. Used to find the Python-Object from all agentlib-core and plugin Module options. If a dict is given,it must contain the keys ‘file’ and ‘class_name’. ‘file’ is the filepath of a python file containing the Module.’class_name’ is the name of the Module class within this file.

field validate_incoming_values: bool | None = True

If true, the validator of the AgentVariable value is called when receiving a new value from the DataBroker.

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)

• log_level (Optional[str])

• module_id (str)

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

• shared_variable_fields (List[str])

• t_sample (float | int)

• type (Union[str, Dict[str, str]])

• validate_incoming_values (Optional[bool])

Validators:

• check_valid_fields » shared_variable_fields

• check_valid_level » log_level

field filename: Path [Required]

The filepath to the data.

Constraints:

• path_type = file

field log_level: str | None = None

The log level for this Module. Default uses the root-loggers level.Options: DEBUG; INFO; WARNING; ERROR; CRITICAL

Validated by:

• check_valid_level

field module_id: str [Required]

The unqiue id of the module within an agent, used only to communicate withing the agent.

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

field type: str | Dict[str, str] [Required]

The type of the Module. Used to find the Python-Object from all agentlib-core and plugin Module options. If a dict is given,it must contain the keys ‘file’ and ‘class_name’. ‘file’ is the filepath of a python file containing the Module.’class_name’ is the name of the Module class within this file.

field validate_incoming_values: bool | None = True

If true, the validator of the AgentVariable value is called when receiving a new value from the DataBroker.

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: