Welcome to the DT4Safeguards documentation!¶

Indices and tables¶

Index

Introduction¶

DT4Safeguards (Digital Twin for Safeguards in Nuclear Waste Management) is a python framework modeling safeguards activities in nuclear facilities related to the process of the disposal of nuclear waste.

The project can be found in the following repository: https://jugit.fz-juelich.de/m.kreutle/DT4Safeguards

Requirements¶

Make sure you have all required packages installed:

pip install .

For the full development setup including docs and test dependencies run:

pip install .[docs,test]

Results¶

The results were visualized with Dash and can be displayed locally. Run the Dash app via:

python3 project/app.py

Results are then visible with a browser at http://127.0.0.1:8050/.

Classes and methods¶

class project.model.database.DatabaseSetup¶

A class to set up a PostgreSQL database. It provides methods to establish connections, create tables and load data from a json file.

get_connection()¶: Establishes a connection to a PostgreSQL database.

get_cursor()¶: Creates a cursor object for executing database queries.

create_database()¶: Creates a PostgreSQL database.

create_tables()¶: Creates necessary tables in the database.

populate_database()¶: Populates the database with facility, room, material, building element, holding area, container, nuclear material, and entry/exit data from a dictionary.

load_database_from_json()¶: Creates a PostgreSQL database and loads data from a JSON file into it.

classmethod create_database(db_name: str, db_user: str, db_password: str, db_host: str, db_port: str) → None¶

Creates a PostgreSQL database.

Parameters:

db_name (str) – The name of the database to create.
db_user (str) – The username for authentication.
db_password (str) – The password for authentication.
db_host (str) – The hostname or IP address of the database server.
db_port (str) – The port number on which the database server is listening.

Returns:

None

classmethod create_tables(cursor: cursor) → None¶

Creates tables in a PostgreSQL database for a given connection.

Parameters:: cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
Returns:: None

classmethod delete_database(db_name: str, db_user: str, db_password: str, db_host: str, db_port: str) → None¶

Deletes a PostgreSQL database.

Parameters:

db_name (str) – The name of the database to delete.
db_user (str) – The username for authentication.
db_password (str) – The password for authentication.
db_host (str) – The hostname or IP address of the database server.
db_port (str) – The port number on which the database server is listening.

Returns:

None

classmethod get_connection(db_name: str, db_user: str, db_password: str, db_host: str, db_port: str) → connection¶

Establishes a connection to a PostgreSQL database.

Parameters:

db_name (str) – The name of the database to connect to.
db_user (str) – The username for authentication.
db_password (str) – The password for authentication.
db_host (str) – The hostname or IP address of the database server.
db_port (str) – The port number on which the database server is listening.

Returns:

Connection object to database.

Return type:

connection (psycopg2.connection)

classmethod get_cursor(conn: connection) → cursor¶

Creates a cursor object for a connection to a PostgreSQL database.

Parameters:: connection (psycopg2.connection) – Connection object to database.
Returns:: The database cursor for executing SQL queries.
Return type:: cursor (psycopg2.cursor)

classmethod load_database_from_json(db_name: str, db_user: str, db_password: str, db_host: str, db_port: str, json_file_path: str, get_cursor: bool | None = False) → cursor | None¶

Creates a PostgreSQL database and loads data from a json file into it.

Parameters:

db_name (str) – The name of the database to connect to.
db_user (str) – The username for authentication.
db_password (str) – The password for authentication.
db_host (str) – The hostname or IP address of the database server.
db_port (str) – The port number on which the database server is listening.
json_file_path (str) – The json file path containing data to be loaded into database.
get_cursor (bool, optional) – Boolean to indicate if courses is to be returned (True) or not (False). (default is False)

Returns:

The database cursor for executing SQL queries.

Return type:

None | psycopg2.cursor

classmethod populate_database(data: Dict[str, Dict[str, Any]], cursor: cursor) → None¶

Populates the database with facility, room, material, building element, holding area, container type, container, nuclear material, and entry/exit data.

Parameters:

data (Dict[str, Dict[str, Any]]) – The data to insert into the database.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.

Returns:

None

class project.model.database.DatabaseWrite¶

A class to interact with a PostgreSQL database. It provides methods to insert data into the database.

insert_facility_type()¶: Inserts a record into the ‘facility_type’ table.

insert_facility()¶: Inserts a record into the ‘facility’ table.

insert_room()¶: Inserts a record into the ‘room’ table.

insert_material()¶: Inserts a record into the ‘material’ table.

insert_building_element()¶: Inserts a record into the ‘building_element’ table.

insert_holding_area()¶: Inserts a record into the ‘holding_area’ table.

insert_container_type()¶: Inserts a record into the ‘container_type’ table.

insert_container()¶: Inserts a record into the ‘container’ table.

insert_nuclear_material()¶: Inserts a record into the ‘nuclear_material’ table.

insert_operator_equipment()¶: Inserts a record into the ‘operator_equipment’ table.

insert_transport_command()¶: Inserts a record into the ‘transport_command’ table.

insert_transport_equipment_inventory()¶: Inserts a record into the ‘transport_equipment_inventory’ table.

insert_transport_node()¶: Inserts a record into the ‘transport_node’ table.

insert_transport_edge()¶: Inserts a record into the ‘transport_edge’ table.

insert_transport_step()¶: Inserts a record into the ‘transport_step’ table.

insert_building_command()¶: Inserts a record into the ‘building_command’ table.

insert_building_equipment_inventory()¶: Inserts a record into the ‘building_equipment_inventory’ table.

insert_processing_command()¶: Inserts a record into the ‘processing_command’ table.

insert_processing_equipment_inventory()¶: Inserts a record into the ‘processing_equipment_inventory’ table.

insert_safeguards_equipment()¶: Inserts a record into the ‘safeguards_equipment’ table.

insert_inspection_command()¶: Inserts a record into the ‘inspection_command’ table.

insert_inspection_equipment_inventory()¶: Inserts a record into the ‘inspection_equipment_inventory’ table.

insert_facility_history()¶: Inserts a record into the ‘facility_history’ table.

insert_room_history()¶: Inserts a record into the ‘room_history’ table.

insert_building_element_history()¶: Inserts a record into the ‘building_element_history’ table.

insert_holding_area_history()¶: Inserts a record into the ‘holding_area_history’ table.

insert_container_history()¶: Inserts a record into the ‘container_history’ table.

insert_nuclear_material_history()¶: Inserts a record into the ‘nuclear_material_history’ table.

insert_operator_equipment_history()¶: Inserts a record into the ‘operator_equipment_history’ table.

insert_safeguards_equipment_history()¶: Inserts a record into the ‘safeguards_equipment_history’ table.

update_record()¶: Updates a record in the specified table.

_create_update_query()¶: Generates an SQL query and values for updating a specified table.

shadow_check()¶: Checks if given record comes with a shadow record, that represents its declared state.

insert_shadow()¶: Inserts a shadow row for a specified record in a specified table.

classmethod insert_building_command(target_facility_id: int, target_room_id: int, target_building_element_id: int, target_holding_area_id: int, target_transport_edge_id: int, building_command_data: Dict[str, Any], cursor: cursor, return_id: bool | None = False) → int | None¶

Inserts data into ‘building_command’ table.

Parameters:

target_facility_id (int) – The ID of the target facility.
target_room_id (int) – The ID of the target room.
target_building_element_id (int) – The ID of the target building element.
target_holding_area_id (int) – The ID of the target holding area.
target_transport_edge_id (int) – The ID of the target transport edge.
building_command_data (Dict[str, Any]) – The data for the building command.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
return_id (bool, optional) – Boolean to indicate if ID of just added record is to be returned (True) or not (False). (default is False)

Returns:

ID of just added record, if return_id is True.

Return type:

None | int

classmethod insert_building_element(parent_room_id: int, second_parent_room_id: int | None, building_element_data: Dict[str, Any], cursor: cursor, return_id: bool | None = False) → int | None¶

Inserts a building element record into the ‘building_element’ table.

Parameters:

parent_room_id (int) – ID of the parent room.
second_parent_room_id (Optional[int]) – ID of the second parent room (if applicable).
building_element_data (Dict[str, Any]) – A dictionary containing building element details.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
return_id (bool, optional) – Boolean to indicate if ID of just added record is to be returned (True) or not (False). (default is False)

Returns:

ID of just added record, if return_id is True.

Return type:

None | int

classmethod insert_building_element_history(building_element_id: int, end_time: str, cursor: cursor, command_declaration_status: bool, building_command_id: int | None = None, inspection_command_id: int | None = None) → None¶

Inserts data into the ‘building_element_history’ table.

Parameters:

building_element_id (int) – ID of the building element.
end_time (str) – End time of the command for the building element history.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
command_declaration_status (bool) – Declaration status of command. False is undeclared.
building_command_id (int, optional) – ID of the building command. Exactly one of the command IDs has to be not None. (default is None)
inspection_command_id (int, optional) – ID of the inspection command. Exactly one of the command IDs has to be not None. (default is None)

Returns:

None

classmethod insert_building_equipment_inventory(operator_equipment_id: int, building_command_id: int, cursor: cursor) → None¶

Inserts data into ‘building_equipment_inventory’ table.

Parameters:

operator_equipment_id (int) – The ID of the operator equipment.
building_command_id (int) – The ID of the building command.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.

Returns:

None

classmethod insert_container(parent_holding_area_id: int, container_data: Dict[str, Any], cursor: cursor, return_id: bool | None = False) → int | None¶

Inserts a container record into the ‘container’ table.

Parameters:

