Calibration store subpackage

This subpackage implements calibration store functionality for MCCS.

class CalibrationStoreComponentManager(logger, communication_state_changed_callback, component_state_changed_callback, database_host, database_port, database_name, database_admin_user, database_admin_password)

A component manager for MccsCalibrationStore.

__init__(logger, communication_state_changed_callback, component_state_changed_callback, database_host, database_port, database_name, database_admin_user, database_admin_password)

Initialise a new instance.

Parameters:

  • logger (Logger) – a logger for this object to use

  • component_state_changed_callback (Callable[[Optional[bool], Optional[PowerState]], None]) – callback to be called when the component state changes

  • communication_state_changed_callback (Callable[[CommunicationStatus], None]) – callback to be called when the status of the communications channel between the component manager and its component changes

  • database_host (str) – the database host

  • database_port (int) – the database port

  • database_name (str) – the database name

  • database_admin_user (str) – the database admin user

  • database_admin_password (str) – the database admin password

create_database_connection(logger, communication_state_changed_callback, database_host, database_port, database_name, database_admin_user, database_admin_password)

Create the database connection object.

Parameters:

  • logger (Logger) – a logger for the connection to use

  • communication_state_changed_callback (Callable[[CommunicationStatus], None]) – callback to be called when the component state changes

  • database_host (str) – the database host

  • database_port (int) – the database port

  • database_name (str) – the database name

  • database_admin_user (str) – the database admin user

  • database_admin_password (str) – the database admin password

Return type:

CalibrationStoreDatabaseConnection

Returns:

the database connection

get_solution(frequency_channel, outside_temperature)

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 (int) – the frequency channel of the desired solution.

  • outside_temperature (float) – the outside temperature of the desired solution.

Return type:

list[float]

Returns:

a calibration solution from the database.

start_communicating()

Establish communication.

Return type:

None

stop_communicating()

Break off communication.

Return type:

None

store_solution(solution, frequency_channel, outside_temperature)

Store the provided solution in the database.

Parameters:

  • solution (list[float]) – the solution to store

  • frequency_channel (int) – the frequency channel that the solution is for

  • outside_temperature (float) – the outside temperature that the solution is for

Return type:

tuple[list[ResultCode], list[str]]

Returns:

tuple of result code and message

class CalibrationStoreDatabaseConnection(logger, communication_state_callback, database_host, database_port, database_name, database_admin_user, database_admin_password, timeout=10, connection_max_tries=5)

A connection to a postgres database for the calibration store.

__init__(logger, communication_state_callback, database_host, database_port, database_name, database_admin_user, database_admin_password, timeout=10, connection_max_tries=5)

Initialise a new instance of a database connection.

Parameters:

  • logger (Logger) – a logger for this object to use

  • communication_state_callback (Callable[[CommunicationStatus], None]) – callback to be called when the status of the communications channel between the component manager and its component changes

  • database_host (str) – the database host

  • database_port (int) – the database port

  • database_name (str) – the database name

  • database_admin_user (str) – the database admin user

  • database_admin_password (str) – the database admin password

  • timeout (float) – the timeout for database operations

  • connection_max_tries (int) – the maximum number of attempts to connect to the database

get_solution(frequency_channel, outside_temperature)

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 (int) – the frequency channel of the desired solution.

  • outside_temperature (float) – the outside temperature of the desired solution.

Raises:

RuntimeError – if there are repeated connection issues with the database

Return type:

list[float]

Returns:

a calibration solution from the database.

store_solution(solution, frequency_channel, outside_temperature)

Store the provided solution in the database.

Parameters:

  • solution (list[float]) – the solution to store

  • frequency_channel (int) – the frequency channel that the solution is for

  • outside_temperature (float) – the outside temperature that the solution is for

Raises:

RuntimeError – if there are repeated connection issues witht the database

Return type:

tuple[list[ResultCode], list[str]]

Returns:

tuple of result code and message.

verify_database_connection()

Verify that connection to the database can be established.

Return type:

None

class CalibrationStoreHealthModel(*args, **kwargs)

A health model for a calibration store.

