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], selection_manager: SelectionManager, timeout: float = 10, connection_max_tries: int = 5)

A connection to a postgres database for the calibration store.

__init__(logger: Logger, communication_state_callback: Callable[[ska_control_model.CommunicationStatus], None], selection_manager: SelectionManager, timeout: float = 10, connection_max_tries: int = 5) → None

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

• 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

check_calibration_id(cal_id: str) → bool

Check if the given cal id is in the database already.

Parameters:

cal_id – the cal ID to check.

Raises:

RuntimeError – if there are repeated connection issues with the database

Returns:

True if the cal ID is already in the database

get_calibration_ids() → list[str]

Get all calibration_ids.

Raises:

RuntimeError – if there are repeated connection issues with the database

Returns:

A list of all calibration_ids

get_solution(station_id: int, frequency_channel: int, calibration_id: str | None = None) → list[float]

Get a solution for the provided frequency and station id.

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

Parameters:

• frequency_channel – the frequency channel of the desired solution.

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

• calibration_id – the optional id of the calibration we want.

Raises:

• RuntimeError – if there are repeated connection issues with the database

• Exception – If we cant make a connection

Returns:

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

store_calibration_job(**kwargs: Any) → tuple[ska_control_model.ResultCode, int]

Store the provided calibration job in the database.

Parameters:

kwargs – fields to populate in database.

Raises:

RuntimeError – if there are repeated connection issues with the database

Returns:

tuple of result code and message.

store_frequency_sweep(**kwargs: Any) → tuple[ska_control_model.ResultCode, int]

Store the provided frequency sweep in the database.

Parameters:

kwargs – fields to populate in database.

Raises:

RuntimeError – if there are repeated connection issues with the database

Returns:

tuple of result code and message.

store_solution(**kwargs: Any) → tuple[list[ska_control_model.ResultCode], list[str]]

Store the provided solution in the database.

Parameters:

kwargs – fields to populate in database.

Raises:

• RuntimeError – if there are repeated connection issues with the database

• Exception – If we cant generate loading instructions

Returns:

tuple of result code and message.

update_frequency_sweep(**kwargs: Any) → tuple[list[ska_control_model.ResultCode], list[str]]

Update the provided frequency sweep in the database.

Parameters:

kwargs – fields to update in database.

Raises:

RuntimeError – if there are repeated connection issues with the database

Returns:

tuple of result code and message.

verify_database_connection() → None

Verify that connection to the database can be established.

class DatabaseSolution(acquisition_time: int, frequency_channel: int, station_id: int, preferred: bool, solution: list[float], calibration_path: str, corrcoeff: list[float] | None = None, residual_max: list[float] | None = None, residual_std: list[float] | None = None, xy_phase: list[float] | None = None, n_masked_initial: int | None = None, n_masked_final: int | None = None, lst: float | None = None, galactic_centre_elevation: float | None = None, sun_elevation: float | None = None, sun_adjustment_factor: float | None = None, masked_antennas: list[int] | None = None, job_id: int | None = None, solution_type: str | None = None)

Class to hold a solution.

This class holds a solution and offers some notion of checking types and mandatory keys before attempting to load into the calibration_per_channel table.

__init__(acquisition_time: int, frequency_channel: int, station_id: int, preferred: bool, solution: list[float], calibration_path: str, corrcoeff: list[float] | None = None, residual_max: list[float] | None = None, residual_std: list[float] | None = None, xy_phase: list[float] | None = None, n_masked_initial: int | None = None, n_masked_final: int | None = None, lst: float | None = None, galactic_centre_elevation: float | None = None, sun_elevation: float | None = None, sun_adjustment_factor: float | None = None, masked_antennas: list[int] | None = None, job_id: int | None = None, solution_type: str | None = None) → None

generate_loading_instruction() → tuple[psycopg.sql.Composed, tuple]

Generate instruction needed to load into calibration_per_channel.

Returns:

Templated information to load into database. templating here will improve security against sql injection attacks.