parent_holding_area_id (int) – ID of the parent holding area.
container_data (Dict[str, Any]) – A dictionary containing container details.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
return_id (bool, optional) – Boolean to indicate if ID of just added record is to be returned (True) or not (False). (default is False)

Returns:

ID of just added record, if return_id is True.

Return type:

None | int

classmethod insert_container_history(container_id: int, end_time: str, cursor: cursor, command_declaration_status: bool, transport_command_id: int | None = None, processing_command_id: int | None = None, inspection_command_id: int | None = None) → None¶

Inserts data into the ‘container_history’ table.

Parameters:

container_id (int) – ID of the container.
end_time (str) – End time of the command for the container history.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
command_declaration_status (bool) – Declaration status of command. False is undeclared.
transport_command_id (int, optional) – ID of the transport command. Exactly one of the command IDs has to be not None. (default is None)
processing_command_id (int, optional) – ID of the processing command. Exactly one of the command IDs has to be not None. (default is None)
inspection_command_id (int, optional) – ID of the inspection command. Exactly one of the command IDs has to be not None. (default is None)

Returns:

None

classmethod insert_container_type(container_type_data: Dict[str, Any], cursor: cursor) → None¶

Inserts a container type record into the ‘container_type’ table.

Parameters:

container_type_data (Dict[str, Any]) – A dictionary containing container type details.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.

Returns:

None

classmethod insert_facility(facility_data: Dict[str, Any], cursor: cursor, return_id: bool | None = False) → int | None¶

Inserts data into the ‘facility’ table.

Parameters:

facility_data (Dict[str, Any]) – Dictionary containing facility details.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
return_id (bool, optional) – Boolean to indicate if ID of just added record is to be returned (True) or not (False). (default is False)

Returns:

ID of just added record, if return_id is True.

Return type:

None | int

classmethod insert_facility_history(facility_id: int, end_time: str, cursor: cursor, command_declaration_status: bool, building_command_id: int) → None¶

Inserts data into the ‘facility_history’ table.

Parameters:

facility_id (int) – ID of the facility.
end_time (str) – End time of the command for the facility history.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
command_declaration_status (bool) – Declaration status of command. False is undeclared.
building_command_id (int) – ID of the building command.

Returns:

None

classmethod insert_facility_type(facility_type_data: Dict[str, Any], cursor: cursor) → int | None¶

Inserts data into the ‘facility_type’ table.

Parameters:

facility_type_data (Dict[str, Any]) – Dictionary containing facility type details.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.

Returns:

None

classmethod insert_holding_area(parent_room_id: int, holding_area_data: Dict[str, Any], cursor: cursor, return_id: bool | None = False) → int | None¶

Inserts a holding area record into the ‘holding_area’ table.

Parameters:

parent_room_id (int) – ID of the parent room.
holding_area_data (Dict[str, Any]) – A dictionary containing holding area details.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
return_id (bool, optional) – Boolean to indicate if ID of just added record is to be returned (True) or not (False). (default is False)

Returns:

ID of just added record, if return_id is True.

Return type:

None | int

classmethod insert_holding_area_history(holding_area_id: int, end_time: str, cursor: cursor, command_declaration_status: bool, transport_command_id: int | None = None, building_command_id: int | None = None, inspection_command_id: int | None = None) → None¶

Inserts data into the ‘holding_area_history’ table.

Parameters:

holding_area_id (int) – ID of the holding area.
end_time (str) – End time of the command for the holding area history.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
command_declaration_status (bool) – Declaration status of command. False is undeclared.
transport_command_id (int, optional) – ID of the transport command. Exactly one of the command IDs has to be not None. (default is None)
building_command_id (int, optional) – ID of the building command. Exactly one of the command IDs has to be not None. (default is None)
inspection_command_id (int, optional) – ID of the inspection command. Exactly one of the command IDs has to be not None. (default is None)

Returns:

None

classmethod insert_inspection_command(target_room_id: int, target_building_element_id: int, target_holding_area_id: int, target_container_id: int, target_nuclear_material_id: int, target_operator_equipment_id: int, inspection_command_data: Dict[str, Any], cursor: cursor, return_id: bool | None = False) → int | None¶

Inserts data into ‘inspection_command’ table.

Parameters:

target_room_id (int) – The ID of the target room.
target_building_element_id (int) – The ID of the target building element.
target_holding_area_id (int) – The ID of the target holding area.
target_container_id (int) – The ID of the target container.
target_nuclear_material_id (int) – The ID of the target nuclear material.
target_operator_equipment_id (int) – The ID of the operator equipment.
inspection_command_data (Dict[str, Any]) – The data for the inspection command.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
return_id (bool, optional) – Boolean to indicate if ID of just added record is to be returned (True) or not (False). (default is False)

Returns:

ID of just added record, if return_id is True.

Return type:

None | int

classmethod insert_inspection_equipment_inventory(safeguards_equipment_id: int, inspection_command_id: int, cursor: cursor) → None¶

Inserts data into ‘inspection_equipment_inventory’ table.

Parameters:

safeguards_equipment_id (int) – The ID of the safeguards equipment.
inspection_command_id (int) – The ID of the inspection command.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.

Returns:

None

classmethod insert_material(material_data: Dict[str, Any], cursor: cursor) → None¶

Inserts material data into the ‘material’ table.

Parameters:

material_data (Dict[str, Any]) – A dictionary containing material details.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.

Returns:

None

classmethod insert_nuclear_material(parent_container_id: int, parent_holding_area_id: int, nuclear_material_data: Dict[str, Any], cursor: cursor, return_id: bool | None = False) → int | None¶

Inserts a nuclear material record into the ‘nuclear_material’ table.

Parameters:

parent_container_id (int) – ID of the parent container.
parent_holding_area_id (int) – ID of the parent holding area.
nuclear_material_data (Dict[str, Any]) – A dictionary containing nuclear material details.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
return_id (bool, optional) – Boolean to indicate if ID of just added record is to be returned (True) or not (False). (default is False)

Returns:

ID of just added record, if return_id is True.

Return type:

None | int

classmethod insert_nuclear_material_history(nuclear_material_id: int, end_time: str, cursor: cursor, command_declaration_status: bool, transport_command_id: int | None = None, processing_command_id: int | None = None, inspection_command_id: int | None = None) → None¶

Inserts data into the ‘nuclear_material_history’ table.

Parameters:

nuclear_material_id (int) – ID of the nuclear material.
end_time (str) – End time of the command for the nuclear material history.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
command_declaration_status (bool) – Declaration status of command. False is undeclared.
transport_command_id (int, optional) – ID of the transport command. Exactly one of the command IDs has to be not None. (default is None)
processing_command_id (int, optional) – ID of the processing command. Exactly one of the command IDs has to be not None. (default is None)
inspection_command_id (int, optional) – ID of the inspection command. Exactly one of the command IDs has to be not None. (default is None)

Returns:

None

classmethod insert_operator_equipment(parent_room_id: int, parent_holding_area_id: int, operator_equipment_data: Dict[str, Any], cursor: cursor, return_id: bool | None = False) → int | None¶

Inserts an operator equipment record into the ‘operator_equipment’ table.

Parameters:

parent_room_id (int) – ID of the parent room.
parent_holding_area_id (int) – ID of the parent holding area.
operator_equipment_data (Dict[str, Any]) – A dictionary containing operator equipment details.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
return_id (bool, optional) – Boolean to indicate if ID of just added record is to be returned (True) or not (False). (default is False)

Returns:

ID of just added record, if return_id is True.

Return type:

None | int

classmethod insert_operator_equipment_history(operator_equipment_id: int, end_time: str, cursor: cursor, command_declaration_status: bool, transport_command_id: int | None = None, building_command_id: int | None = None, processing_command_id: int | None = None, inspection_command_id: int | None = None) → None¶

Inserts data into the ‘operator_equipment_history’ table.

Parameters:

operator_equipment_id (int) – ID of the operator equipment.
end_time (str) – End time of the command for the operator equipment history.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
command_declaration_status (bool) – Declaration status of command. False is undeclared.
transport_command_id (int, optional) – ID of the transport command. Exactly one of the command IDs has to be not None. (default is None)
building_command_id (int, optional) – ID of the building command. Exactly one of the command IDs has to be not None. (default is None)
processing_command_id (int, optional) – ID of the processing command. Exactly one of the command IDs has to be not None. (default is None)
inspection_command_id (int, optional) – ID of the inspection command. Exactly one of the command IDs has to be not None. (default is None)

Returns:

None

classmethod insert_processing_command(target_nuclear_material_id: int, target_container_id: int, processing_command_data: Dict[str, Any], cursor: cursor, return_id: bool | None = False) → int | None¶

Inserts data into ‘processing_command’ table.

Parameters:

target_nuclear_material_id (int) – The ID of the target nuclear material.
target_container_id (int) – The ID of the target container.
processing_command_data (Dict[str, Any]) – The data for the processing command.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
return_id (bool, optional) – Boolean to indicate if ID of just added record is to be returned (True) or not (False). (default is False)

Returns:

ID of just added record, if return_id is True.

Return type:

None | int

classmethod insert_processing_equipment_inventory(operator_equipment_id: int, processing_command_id: int, cursor: cursor) → None¶

Inserts data into ‘processing_equipment_inventory’ table.

Parameters:

operator_equipment_id (int) – The ID of the operator equipment.
processing_command_id (int) – The ID of the processing command.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.

Returns:

None

classmethod insert_room(parent_facility_id: int, room_data: Dict[str, Any], cursor: cursor, return_id: bool | None = False) → int | None¶