At present this uses the base health model; this is a placeholder for a future, better implementation.

class MccsCalibrationStore(*args, **kwargs)

An implementation of the CalibrationStore Tango device.

GetSolution(argin)

Get a calibration solution from the database.

Parameters:

argin (str) – json-dictionary of field conditions and channel data.

Return type:

list[float]

Returns:

a calibration solution from the database.

class GetSolutionCommand(component_manager, logger=None)

Class for handling the GetSolution() command.

This command takes as input a JSON string that conforms to the following schema:

{
    "$schema": "https://json-schema.org/draft/2020-12/schema",
    "$id": "https://skao.int/MccsCalibrationStore_GetSolution.json",
    "title": "MccsCalibrationStore GetSolution schema",
    "description": "Schema for MccsCalibrationStore's GetSolution command",
    "type": "object",
    "properties": {
        "frequency_channel": {
            "description": "Freqency channel to calibrate for",
            "type": "integer",
            "minimum": 0,
            "maximum": 511
        },
        "outside_temperature": {
            "description": "Outside temperature to calibrate for",
            "type": "number"
        }
    },
    "required": ["frequency_channel", "outside_temperature"]
}
__init__(component_manager, logger=None)

Initialise a new instance.

Parameters:
do(*args, **kwargs)

Implement MccsCalibrationStore.GetSolution() command.

Parameters:

  • args (Any) – Positional arguments. This should be empty and is provided for type hinting purposes only.

  • kwargs (Any) – keyword arguments unpacked from the JSON argument to the command.

Return type:

list[float]

Returns:

a calibration solution from the database.

class InitCommand(*args, **kwargs)

A class for MccsCalibrationStore’s Init command.

The do() method below is called upon MccsCalibrationStore’s initialisation.

do(*args, **kwargs)

Initialise the attributes and properties of the MccsCalibrationStore.

Parameters:

  • args (Any) – positional args to the component manager method

  • kwargs (Any) – keyword args to the component manager method

Return type:

tuple[ResultCode, str]

Returns:

A tuple containing a return code and a string message indicating status. The message is for information purpose only.

StoreSolution(argin)

Store a solution in the database.

Parameters:

argin (str) – json-dictionary of solution, field conditions and channel data.

Return type:

tuple[list[ResultCode], list[str]]

Returns:

tuple of result code and message.

class StoreSolutionCommand(component_manager, logger=None)

Class for handling the StoreSolution() command.

This command takes as input a JSON string that conforms to the following schema:

{
    "$schema": "https://json-schema.org/draft/2020-12/schema",
    "$id": "https://skao.int/MccsCalibrationStore_StoreSolution.json",
    "title": "MccsCalibrationStore StoreSolution schema",
    "description": "Schema for MccsCalibrationStore's StoreSolution command",
    "type": "object",
    "properties": {
        "frequency_channel": {
            "description": "Freqency channel to store a solution for",
            "type": "integer",
            "minimum": 0,
            "maximum": 511
        },
        "outside_temperature": {
            "description": "Outside temperature to store a solution for",
            "type": "number"
        },
        "solution": {
            "description": "The calibration solution to store",
            "type": "array",
            "items": {
                "type": "number"
            }
        }
    },
    "required": ["frequency_channel", "outside_temperature"]
}
__init__(component_manager, logger=None)

Initialise a new instance.

Parameters:
do(*args, **kwargs)

Implement MccsCalibrationStore.StoreSolution() command.

Parameters:

  • args (Any) – Positional arguments. This should be empty and is provided for type hinting purposes only.

  • kwargs (Any) – keyword arguments unpacked from the JSON argument to the command.

Return type:

tuple[list[ResultCode], list[str]]

Returns:

A tuple containing a return code and a string message indicating status. The message is for information purpose only.

__init__(*args, **kwargs)

Initialise this device object.

Parameters:

  • args (Any) – positional args to the init

  • kwargs (Any) – keyword args to the init

create_component_manager()

Create and return a component manager for this device.

Return type:

CalibrationStoreComponentManager

Returns:

a component manager for this device.

init_command_objects()

Initialise the command handlers for this device.

Return type:

None