Calibration Store Database Connection

This module implements the connection to the calibration database.

class CalibrationStoreDatabaseConnection(logger: Logger, communication_state_callback: Callable[[ska_control_model.CommunicationStatus], None], database_host: str, database_port: int, database_name: str, database_admin_user: str, database_admin_password: str, selection_manager: SelectionManager, timeout: float = 10, connection_max_tries: int = 5)[source]

A connection to a postgres database for the calibration store.

__init__(logger: Logger, communication_state_callback: Callable[[ska_control_model.CommunicationStatus], None], database_host: str, database_port: int, database_name: str, database_admin_user: str, database_admin_password: str, selection_manager: SelectionManager, timeout: float = 10, connection_max_tries: int = 5) None[source]

Initialise a new instance of a database connection.

Parameters:
  • logger – a logger for this object to use

  • communication_state_callback – callback to be called when the status of the communications channel between the component manager and its component changes

  • database_host – the database host

  • database_port – the database port

  • database_name – the database name

  • database_admin_user – the database admin user

  • database_admin_password – the database admin password

  • selection_manager – The SelectionManager for device should use.

  • timeout – the timeout for database operations

  • connection_max_tries – the maximum number of attempts to connect to the database

get_solution(frequency_channel: int, outside_temperature: float, station_id: int) list[float][source]

Get a solution for the provided frequency and outside temperature.

This at present will return the most recently stored solution for the inputs.

Parameters:
  • frequency_channel – the frequency channel of the desired solution.

  • outside_temperature – the outside temperature of the desired solution.

  • station_id – the id of the station to get a soluion for.

Raises:

RuntimeError – if there are repeated connection issues with the database

Returns:

a calibration solution from the database. Or a empty list if a solution could not be read from artefact.

store_solution(solution_path: str, frequency_channel: int, outside_temperature: float, station_id: int, preferred: bool = False) tuple[list[ska_control_model.ResultCode], list[str]][source]

Store the provided solution in the database.

Parameters:
  • solution_path – the path to the stored solution

  • frequency_channel – the frequency channel that the solution is for

  • outside_temperature – the outside temperature that the solution is for

  • station_id – the id of the station we want to store this solution for.

  • preferred – mark solution as preferred.

Raises:

RuntimeError – if there are repeated connection issues witht the database

Returns:

tuple of result code and message.

verify_database_connection() None[source]

Verify that connection to the database can be established.