Inserts a room record into the ‘room’ table.

Parameters:

parent_facility_id (int) – ID of the parent facility.
room_data (Dict[str, Any]) – A dictionary containing room details.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
return_id (bool, optional) – Boolean to indicate if ID of just added record is to be returned (True) or not (False). (default is False)

Returns:

ID of just added record, if return_id is True.

Return type:

None | int

classmethod insert_room_history(room_id: int, end_time: str, cursor: cursor, command_declaration_status: bool, building_command_id: int | None = None, inspection_command_id: int | None = None) → None¶

Inserts data into the ‘room_history’ table.

Parameters:

room_id (int) – ID of the room.
end_time (str) – End time of the command for the room history.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
command_declaration_status (bool) – Declaration status of command. False is undeclared.
building_command_id (int, optional) – ID of the building command. Exactly one of the command IDs has to be not None. (default is None)
inspection_command_id (int, optional) – ID of the inspection command. Exactly one of the command IDs has to be not None. (default is None)

Returns:

None

classmethod insert_safeguards_equipment(parent_room_id: int, safeguards_equipment_data: Dict[str, Any], cursor: cursor, return_id: bool | None = False) → int | None¶

Inserts an safeguards equipment record into the ‘safeguards_equipment’ table.

Parameters:

parent_room_id (int) – ID of the parent room.
safeguards_equipment_data (Dict[str, Any]) – A dictionary containing safeguards equipment details.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
return_id (bool, optional) – Boolean to indicate if ID of just added record is to be returned (True) or not (False). (default is False)

Returns:

ID of just added record, if return_id is True.

Return type:

None | int

insert_safeguards_equipment_history(safeguards_equipment_id: int, end_time: str, cursor: cursor, command_declaration_status: bool, processing_command_id: int | None = None, inspection_command_id: int | None = None) → None¶

Inserts data into the ‘safeguards_equipment_history’ table.

Parameters:

safeguards_equipment_id (int) – ID of the safeguards equipment.
end_time (str) – End time of the command for the safeguards equipment history.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
command_declaration_status (bool) – Declaration status of command. False is undeclared.
processing_command_id (int, optional) – ID of a processing command, that accidentally or willfully altered safeguards equipment. Exactly one of the command IDs has to be not None. (default is None)
inspection_command_id (int, optional) – ID of an inspection command. Exactly one of the command IDs has to be not None. (default is None)

Returns:

None

classmethod insert_shadow(table_name: str, record_id: int, cursor: cursor, return_id: bool | None = False) → int | None¶

Inserts a shadow row for a specified record in a specified table.

Parameters:

table_name (str) – Name of the table to update.
record_id (int) – ID of the record to update.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
return_id (bool, optional) – Boolean to indicate if ID of just added record is to be returned (True) or not (False). (default is False)

Returns:

ID of just added record, if return_id is True.

Return type:

None | int

classmethod insert_transport_command(target_container_id: int, destination_holding_area_id: int, transport_command_data: Dict[str, Any], cursor: cursor, return_id: bool | None = False) → int | None¶

Inserts data into ‘transport_command’ table.

Parameters:

target_container_id (int) – The ID of the target container.
destination_holding_area_id (int) – The ID of the destination holding area.
transport_command_data (Dict[str, Any]) – The data for the transport command.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
return_id (bool, optional) – Boolean to indicate if ID of just added record is to be returned (True) or not (False). (default is False)

Returns:

ID of just added record, if return_id is True.

Return type:

None | int

classmethod insert_transport_edge(source_id: int, target_id: int, linked_building_element_id: int, edge_data: Dict[str, any], cursor: cursor, return_id: bool | None = False) → int | None¶

Inserts data into ‘transport_edge’ table.

Parameters:

source_id (int) – The ID of the source node.
target_id (int) – The ID of the target node.
edge_data (Dict[str, Any]) – The edge data to insert.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
return_id (bool, optional) – Boolean to indicate if ID of just added record is to be returned (True) or not (False). (default is False)

Returns:

ID of just added record, if return_id is True.

Return type:

None | int

classmethod insert_transport_equipment_inventory(operator_equipment_id: int, transport_command_id: int, cursor: cursor) → None¶

Inserts data into ‘transport_equipment_inventory’ table.

Parameters:

operator_equipment_id (int) – The ID of the operator equipment.
transport_command_id (int) – The ID of the transport command.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.

Returns:

None

classmethod insert_transport_node(cursor: cursor, facility_id: int | None = None, room_id: int | None = None, return_id: bool | None = False) → int | None¶

Inserts data into ‘transport_node’ table.

Parameters:

cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
facility_id (int, optional) – The ID of the linked facility. Either facility_id or room_id has be not None. (default is None)
room_id (int, optional) – The ID of the linked room. Either facility_id or room_id has to be not None. (default is None)
return_id (bool, optional) – Boolean to indicate if ID of just added record is to be returned (True) or not (False). (default is False)

Returns:

ID of just added record, if return_id is True.

Return type:

None | int

Raises:

ValueError – If, out of facility_id and room_id, not one is other than None.

classmethod insert_transport_step(transport_edge_id: int, transport_command_id: int, step_data: Dict[str, Any], cursor: cursor) → None¶

Inserts data into ‘transport_step’ table.

Parameters:

transport_edge_id (int) – The ID of an transport_edge record.
transport_command_id (int) – The ID of the transport command.
step_data (Dict[str, Any]) – The step data to insert.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.

Returns:

None

classmethod shadow_check(table_name: str, record_id: int, declaration_status: bool, cursor: cursor) → int¶

Checks if given record comes with a shadow record, that represents its declared state.

Parameters:

table_name (str) – Name of the table to update.
record_id (int) – ID of the record to update.
declaration_status (bool) – Declaration status of update. False is undeclared.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.

Returns:

Original ID or shadow ID, based on declaration_status and existence of shadow record.

Return type:

int

classmethod update_building_element(building_element_id: int, end_time: str, updates: Dict, declaration_status: bool, cursor: cursor, building_command_id: int | None = None, inspection_command_id: int | None = None) → None¶

Updates a record in the ‘building_element’ table after writing to the ‘building_element_history’ table.

Parameters:

building_element_id (int) – ID of the building_element record to update.
end_time (str) – End time of the command for the building_element history.
updates (Dict) – Dictionary with column names and their new values.
declaration_status (bool) – Declaration status of update. False is undeclared.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
building_command_id (int, optional) – ID of the building command. Exactly one of the command IDs has to be not None. (default is None)
inspection_command_id (int, optional) – ID of the inspection command. Exactly one of the command IDs has to be not None. (default is None)

Returns:

None

classmethod update_container(container_id: int, end_time: str, updates: Dict, declaration_status: bool, cursor: cursor, transport_command_id: int | None = None, processing_command_id: int | None = None, inspection_command_id: int | None = None) → None¶

Updates a record in the ‘container’ table after writing to the ‘container_history’ table.

Parameters:

container_id (int) – ID of the container record to update.
end_time (str) – End time of the command for the container history.
updates (Dict) – Dictionary with column names and their new values.
declaration_status (bool) – Declaration status of update. False is undeclared.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
transport_command_id (int, optional) – ID of the transport command. Exactly one of the command IDs has to be not None. (default is None)
processing_command_id (int, optional) – ID of the processing command. Exactly one of the command IDs has to be not None. (default is None)
inspection_command_id (int, optional) – ID of the inspection command. Exactly one of the command IDs has to be not None. (default is None)

Returns:

None

classmethod update_facility(facility_id: int, end_time: str, updates: Dict, declaration_status: bool, cursor: cursor, building_command_id: int) → None¶

Updates a record in the ‘facility’ table after writing to the ‘facility_history’ table.

Parameters:

facility_id (int) – ID of the facility record to update.
end_time (str) – End time of the command for the facility history.
updates (Dict) – Dictionary with column names and their new values.
declaration_status (bool) – Declaration status of update. False is undeclared.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
building_command_id (int) – ID of the building command.

Returns:

None

classmethod update_holding_area(holding_area_id: int, end_time: str, updates: Dict, declaration_status: bool, cursor: cursor, transport_command_id: int | None = None, building_command_id: int | None = None, inspection_command_id: int | None = None) → None¶

Updates a record in the ‘holding_area’ table after writing to the ‘holding_area_history’ table.

Parameters:

holding_area_id (int) – ID of the holding_area record to update.
end_time (str) – End time of the command for the holding_area history.
updates (Dict) – Dictionary with column names and their new values.
declaration_status (bool) – Declaration status of update. False is undeclared.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
transport_command_id (int, optional) – ID of the transport command. Exactly one of the command IDs has to be not None. (default is None)
building_command_id (int, optional) – ID of the building command. Exactly one of the command IDs has to be not None. (default is None)
inspection_command_id (int, optional) – ID of the inspection command. Exactly one of the command IDs has to be not None. (default is None)

Returns:

None

classmethod update_nuclear_material(nuclear_material_id: int, end_time: str, updates: Dict, declaration_status: bool, cursor: cursor, transport_command_id: int | None = None, processing_command_id: int | None = None, inspection_command_id: int | None = None) → None¶

Updates a record in the ‘nuclear_material’ table after writing to the ‘nuclear_material_history’ table.

Parameters:

container_id (int) – ID of the container record to update.
end_time (str) – End time of the command for the nuclear_material history.
updates (Dict) – Dictionary with column names and their new values.
declaration_status (bool) – Declaration status of update. False is undeclared.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
transport_command_id (int, optional) – ID of the transport command. Exactly one of the command IDs has to be not None. (default is None)
processing_command_id (int, optional) – ID of the processing command. Exactly one of the command IDs has to be not None. (default is None)
inspection_command_id (int, optional) – ID of the inspection command. Exactly one of the command IDs has to be not None. (default is None)

