Calibration Store Component Manager

This module implements the component management for a calibration store.

class CalibrationStoreComponentManager(*args: Any, **kwargs: Any)[source]

A component manager for MccsCalibrationStore.

__init__(logger: Logger, communication_state_changed_callback: Callable[[ska_control_model.CommunicationStatus], None], component_state_changed_callback: Callable[[bool | None, ska_control_model.PowerState | None], None], database_host: str, database_port: int, database_name: str, database_admin_user: str, database_admin_password: str) None[source]

Initialise a new instance.

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

  • component_state_changed_callback – callback to be called when the component state changes

  • communication_state_changed_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

create_database_connection(logger: Logger, communication_state_changed_callback: Callable[[ska_control_model.CommunicationStatus], None], database_host: str, database_port: int, database_name: str, database_admin_user: str, database_admin_password: str) CalibrationStoreDatabaseConnection[source]

Create the database connection object.

Parameters:
  • logger – a logger for the connection to use

  • communication_state_changed_callback – callback to be called when the component state 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

Returns:

the database connection

get_policy() str[source]

Return a description of the SelectionPolicy active.

Returns:

a string with information about the policy.

get_solution(**kwargs: Any) 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:

kwargs – kwargs to feed policy.

Returns:

the solution or an empty list

set_selection_policy(policy_name: str, **kwargs: Any) tuple[list[ska_control_model.ResultCode], list[str]][source]

Set a new policy for selecting solutions.

Parameters:
  • policy_name – The policy to use.

  • kwargs – kwargs to pass to policy.

kwargs are used to configure the SelectionPolicy, See details on the configure command for selected policy for details: https://developer.skao.int/projects/ska-low-mccs/en/latest/api/calibration_store/selection_policy/index.html

Returns:

a tuple containing the ResultCode and a message.

start_communicating() None[source]

Establish communication.

stop_communicating() None[source]

Break off communication.

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 – 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 to store this solution for.

  • preferred – mark solution as preferred.

Returns:

tuple of result code and message