src.digitaltwin.instructions_records_to_db

This script processes ‘static_boundary_instructions’ records, validates URLs and instruction fields, and stores them in the ‘geospatial_layers’ table of the database.

Attributes

log

Functions

validate_url_reachability(→ None)

Validate the URL by checking its format and reachability.

validate_instruction_fields(→ None)

Validate the fields of an instruction.

read_and_check_instructions_file(→ pandas.DataFrame)

Read and check the static_boundary_instructions file, validating URLs and instruction fields.

get_existing_geospatial_layers(→ pandas.DataFrame)

Retrieve existing geospatial layers from the 'geospatial_layers' table.

get_non_existing_records(→ pandas.DataFrame)

Get 'static_boundary_instructions' records that are not available in the database.

store_instructions_records_to_db(→ None)

Store 'static_boundary_instructions' records in the 'geospatial_layers' table in the database.

Module Contents

src.digitaltwin.instructions_records_to_db.log
src.digitaltwin.instructions_records_to_db.validate_url_reachability(section: str, url: str) → None

Validate the URL by checking its format and reachability.

Parameters:

  • section (str) – The section identifier of the instruction.

  • url (str) – The URL to validate.

Returns:

This function does not return any value.

Return type:

None

Raises:

ValueError –

  • If the URL is invalid.

  • If the URL is unreachable.

src.digitaltwin.instructions_records_to_db.validate_instruction_fields(section: str, instruction: Dict[str, str | int]) → None

Validate the fields of an instruction. Each instruction should provide either ‘coverage_area’ or ‘unique_column_name’, but not both.

Parameters:

  • section (str) – The section identifier of the instruction.

  • instruction (Dict[str, Union[str, int]]) – The instruction details.

Returns:

This function does not return any value.

Return type:

None

Raises:

ValueError –

  • If both ‘coverage_area’ and ‘unique_column_name’ are provided.

  • If both ‘coverage_area’ and ‘unique_column_name’ are not provided.

src.digitaltwin.instructions_records_to_db.read_and_check_instructions_file() → pandas.DataFrame

Read and check the static_boundary_instructions file, validating URLs and instruction fields.

Returns:

The processed instructions DataFrame.

Return type:

pd.DataFrame

src.digitaltwin.instructions_records_to_db.get_existing_geospatial_layers(engine: sqlalchemy.engine.Engine) → pandas.DataFrame

Retrieve existing geospatial layers from the ‘geospatial_layers’ table.

Parameters:

engine (Engine) – The engine used to connect to the database.

Returns:

Data frame containing the existing geospatial layers.

Return type:

pd.DataFrame

src.digitaltwin.instructions_records_to_db.get_non_existing_records(instructions_df: pandas.DataFrame, existing_layers_df: pandas.DataFrame) → pandas.DataFrame

Get ‘static_boundary_instructions’ records that are not available in the database.

Parameters:

  • instructions_df (pd.DataFrame) – Data frame containing the ‘static_boundary_instructions’ records.

  • existing_layers_df (pd.DataFrame) – Data frame containing the existing ‘static_boundary_instructions’ records from the database.

Returns:

Data frame containing the ‘static_boundary_instructions’ records that are not available in the database.

Return type:

pd.DataFrame

src.digitaltwin.instructions_records_to_db.store_instructions_records_to_db(engine: sqlalchemy.engine.Engine) → None

Store ‘static_boundary_instructions’ records in the ‘geospatial_layers’ table in the database.

Parameters:

engine (Engine) – The engine used to connect to the database.

Returns:

This function does not return any value.

Return type:

None