Returns:

None

classmethod update_operator_equipment(operator_equipment_id: int, end_time: str, updates: Dict, declaration_status: bool, cursor: cursor, transport_command_id: int | None = None, building_command_id: int | None = None, processing_command_id: int | None = None, inspection_command_id: int | None = None) → None¶

Updates a record in the ‘operator_equipment’ table after writing to the ‘operator_equipment_history’ table.

Parameters:

container_id (int) – ID of the container record to update.
end_time (str) – End time of the command for the operator_equipment history.
updates (Dict) – Dictionary with column names and their new values.
declaration_status (bool) – Declaration status of update. False is undeclared.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
transport_command_id (int, optional) – ID of the transport command. Exactly one of the command IDs has to be not None. (default is None)
building_command_id (int, optional) – ID of the building command. Exactly one of the command IDs has to be not None. (default is None)
processing_command_id (int, optional) – ID of the processing command. Exactly one of the command IDs has to be not None. (default is None)
inspection_command_id (int, optional) – ID of the inspection command. Exactly one of the command IDs has to be not None. (default is None)

Returns:

None

classmethod update_record(table_name: str, record_id: int, updates: Dict, cursor: cursor) → None¶

Updates a record in the specified table.

Parameters:

table_name (str) – Name of the table to update.
record_id (int) – ID of the record to update.
updates (Dict) – Dictionary with column names and their new values.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.

Returns:

None

classmethod update_room(room_id: int, end_time: str, updates: Dict, declaration_status: bool, cursor: cursor, building_command_id: int | None = None, inspection_command_id: int | None = None) → None¶

Updates a record in the ‘room’ table after writing to the ‘room_history’ table.

Parameters:

room_id (int) – ID of the room record to update.
end_time (str) – End time of the command for the room history.
updates (Dict) – Dictionary with column names and their new values.
declaration_status (bool) – Declaration status of update. False is undeclared.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
building_command_id (int, optional) – ID of the building command. Exactly one of the command IDs has to be not None. (default is None)
inspection_command_id (int, optional) – ID of the inspection command. Exactly one of the command IDs has to be not None. (default is None)

Returns:

None

classmethod update_safeguards_equipment(safeguards_equipment_id: int, end_time: str, updates: Dict, declaration_status: bool, cursor: cursor, processing_command_id: int | None = None, inspection_command_id: int | None = None) → None¶

Updates a record in the ‘safeguards_equipment’ table after writing to the ‘safeguards_equipment_history’ table.

Parameters:

container_id (int) – ID of the container record to update.
end_time (str) – End time of the command for the safeguards_equipment history.
updates (Dict) – Dictionary with column names and their new values.
declaration_status (bool) – Declaration status of update. False is undeclared.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
processing_command_id (int, optional) – ID of the processing command. Exactly one of the command IDs has to be not None. (default is None)
inspection_command_id (int, optional) – ID of the inspection command. Exactly one of the command IDs has to be not None. (default is None)

Returns:

None

class project.model.database.DatabaseRead¶

A class to interact with a PostgreSQL database. It provides methods to retrieve data.

_valid_target_types¶

A set of valid target types.

Type:: set[str]

_valid_command_types¶

A set of valid command types.

Type:: set[str]

_valid_equipment_types¶

A set of valid equipment types.

Type:: set[str]

_valid_matches¶

A set of valid (equipment type, command types) pairs.

Type:: set[Tuple]

_time_format¶

A string specifying the time format used to print history data.

Type:: str

_validate_equipment_type()¶: Validates the equipment type and returns the corresponding column or table name.

_validate_target_type()¶: Validates the target type and returns the corresponding column name.

_validate_command_type()¶: Validates the command type and returns the corresponding table or inventory table name.

_create_declaration_query_addon()¶: Creates a query addition for getter functions based on declaration requirement and status.

_create_history_query()¶: Generates an SQL query and parameters for getting a specified history table.

get_facility_type()¶: Retrieves a facility type record.

get_facility()¶: Retrieves a facility record.

get_room()¶: Retrieves a room record.

get_material()¶: Retrieves a material record.

get_building_element()¶: Retrieves a building element record.

get_holding_area()¶: Retrieves a holding area record.

get_container_type()¶: Retrieves a container type record.

get_container()¶: Retrieves a container record.

get_nuclear_material()¶: Retrieves a nuclear material record.

get_operator_equipment()¶: Retrieves an operator equipment record.

get_safeguards_equipment()¶: Retrieves a safeguards equipment record.

get_facility_ids()¶: Retrieves all facility IDs from the database.

get_room_ids_by_facility_id()¶: Retrieves all room IDs for a given facility.

get_room_ids_with_holding_areas()¶: Retrieves all IDs of rooms with holding areas.

get_room_id_by_container_id()¶: Retrieves room ID for a given container.

get_building_element_ids_by_room_id()¶: Retrieve Building Element IDs associated with a specific room by its ID.

get_holding_area_ids_by_room_id()¶: Retrieve Holding Area IDs associated with a specific room by its ID.

get_container_ids_by_holding_area_id()¶: Retrieve Container IDs associated with a specific holding area by its ID.

get_container_ids_by_room_id()¶: Retrieve Container IDs associated with a specific room by its ID.

get_container_ids_by_facility_id()¶: Retrieve Container IDs associated with a specific facility by its ID.

get_nuclear_material_ids_by_container_id()¶: Retrieve Nuclear Material IDs associated with a specific container by its ID.

get_nuclear_material_id_by_holding_area_id()¶: Retrieve Nuclear Material IDs associated with a specific holding area by its ID.

get_nuclear_material_ids_by_room_id()¶: Retrieve Nuclear Material IDs associated with a specific room by its ID.

get_nuclear_material_ids_by_facility_id()¶: Retrieve Nuclear Material IDs associated with a specific facility by its ID.

get_equipment_ids_by_target_id()¶: Retrieves equipment IDs used on a target based on the type of equipment, command, target type, and target ID.

get_equipment_ids_by_room_id()¶: Retrieves equipment IDs associated with a specific room by its ID.

get_equipment_ids_by_facility_id()¶: Retrieves equipment IDs associated with a specific facility by its ID.

get_command_ids()¶: Retrieves command IDs associated with a specific command type.

get_command_ids_by_target_id()¶: Retrieves command IDs associated with a specific target ID, target type, and command type.

get_command_ids_by_equipment_id()¶: Retrieves command IDs associated with a specific equipment ID, equipment type, and command type.

get_equipment_ids_by_command_id()¶: Retrieves equipment IDs associated with a specific command ID, command type, and equipment type.

get_transport_command()¶: Retrieves transport command record based on a given ID.

get_transport_node()¶: Retrieves transport node record based on a given transport node ID.

get_transport_node_ids()¶: Retrieves all transport node IDs.

get_transport_node_id_by_facility_id()¶: Retrieves a transport node ID based on a given facility ID.

get_transport_node_id_by_room_id()¶: Retrieves a transport node ID based on a given room ID.

get_transport_edge()¶: Retrieves a transport edge record based on a given transport edge ID.

get_transport_edge_ids()¶: Retrieves all transport edge IDs.

get_transport_edge_ids_by_source_node_id()¶: Retrieves transport edge IDs based on a given source node ID.

get_transport_edge_ids_by_target_node_id()¶: Retrieves transport edge IDs based on a given target node ID.

get_transport_edge_ids_by_building_element_id()¶: Retrieves transport edge IDs based on a given building element ID.

get_transport_steps_by_command_id()¶: Retrieves transport step records for a specific transport command ID.

get_transport_steps_by_transport_edge_id()¶: Retrieves transport step records for a specific transport edge ID.

get_transport_steps_by_transported_container_id()¶: Retrieves transport step records for a specific transported container ID.

get_building_command()¶: Retrieves building command record based on a given ID.

get_processing_command()¶: Retrieves processing command record based on a given ID.

get_inspection_command()¶: Retrieves inspection command record based on a given ID.

get_facility_history()¶: Retrieves historical data for a facility by ID, optionally filtered by time.

get_room_history()¶: Retrieves historical data for a room by ID, optionally filtered by time.

get_building_element_history()¶: Retrieves historical data for a building element by ID, optionally filtered by time.

get_holding_area_history()¶: Retrieves historical data for a holding area by ID, optionally filtered by time.

get_container_history()¶: Retrieves historical data for a container by ID, optionally filtered by time.

get_nuclear_material_history()¶: Retrieves historical data for nuclear materials by ID, optionally filtered by time.

get_operator_equipment_history()¶: Retrieves historical data for operator equipment by ID, optionally filtered by time.

get_safeguards_equipment_history()¶: Retrieves historical data for safeguards equipment by ID, optionally filtered by time.

get_geometry_data()¶: Retrieves the geometry data for a given facility, including building elements, containers, and nuclear materials.

classmethod get_building_command(building_command_id: int, cursor: cursor) → Dict | None¶

Retrieves building command record based on a given building command ID.

Parameters:

building_command_id (int) – The building command ID.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.

Returns:

The building command records or None if not found.

Return type:

Dict | None

classmethod get_building_element(building_element_id: int, cursor: cursor, declaration_status: bool | None = None) → Dict[str, Any] | None¶

Retrieves a building element record from the database based on building_element_id.

Parameters:

