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], policy_name: str, policy_frequency_tolerance: int, policy_temperature_tolerance: float, window_failure_size: float, window_failure_threshold: int, circuit_open_timeout: float) → 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

  • policy_name – The policy type to use. Currently [“preferred”, “closest_in_range”] supported.

  • policy_frequency_tolerance – a configuration parameter that may be used by policy [channel]

  • policy_temperature_tolerance – a configuration parameter that may be used by policy [Degrees]

  • window_failure_size – The time in seconds to evaluate failures in.

  • window_failure_threshold – The number of failures within a window before we class is as a database failure

  • circuit_open_timeout – A time to block db connections after a database failure if detected.

Raises:

ValueError – When the SelectionPolicy configuration is incorrect from deployment.

check_calibration_id(cal_id: str) → bool[source]

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

Parameters:

cal_id – the cal ID to check.

Returns:

True if the cal ID is already in the database

Raises:

RuntimeError – If the database connection has not been established.

cleanup() → None[source]

Cleanup database connections.

get_calibration_ids() → list[str][source]

Get all calibration_ids.

Returns:

A list of all calibration_ids

Raises:

RuntimeError – If the database connection has not been established.

get_fit_params(station_id: int, calibration_id: str | None = None) → tuple[list[float], list[float], list[bool] | None] | None[source]

Get per-antenna phase fit parameters for the given station.

Parameters:

  • station_id – the id of the station to get fit params for.

  • calibration_id – optional unique identifier for the calibration.

Returns:

tuple of (phase, phase_gradient, final_mask), or None when no matching row is found.

Raises:

  • ValueError – when no selection policy is configured and calibration_id is not provided.

  • RuntimeError – If the database connection has not been established.

get_latest_preferred_job_id(station_id: int) → str[source]

Return the user_friendly_name of the most recent preferred job for a station.

Parameters:

station_id – the station to query for.

Returns:

the user_friendly_name of the most recent preferred job, or an empty string if none exists.

Raises:

RuntimeError – If the database connection has not been established.

get_policy() → str[source]

Return the active SelectionPolicy.

Returns:

a JSON servialised dictionary with information in a format validated.

get_policy_description() → 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 station id.

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

Raises:

  • ValueError – when no SelectionPolicy is loaded.

  • RuntimeError – when database connection not established.

mark_calibration_job_preferred(user_friendly_name: str) → tuple[list[ska_control_model.ResultCode], list[str]][source]

Mark a calibration job as preferred.

Parameters:

user_friendly_name – the unique name of the calibration job to mark.

Returns:

tuple of result code and message

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 with database.

stop_communicating() → None[source]

Break off communication to the database.

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

Store the provided calibration job in the database.

Parameters:

  • sweep_id – the frequency sweep ID this job refers to.

  • kwargs – any optional validated kwargs.

Returns:

tuple of result code and message

store_frequency_sweep(station_id: int, channel_start: int, channel_stop: int, **kwargs: Any) → tuple[list[ska_control_model.ResultCode], list[str]][source]

Store the provided frequency sweep in the database.

Parameters:

  • station_id – the id of the station to store this frequency sweep for.

  • channel_start – the first channel in the sweep.

  • channel_stop – the last channel in the sweep.

  • kwargs – any optional validated kwargs.

Returns:

tuple of result code and message

store_phase_fit_params(job_id: int, **kwargs: Any) → tuple[list[ska_control_model.ResultCode], list[str]][source]

Store per-antenna phase fit parameters in the database.

Parameters:

  • job_id – the calibration job ID these fit params belong to.

  • kwargs – any optional validated kwargs.

Returns:

tuple of result code and message

store_solution(solution: list[float], frequency_channel: int, station_id: int, calibration_path: str, acquisition_time: int, preferred: bool = False, **kwargs: Any) → tuple[list[ska_control_model.ResultCode], list[str]][source]

Store the provided solution in the database.

Parameters:

  • solution – the solution to store.

  • frequency_channel – the frequency channel that the solution is for

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

  • preferred – mark solution as preferred.

  • calibration_path – the path to the correlation matrix used to get this solution.

  • acquisition_time – the time the data was acquired

  • kwargs – any optional validated kwargs.

Returns:

tuple of result code and message

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

Update the provided frequency sweep in the database.

Parameters:

  • sweep_id – the sweep up update.

  • kwargs – any optional validated kwargs.

Returns:

tuple of result code and message