building_element_id (int) – ID of the building element.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
declaration_status (bool, optional) – If True, record of declared state is returned, possibly being a shadow record representing last declared state. If None or False, record of declared or undeclared state is returned. (default is None)

Returns:

Building element data, or None if no record is found.

Return type:

dict

classmethod get_building_element_history(building_element_id: int, cursor: cursor, query_time: str | None = None, declaration_status: bool | None = None) → Dict[str, Dict[str, Any]]¶

Retrieve historical data for a building element by ID, optionally filtered by time.

Parameters:

building_element_id (int) – The ID of the building element.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
query_time (str, optional) – Optional time limit in the format YYYY-MM-DD HH:MM:SS to which history shall be retrieved. (default is None) (default is None)
declaration_status (bool, optional) – If True, record of declared state is returned, possibly being a shadow record representing last declared state. If None or False, record of declared or undeclared state is returned. (default is None)

Returns:

A dictionary containing the historical data.

Return type:

Dict[str, Dict[str, Any]]

classmethod get_building_element_ids_by_room_id(room_id: int, cursor: cursor, declaration_requirement: bool | None = None, declaration_status: bool | None = None) → List[Tuple[int]]¶

Retrieve Building Element IDs by Room ID.

Parameters:

room_id (int) – The ID of the room.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
declaration_requirement (bool, optional) – If None, all records are returned. If True, only records that require dclaration are returned. If False, only records that do not require declaration are returned. (default is None)
declaration_status (bool, optional) – If None, declared und undeclared records are returned. If True, only declared records including possible shadow records, representing last declared state, are returned. If False, only undeclared records are returned. (default is None)

Returns:

A list of building element IDs associated with the room given as integer in a tuple.

Return type:

List[Tuple[int]]

classmethod get_command_ids(command_type: str, cursor: cursor, declaration_requirement: bool | None = None, declaration_status: bool | None = None) → List[Tuple[int]]¶

Retrieve all command IDs by command type.

Parameters:

command_type (str) – The type of command (e.g. ‘transport’, ‘inspection’, etc.)
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
declaration_requirement (bool, optional) – If None, all records are returned. If True, only records that require dclaration are returned. If False, only records that do not require declaration are returned. (default is None)
declaration_status (bool, optional) – If None, declared und undeclared records are returned. If True, only declared records including possible shadow records, representing last declared state, are returned. If False, only undeclared records are returned. (default is None)

Returns:

A list of command IDs associated with the target given as integer in a tuple.

Return type:

List[Tuple[int]]

classmethod get_command_ids_by_equipment_id(equipment_id: int, equipment_type: str, command_type: str, cursor: cursor, declaration_requirement: bool | None = None, declaration_status: bool | None = None) → List[Tuple[int]]¶

Retrieves command IDs by equipment ID, equipment type and command type.

Parameters:

equipment_id (int) – The ID of the equipment.
equipment_type (str) – The type of equipment (‘operator’ or ‘safeguards’).
command_type (str) – The type of command (e.g. ‘transport’, ‘inspection’, etc.)
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
declaration_requirement (bool, optional) – If None, all records are returned. If True, only records that require dclaration are returned. If False, only records that do not require declaration are returned. (default is None)
declaration_status (bool, optional) – If None, declared und undeclared records are returned. If True, only declared records including possible shadow records, representing last declared state, are returned. If False, only undeclared records are returned. (default is None)

Returns:

A list of command IDs associated with the equipment given as integer in a tuple.

Return type:

List[Tuple[int]]

classmethod get_command_ids_by_target_id(target_id: int, target_type: str, command_type: str, cursor: cursor, declaration_requirement: bool | None = None, declaration_status: bool | None = None) → List[Tuple[int]]¶

Retrieve command IDs by target ID, target type and command type.

Parameters:

target_id (int) – The ID of the target.
target_type (str) – The type of target (e.g., ‘container’, ‘room’, etc.).
command_type (str) – The type of command (e.g. ‘transport’, ‘inspection’, etc.)
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
declaration_requirement (bool, optional) – If None, all records are returned. If True, only records that require dclaration are returned. If False, only records that do not require declaration are returned. (default is None)
declaration_status (bool, optional) – If None, declared und undeclared records are returned. If True, only declared records including possible shadow records, representing last declared state, are returned. If False, only undeclared records are returned. (default is None)

Returns:

A list of command IDs associated with the target given as integer in a tuple.

Return type:

List[Tuple[int]]

classmethod get_container(container_id: int, cursor: cursor, declaration_status: bool | None = None) → Dict[str, Any] | None¶

Retrieves a container record from the database based on container_id.

Parameters:

container_id (int) – ID of the container.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
declaration_status (bool, optional) – If True, record of declared state is returned, possibly being a shadow record representing last declared state. If None or False, record of declared or undeclared state is returned. (default is None)

Returns:

Container data, or None if no record is found.

Return type:

dict

classmethod get_container_history(container_id: int, cursor: cursor, query_time: str | None = None, declaration_status: bool | None = None) → Dict[str, Dict[str, Any]]¶

Retrieve historical data for a container by ID, optionally filtered by time.

Parameters:

container_id (int) – The ID of the container.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
query_time (str, optional) – Optional time limit in the format YYYY-MM-DD HH:MM:SS to which history shall be retrieved. (default is None) (default is None)
declaration_status (bool, optional) – If True, record of declared state is returned, possibly being a shadow record representing last declared state. If None or False, record of declared or undeclared state is returned. (default is None)

Returns:

A dictionary containing the historical data.

Return type:

Dict[str, Dict[str, Any]]

classmethod get_container_ids_by_facility_id(facility_id: int, cursor: cursor, declaration_requirement: bool | None = None, declaration_status: bool | None = None) → List[Tuple[int]]¶

Retrieve Container IDs by Facility ID.

Parameters:

facility_id (int) – The ID of the facility.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
declaration_requirement (bool, optional) – If None, all records are returned. If True, only records that require dclaration are returned. If False, only records that do not require declaration are returned. (default is None)
declaration_status (bool, optional) – If None, declared und undeclared records are returned. If True, only declared records including possible shadow records, representing last declared state, are returned. If False, only undeclared records are returned. (default is None)

Returns:

A list of container IDs associated with the facility given as integer in a tuple.

Return type:

List[Tuple[int]]

classmethod get_container_ids_by_holding_area_id(holding_area_id: int, cursor: cursor, declaration_requirement: bool | None = None, declaration_status: bool | None = None) → List[Tuple[int]]¶

Retrieve Container IDs by Holding Area ID.

Parameters:

holding_area_id (int) – The ID of the holding area.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
declaration_requirement (bool, optional) – If None, all records are returned. If True, only records that require dclaration are returned. If False, only records that do not require declaration are returned. (default is None)
declaration_status (bool, optional) – If None, declared und undeclared records are returned. If True, only declared records including possible shadow records, representing last declared state, are returned. If False, only undeclared records are returned. (default is None)

Returns:

A list of container IDs associated with the holding area given as integer in a tuple.

Return type:

List[Tuple[int]]

classmethod get_container_ids_by_room_id(room_id: int, cursor: cursor, declaration_requirement: bool | None = None, declaration_status: bool | None = None) → List[Tuple[int]]¶

Retrieve Container IDs by Room ID.

Parameters:

room_id (int) – The ID of the room.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
declaration_requirement (bool, optional) – If None, all records are returned. If True, only records that require dclaration are returned. If False, only records that do not require declaration are returned. (default is None)
declaration_status (bool, optional) – If None, declared und undeclared records are returned. If True, only declared records including possible shadow records, representing last declared state, are returned. If False, only undeclared records are returned. (default is None)

Returns:

A list of container IDs associated with the room given as integer in a tuple.

Return type:

List[Tuple[int]]

classmethod get_container_type(container_type_name: str, cursor: cursor) → Dict[str, Any] | None¶

Retrieves a container type record from the database based on container_name.

Parameters:

container_type_name (str) – Name of the container type.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.

Returns:

Container type data, or None if no record is found.

Return type:

dict

classmethod get_equipment_ids_by_command_id(command_id: int, command_type: str, equipment_type: str, cursor: cursor, declaration_requirement: bool | None = None, declaration_status: bool | None = None) → List[Tuple[int]]¶

Retrieves equipment IDs by command ID, command type and equipment type.

Parameters:

command_id (int) – The ID of the command.
command_type (str) – The type of command (e.g. ‘transport’, ‘inspection’, etc.)
equipment_type (str) – The type of equipment (‘operator’ or ‘safeguards’).
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
declaration_requirement (bool, optional) – If None, all records are returned. If True, only records that require dclaration are returned. If False, only records that do not require declaration are returned. (default is None)
declaration_status (bool, optional) – If None, declared und undeclared records are returned. If True, only declared records including possible shadow records, representing last declared state, are returned. If False, only undeclared records are returned. (default is None)

Returns:

A list of equipment IDs associated with the command given as integer in a tuple.

Return type:

List[Tuple[int]]

classmethod get_equipment_ids_by_facility_id(equipment_type: str, facility_id: int, cursor: cursor, declaration_requirement: bool | None = None, declaration_status: bool | None = None) → List[Tuple[int]]¶

Retrieve IDs of equipment located within a facility given by its ID.

Parameters:

equipment_type (str) – The type of equipment (‘safeguards’ or ‘operator’).
facility_id (int) – The ID of the facility.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
declaration_requirement (bool, optional) – If None, all records are returned. If True, only records that require dclaration are returned. If False, only records that do not require declaration are returned. (default is None)
declaration_status (bool, optional) – If None, declared und undeclared records are returned. If True, only declared records including possible shadow records, representing last declared state, are returned. If False, only undeclared records are returned. (default is None)

Returns:

A list of operator equipment IDs associated with the facility given as integer in a tuple.

Return type:

List[Tuple[int]]

classmethod get_equipment_ids_by_room_id(equipment_type: str, room_id: int, cursor: cursor, declaration_requirement: bool | None = None, declaration_status: bool | None = None) → List[Tuple[int]]¶

Retrieve IDs of equipment located within a room given by its ID.

Parameters:

equipment_type (str) – The type of equipment (‘safeguards’ or ‘operator’).
room_id (int) – The ID of the room.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
declaration_requirement (bool, optional) – If None, all records are returned. If True, only records that require dclaration are returned. If False, only records that do not require declaration are returned. (default is None)
declaration_status (bool, optional) – If None, declared und undeclared records are returned. If True, only declared records including possible shadow records, representing last declared state, are returned. If False, only undeclared records are returned. (default is None)

Returns:

A list of operator equipment IDs associated with the room given as integer in a tuple.

Return type:

List[Tuple[int]]

classmethod get_equipment_ids_by_target_id(equipment_type: str, command_type: str, target_type: str, target_id: int, cursor: cursor, declaration_requirement: bool | None = None, declaration_status: bool | None = None) → List[Tuple[int]]¶

Retrieve IDs of equipment used on a target given by its type and ID.

Parameters:

equipment_type (str) – The type of equipment (‘safeguards’ or ‘operator’).
command_type (str) – The type of operator command (e.g. ‘transport’, ‘building’)
target_type (str) – The type of target (e.g., ‘container’, ‘room’).
target_id (int) – The ID of the target.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
declaration_requirement (bool, optional) – If None, all records are returned. If True, only records that require dclaration are returned. If False, only records that do not require declaration are returned. (default is None)
declaration_status (bool, optional) – If None, declared und undeclared records are returned. If True, only declared records including possible shadow records, representing last declared state, are returned. If False, only undeclared records are returned. (default is None)

Returns:

A list of operator equipment IDs associated with the target given as integer in a tuple.

Return type:

List[Tuple[int]]

classmethod get_facility(facility_id: int, cursor: cursor, declaration_status: bool | None = None) → Dict[str, Any] | None¶

Retrieves a facility record from the database based on facility_id.

Parameters:

facility_id (int) – ID of the facility.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
declaration_status (bool, optional) – If True, record of declared state is returned, possibly being a shadow record representing last declared state. If None or False, record of declared or undeclared state is returned. (default is None)

Returns:

Facility data, or None if no record is found.

Return type:

dict

classmethod get_facility_history(facility_id: int, cursor: cursor, query_time: str | None = None, declaration_status: bool | None = None) → Dict[str, Dict[str, Any]]¶

Retrieve historical data for a facility by ID, optionally filtered by time.

Parameters:

facility_id (int) – The ID of the facility.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
query_time (str, optional) – Optional time limit in the format YYYY-MM-DD HH:MM:SS to which history shall be retrieved. (default is None) (default is None)
declaration_status (bool, optional) – If True, record of declared state is returned, possibly being a shadow record representing last declared state. If None or False, record of declared or undeclared state is returned. (default is None)

Returns:

A dictionary containing the historical data.

Return type:

Dict[str, Dict[str, Any]]

classmethod get_facility_ids(cursor: cursor, declaration_requirement: bool | None = None, declaration_status: bool | None = None) → List[Tuple[int]]¶

Retrieve Facility IDs from database.

Parameters:

cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
declaration_requirement (bool, optional) – If None, all records are returned. If True, only records that require dclaration are returned. If False, only records that do not require declaration are returned. (default is None)
declaration_status (bool, optional) – If None, declared und undeclared records are returned. If True, only declared records including possible shadow records, representing last declared state, are returned. If False, only undeclared records are returned. (default is None)

Returns:

A list of facility IDs given as integer in a tuple.

Return type:

List[Tuple[int]]

classmethod get_facility_ids_with_holding_areas(cursor: cursor, declaration_requirement: bool | None = None, declaration_status: bool | None = None) → List[Tuple[int]]¶

Retrieve Facility IDs from database.

Parameters:

cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
declaration_requirement (bool, optional) – If None, all records are returned. If True, only records that require dclaration are returned. If False, only records that do not require declaration are returned. (default is None)
declaration_status (bool, optional) – If None, declared und undeclared records are returned. If True, only declared records including possible shadow records, representing last declared state, are returned. If False, only undeclared records are returned. (default is None)

Returns:

A list of facility IDs given as integer in a tuple.

Return type:

List[Tuple[int]]

classmethod get_facility_type(facility_type_name: str, cursor: cursor) → Dict[str, Any] | None¶

Retrieves a facility type record from the database based on facility_type_name.

Parameters:

facility_id (int) – ID of the facility.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.

Returns:

Facility type data, or None if no record is found.

Return type:

dict

classmethod get_geometry_data(facility_id: int, cursor: cursor) → Dict¶

Retrieves the geometry data for a given facility, including building elements, containers, and nuclear materials.

Parameters:

facility_id (int) – The ID of the facility.
cursor – Database cursor for executing queries.

Returns:

A dictionary containing geometry data, including world radius, building elements, containers, and nuclear materials.

Return type:

Dict

classmethod get_holding_area(holding_area_id: int, cursor: cursor, declaration_status: bool | None = None) → Dict[str, Any] | None¶

Retrieves a holding area record from the database based on holding_area_id.

Parameters:

holding_area_id (int) – ID of the holding area.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
declaration_status (bool, optional) – If True, record of declared state is returned, possibly being a shadow record representing last declared state. If None or False, record of declared or undeclared state is returned. (default is None)

Returns:

Holding area data, or None if no record is found.

Return type:

dict

classmethod get_holding_area_history(holding_area_id: int, cursor: cursor, query_time: str | None = None, declaration_status: bool | None = None) → Dict[str, Dict[str, Any]]¶

Retrieve historical data for a holding area by ID, optionally filtered by time.

Parameters:

holding_area_id (int) – The ID of the holding area.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
query_time (str, optional) – Optional time limit in the format YYYY-MM-DD HH:MM:SS to which history shall be retrieved. (default is None) (default is None)
declaration_status (bool, optional) – If True, record of declared state is returned, possibly being a shadow record representing last declared state. If None or False, record of declared or undeclared state is returned. (default is None)

Returns:

A dictionary containing the historical data.

Return type:

Dict[str, Dict[str, Any]]

classmethod get_holding_area_ids_by_room_id(room_id: int, cursor: cursor, declaration_requirement: bool | None = None, declaration_status: bool | None = None) → List[Tuple[int]]¶

Retrieve Holding Area IDs by Room ID.

Parameters:

room_id (int) – The ID of the room.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
declaration_requirement (bool, optional) – If None, all records are returned. If True, only records that require dclaration are returned. If False, only records that do not require declaration are returned. (default is None)
declaration_status (bool, optional) – If None, declared und undeclared records are returned. If True, only declared records including possible shadow records, representing last declared state, are returned. If False, only undeclared records are returned. (default is None)

Returns:

A list of holding area IDs associated with the room given as integer in a tuple.

Return type:

List[Tuple[int]]

classmethod get_inspection_command(inspection_command_id: int, cursor: cursor) → Dict | None¶

Retrieves inspection command record based on a given inspection command ID.

Parameters:

inspection_command_id (int) – The inspection command ID.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.

Returns:

The inspection command records or None if not found.

Return type:

Dict | None

classmethod get_material(material_name: str, cursor: cursor) → Dict[str, Any] | None¶

Retrieves a material record from the database based on material_name.

Parameters:

material_name (str) – Name of the material.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.

Returns:

Material data, or None if no record is found.

Return type:

dict

classmethod get_nuclear_material(nuclear_material_id: int, cursor: cursor, declaration_status: bool | None = None) → Dict[str, Any] | None¶

Retrieves a nuclear material record from the database based on nuclear_material_id.

Parameters:

nuclear_material_id (int) – ID of the nuclear material.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
declaration_status (bool, optional) – If True, record of declared state is returned, possibly being a shadow record representing last declared state. If None or False, record of declared or undeclared state is returned. (default is None)

Returns:

Nuclear material data, or None if no record is found.

Return type:

dict

classmethod get_nuclear_material_history(nuclear_material_id: int, cursor: cursor, query_time: str | None = None, declaration_status: bool | None = None) → Dict[str, Dict[str, Any]]¶

Retrieve historical data for nuclear materials by ID, optionally filtered by time.

Parameters:

nuclear_material_id (int) – The ID of the nuclear material.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
query_time (str, optional) – Optional time limit in the format YYYY-MM-DD HH:MM:SS to which history shall be retrieved. (default is None) (default is None)
declaration_status (bool, optional) – If True, record of declared state is returned, possibly being a shadow record representing last declared state. If None or False, record of declared or undeclared state is returned. (default is None)

Returns:

A dictionary containing the historical data.

Return type:

Dict[str, Dict[str, Any]]

classmethod get_nuclear_material_id_by_holding_area_id(holding_area_id: int, cursor: cursor, declaration_requirement: bool | None = None, declaration_status: bool | None = None) → Tuple[int]¶

Retrieve Nuclear Material ID by Holding Area ID.

Parameters:

holding_area_id (int) – The ID of the holding area.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
declaration_requirement (bool, optional) – If None, all records are returned. If True, only records that require dclaration are returned. If False, only records that do not require declaration are returned. (default is None)
declaration_status (bool, optional) – If None, declared und undeclared records are returned. If True, only declared records including possible shadow records, representing last declared state, are returned. If False, only undeclared records are returned. (default is None)

Returns:

A nuclear material ID associated with the holding area given as integer in a tuple.

Return type:

Tuple[int]

classmethod get_nuclear_material_ids_by_container_id(container_id: int, cursor: cursor, declaration_requirement: bool | None = None, declaration_status: bool | None = None) → List[Tuple[int]]¶

Retrieve Nuclear Material IDs by Container ID.

Parameters:

container_id (int) – The ID of the container.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
declaration_requirement (bool, optional) – If None, all records are returned. If True, only records that require dclaration are returned. If False, only records that do not require declaration are returned. (default is None)
declaration_status (bool, optional) – If None, declared und undeclared records are returned. If True, only declared records including possible shadow records, representing last declared state, are returned. If False, only undeclared records are returned. (default is None)

Returns:

A list of nuclear material IDs associated with the container given as integer in a tuple.

Return type:

List[Tuple[int]]

classmethod get_nuclear_material_ids_by_facility_id(facility_id: int, cursor: cursor, declaration_requirement: bool | None = None, declaration_status: bool | None = None) → List[Tuple[int]]¶

Retrieve Nuclear Material IDs by Facility ID.

Parameters:

facility_id (int) – The ID of the facility.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
declaration_requirement (bool, optional) – If None, all records are returned. If True, only records that require dclaration are returned. If False, only records that do not require declaration are returned. (default is None)
declaration_status (bool, optional) – If None, declared und undeclared records are returned. If True, only declared records including possible shadow records, representing last declared state, are returned. If False, only undeclared records are returned. (default is None)

Returns:

A list of nuclear material IDs associated with the facility given as integer in a tuple.

Return type:

List[Tuple[int]]

classmethod get_nuclear_material_ids_by_room_id(room_id: int, cursor: cursor, declaration_requirement: bool | None = None, declaration_status: bool | None = None) → List[Tuple[int]]¶

Retrieve Nuclear Material IDs by Room ID.

Parameters:

facility_id (int) – The ID of the room.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
declaration_requirement (bool, optional) – If None, all records are returned. If True, only records that require dclaration are returned. If False, only records that do not require declaration are returned. (default is None)
declaration_status (bool, optional) – If None, declared und undeclared records are returned. If True, only declared records including possible shadow records, representing last declared state, are returned. If False, only undeclared records are returned. (default is None)

Returns:

A list of nuclear material IDs associated with the room given as integer in a tuple.

Return type:

List[Tuple[int]]

classmethod get_operator_equipment(operator_equipment_id: int, cursor: cursor, declaration_status: bool | None = None) → Dict[str, Any] | None¶

Retrieves an operator equipment record from the database based on operator_equipment_id.

Parameters:

operator_equipment_id (int) – ID of the operator equipment.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
declaration_status (bool, optional) – If True, record of declared state is returned, possibly being a shadow record representing last declared state. If None or False, record of declared or undeclared state is returned. (default is None)

Returns:

Operator equipment data, or None if no record is found.

Return type:

dict

classmethod get_operator_equipment_history(operator_equipment_id: int, cursor: cursor, query_time: str | None = None, declaration_status: bool | None = None) → Dict[str, Dict[str, Any]]¶

Retrieve historical data for operator equipment by ID, optionally filtered by time.

Parameters:

operator_equipment_id (int) – The ID of the operator equipment.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
query_time (str, optional) – Optional time limit in the format YYYY-MM-DD HH:MM:SS to which history shall be retrieved. (default is None) (default is None)
declaration_status (bool, optional) – If True, record of declared state is returned, possibly being a shadow record representing last declared state. If None or False, record of declared or undeclared state is returned. (default is None)

Returns:

A dictionary containing the historical data.

Return type:

Dict[str, Dict[str, Any]]

classmethod get_processing_command(processing_command_id: int, cursor: cursor) → Dict | None¶

Retrieves processing command record based on a given processing command ID.

Parameters:

processing_command_id (int) – The processing command ID.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.

Returns:

The processing command records or None if not found.

Return type:

Dict | None

classmethod get_room(room_id: int, cursor: cursor, declaration_status: bool | None = None) → Dict[str, Any] | None¶

Retrieves a room record from the database based on room_id.

Parameters:

room_id (int) – ID of the room.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
declaration_status (bool, optional) – If True, record of declared state is returned, possibly being a shadow record representing last declared state. If None or False, record of declared or undeclared state is returned. (default is None)

Returns:

Room data, or None if no record is found.

Return type:

dict

classmethod get_room_history(room_id: int, cursor: cursor, query_time: str | None = None, declaration_status: bool | None = None) → Dict[str, Dict[str, Any]]¶

Retrieve historical data for a room by ID, optionally filtered by time.

Parameters:

room_id (int) – The ID of the room.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
query_time (str, optional) – Optional time limit in the format YYYY-MM-DD HH:MM:SS to which history shall be retrieved. (default is None) (default is None)
declaration_status (bool, optional) – If True, record of declared state is returned, possibly being a shadow record representing last declared state. If None or False, record of declared or undeclared state is returned. (default is None)

Returns:

A dictionary containing the historical data.

Return type:

Dict[str, Dict[str, Any]]

classmethod get_room_id_by_container_id(container_id: int, cursor: cursor) → Tuple[int]¶

Retrieve Room ID by Container ID.

Parameters:

container_id (int) – The ID of the container.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.

Returns:

A room ID associated with the container given as integer in a tuple.

Return type:

Tuple[int]

classmethod get_room_ids_by_facility_id(facility_id: int, cursor: cursor, declaration_requirement: bool | None = None, declaration_status: bool | None = None) → List[Tuple[int]]¶

Retrieve Room IDs by Facility ID.

Parameters:

facility_id (int) – The ID of the facility.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
declaration_requirement (bool, optional) – If None, all records are returned. If True, only records that require dclaration are returned. If False, only records that do not require declaration are returned. (default is None)
declaration_status (bool, optional) – If None, declared und undeclared records are returned. If True, only declared records including possible shadow records, representing last declared state, are returned. If False, only undeclared records are returned. (default is None)

Returns:

A list of room IDs associated with the facility given as integer in a tuple.

Return type:

List[Tuple[int]]

classmethod get_room_ids_with_holding_areas(cursor: cursor, declaration_requirement: bool | None = None, declaration_status: bool | None = None) → List[Tuple[int]]¶

Retrieve IDs of rooms with holding areas.

Parameters:

cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
declaration_requirement (bool, optional) – If None, all records are returned. If True, only records that require dclaration are returned. If False, only records that do not require declaration are returned. (default is None)
declaration_status (bool, optional) – If None, declared und undeclared records are returned. If True, only declared records including possible shadow records, representing last declared state, are returned. If False, only undeclared records are returned. (default is None)

Returns:

A list of IDs of rooms with holding areas as integer in a tuple.

Return type:

List[Tuple[int]]

classmethod get_safeguards_equipment(safeguards_equipment_id: int, cursor: cursor, declaration_status: bool | None = None) → Dict[str, Any] | None¶

Retrieves a safeguards equipment record from the database based on safeguards_equipment_id.

Parameters:

safeguards_equipment_id (int) – ID of the safeguards equipment.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
declaration_status (bool, optional) – If True, record of declared state is returned, possibly being a shadow record representing last declared state. If None or False, record of declared or undeclared state is returned. (default is None)

Returns:

Safeguards equipment data, or None if no record is found.

Return type:

dict

classmethod get_safeguards_equipment_history(safeguards_equipment_id: int, cursor: cursor, query_time: str | None = None, declaration_status: bool | None = None) → Dict[str, Dict[str, Any]]¶

Retrieve historical data for safeguards equipment by ID, optionally filtered by time.

Parameters:

safeguards_equipment_id (int) – The ID of the safeguards equipment.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
query_time (str, optional) – Optional time limit in the format YYYY-MM-DD HH:MM:SS to which history shall be retrieved. (default is None) (default is None)
declaration_status (bool, optional) – If True, record of declared state is returned, possibly being a shadow record representing last declared state. If None or False, record of declared or undeclared state is returned. (default is None)

Returns:

A dictionary containing the historical data.

Return type:

Dict[str, Dict[str, Any]]

get_shadow_id(record_id: int, cursor: cursor) → int | None¶

Retrieves ID of shadow row for given table name and record ID, if a shadow exists.

Parameters:

table_name (str) – Name of the table to update.
record_id (int) – ID of the record to update.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.

Returns:

ID of shadow corresponding to record ID, or None if no shadow is found.

Return type:

int

classmethod get_transport_command(transport_command_id: int, cursor: cursor) → Dict | None¶

Retrieves a transport command record based on a given transport command ID.

Parameters:

transport_command_id (int) – The ID of the transport command.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.

Returns:

The transport command records or None if not found.

Return type:

Dict | None

classmethod get_transport_edge(transport_edge_id: int, cursor: cursor) → Dict | None¶

Retrieves a transport edge record based on a given transport edge ID.

Parameters:

transport_edge_id (int) – The ID of the transport edge.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.

Returns:

The transport edge records or None if not found.

Return type:

Dict | None

classmethod get_transport_edge_ids(cursor: cursor) → List[Tuple[int]]¶

Retrieves all transport edge IDs.

Parameters:: cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
Returns:: A list of all transport edge IDs as integer in a tuple.
Return type:: List[Tuple[int]]

classmethod get_transport_edge_ids_by_building_element_id(building_element_id: int, cursor: cursor) → List[Tuple[int]]¶

Retrieves transport edge IDs based on a given building element ID.

Parameters:

building_element_id (int) – The ID of the linked building element.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.

Returns:

A list of transport edge IDs as integer in a tuple. Length of list is 1 for one-directional connection and 2 for bi-directional connection.

Return type:

List[Tuple[int]]

classmethod get_transport_edge_ids_by_source_node_id(source_node_id: int, cursor: cursor) → List[Tuple[int]]¶

Retrieves transport edge IDs based on a given source node ID.

Parameters:

source_node_id (int) – The ID of the source node.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.

Returns:

A list of transport edge IDs as integer in a tuple.

Return type:

List[Tuple[int]]

classmethod get_transport_edge_ids_by_target_node_id(target_node_id: int, cursor: cursor) → List[Tuple[int]]¶

Retrieves transport edge IDs based on a given target node ID.

Parameters:

target_node_id (int) – The ID of the target node.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.

Returns:

A list of transport edge IDs as integer in a tuple.

Return type:

List[Tuple[int]]

classmethod get_transport_node(transport_node_id: int, cursor: cursor) → Dict | None¶

Retrieves a transport node record based on a given transport node ID.

Parameters:

transport_node_id (int) – The ID of the transport node.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.

Returns:

The transport node records or None if not found.

Return type:

Dict | None

classmethod get_transport_node_id_by_facility_id(facility_id: int, cursor: cursor) → int | None¶

Retrieves a transport node ID based on a given facility ID.

Parameters:

facility_id (int) – The ID of the facility.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.

Returns:

The transport node ID or None if not found.

Return type:

int | None

classmethod get_transport_node_id_by_room_id(room_id: int, cursor: cursor) → int | None¶

Retrieves a transport node ID based on a given room ID.

Parameters:

room_id (int) – The ID of the room.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.

Returns:

The transport node ID or None if not found.

Return type:

int | None

classmethod get_transport_node_ids(cursor: cursor) → List[Tuple[int]]¶

Retrieves all transport node IDs.

Parameters:: cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
Returns:: A list of all transport node IDs as integer in a tuple.
Return type:: List[Tuple[int]]

classmethod get_transport_steps_by_command_id(transport_command_id: int, cursor: cursor) → List[Dict]¶

Retrieves transport step records based on a given transport command ID.

Parameters:

transport_command_id (int) – The transport command ID.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.

Returns:

A list of dictionaries containing transport step data.

Return type:

List[Dict]

classmethod get_transport_steps_by_transport_edge_id(transport_edge_id: int, cursor: cursor, declaration_requirement: bool | None = None, declaration_status: bool | None = None) → List[Dict]¶

Retrieves transport step records based on a given transport edge ID.

Parameters:

transport_edge_id (int) – The entry/exit ID.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
declaration_requirement (bool, optional) – If None, all records are returned. If True, only records that require dclaration are returned. If False, only records that do not require declaration are returned. (default is None)
declaration_status (bool, optional) – If None, declared und undeclared records are returned. If True, only declared records including possible shadow records, representing last declared state, are returned. If False, only undeclared records are returned. (default is None)

Returns:

A list of dictionaries containing transport step data.

Return type:

List[Dict]

classmethod get_transport_steps_by_transported_container_id(transported_container_id: int, cursor: cursor, declaration_requirement: bool | None = None, declaration_status: bool | None = None) → List[Dict]¶

Retrieves transport step records based on a given transported container ID.

Parameters:

transported_container_id (int) – The transported container ID.
cursor (psycopg2.cursor) – The database cursor for executing SQL queries.
declaration_requirement (bool, optional) – If None, all records are returned. If True, only records that require dclaration are returned. If False, only records that do not require declaration are returned. (default is None)
declaration_status (bool, optional) – If None, declared und undeclared records are returned. If True, only declared records including possible shadow records, representing last declared state, are returned. If False, only undeclared records are returned. (default is None)

Returns:

A list of dictionaries containing transport step data.

Return type:

List[Dict]

class project.model.commander.TransportCommander(pathfinder: PathFinder)¶

A class that creates transport commands based on user input.

_pathfinder¶

Class-level reference to Pathfinder instance.

Type:: PathFinder

get_pathfinder()¶: Gets class-level PathFinder instance.

issue_transport_command()¶: Creates transport_command record in database, sets transport route, updates transport equipment inventory and related history tables.

_set_route()¶: Verify and sets a given route or, in case of no given route, chooses and sets a valid route.

_insert_transport_route()¶: Inserts transport route stepwise into the database.

__init__(pathfinder: PathFinder)¶

Initializes transport commander instance.

Parameters:: pathfinder (PathFinder) – Pathfinder instance.

_insert_transport_route(route: List[int], transport_command_id: int, transported_container_id: int, start_time: str, end_time: str, cursor: cursor)¶

Inserts transport route stepwise into the database.

For each transport step, it is assumed that an equal part of the duration between start and end time have elapsed.

Parameters:

route (List[int], optional) – List of transport_edge IDs intended to be used as route of transport. (default is None)
transport_command_id (int) – Transport command ID.
transported_container_id (int) – The ID of the container being transported.
start_time (str) – Start time in the format YYYY-MM-DD HH:MM:SS.
end_time (str) – End time in the format YYYY-MM-DD HH:MM:SS.
cursor (cursor)

Returns:

None

_pathfinder: ClassVar[PathFinder] = None¶

_set_route(transport_command_id: int, target_container_id: int, destination_holding_area_id: int, declaration_status: bool, start_time: str, end_time: str, cursor: cursor, route: List[int] | None = None) → None¶

Verify and sets a given route or, in case of no given route, chooses and sets a valid route. In case of multiple possible route, prompts user to choose from a list of possible transport routes.

Parameters:

transport_command_id (int) – Transport command ID.
origin_holding_area_id (int) – Origin holding area ID.
destination_holding_area_id (int) – Destination holding area ID.
declaration_status (bool) – Declaration status of transport. False is undeclared.
start_time (str) – Start time in the format YYYY-MM-DD HH:MM:SS.
end_time (str) – End time in the format YYYY-MM-DD HH:MM:SS.
cursor (cursor) – Cursor specifingy connection to database.
route (List[int], optional) – List of transport_edge IDs intended to be used as route of transport. (default is None)

classmethod get_ongoing_commands_at_time(time: str)¶

Creates a History instance with ongoing commands at a specific time

Parameters:: time (str) – Time in the format YYYY-MM-DD HH:MM:SS.
Returns:: History instance with ongoing commands.
Return type:: History

get_pathfinder() → PathFinder¶

Gets class-level PathFinder instance.

Returns:: Pathfinder instance.
Return type:: PathFinder

issue_transport_command(target_container_id: int, destination_holding_area_id: int, transport_type: str, start_time: str, end_time: str, equipment_id_list: List[int], declaration_requirement: bool, declaration_status: bool, cursor: cursor, route: List[int] | None = None) → None¶

Creates transport_command record in database, sets transport route, updates transport equipment inventory and related history tables.

Parameters:

tatarget_container_idrget (int) – Targeted container instance.
destination_holding_area_id (int) – Destination of transport.
type (str) – Type of transport (e.g. ‘hauling’, “lifting’, ‘lowering’)
start_time (str) – Start time in the format YYYY-MM-DD HH:MM:SS.
end_time (str) – End time in the format YYYY-MM-DD HH:MM:SS.
equipment_id_list (List[int]) – List of IDs of operator equipent that is involved in transport.
declaration_requirement (bool) – Declaration requirement of transport. True means, declaration is required.
declaration_status (bool) – Declaration status of transport. False is undeclared.
cursor (cursor) – Cursor specifingy connection to database.
route (List[int], optional) – List of transport_edge IDs intended as route of transport. If no route is provided, valid route will be chosen automatically. If no route provided and multiple routes are possible, the user will be given choice to select preferred option. (default is None)

class project.model.commander.MonitoringSystem¶

A class governing global variables.

_global_time¶

Class-level global time in datetime format.

Type:: datetime

_verbosity¶

Class-level verbosity setting.

Type:: int

Methods

set_verbosity(level: int) -> None: Sets the verbosity level.
get_verbosity() -> int: Gets the verbosity level.
get_time() -> str: Gets current global time in the format YYYY-MM-DD HH:MM:SS.
set_time(new_time: datetime) -> None: Sets global time to a new value.
get_ongoing_commands_at_time(time: str) -> History: Creates a History instance with ongoing commands at a specific time.

_global_time: ClassVar[datetime] = datetime.datetime(2024, 1, 1, 0, 0)¶

_verbosity: ClassVar[int] = 1¶

classmethod get_time() → str¶

Gets current global time in the format YYYY-MM-DD HH:MM:SS.

Returns:: Current global time.
Return type:: str

classmethod get_verbosity() → int¶

Gets the verbosity level.

Returns:: Verbosity level (0: Silent, 1: Verbose).
Return type:: int

classmethod set_time(new_time: datetime) → None¶

Sets global time to a new value.

Parameters:: new_time (datetime) – New global time.

classmethod set_verbosity(level: int) → None¶

Sets the verbosity level.

Parameters:: level (int) – Verbosity level (0: Silent, 1: Verbose).

Impressum¶

Verantwortliche nach § 18, Abs. 2, Medienstaatsvertrag (MStV):

Manuel Kreutle

Institute of Fusion Energy and Nuclear Waste Management (IFN)
IFN-2: Nuclear Waste Management
Nuclear Safeguards and Security

Forschungszentrum Jülich
Wilhelm-Johnen-Straße
52428 Jülich

Phone:  +49 2461 61-5829
Email:  m.kreutle@fz-juelich.de