Selection Policy

The SelectionManager is the owner of SelectionPolicies and will use them to execute sql queries to the database.

The SelectionManager has a 1-1 relationship with the CalibrationStore, new configurable policies can be injected into the manager from the tango interface using UpdateSelectionPolicy

SelectionManager(logger)	A SelectionManager.
SelectPreferred(logger, *args, **kwargs)	A SelectPreferred policy preferred.

This subpackage implements calibration store functionality for MCCS.

class SelectClosestInRange(logger: Logger, *args: Any, **kwargs: Any)

A SelectClosestInRange policy closest_in_range.

The most recent solution within a defined frequency tolerance.

>>>    "SELECT frequency_channel, "
>>>    "calibration_path, creation_time, station_id, preferred, "
>>>    "solution "
>>>    "FROM calibration_per_channel "
>>>    f"WHERE station_id = {station_id} "
>>>    f"AND frequency_channel >= {frequency_channel-self.frequency_tolerance} "
>>>    f"AND frequency_channel <= {frequency_channel+self.frequency_tolerance} "
>>>    "ORDER BY creation_time DESC "
>>>    "LIMIT 1"

configure(frequency_tolerance: int = 0, **kwargs: Any) → None

Configure the SelectClosestInRange policy.

Parameters:

• frequency_tolerance – the absolute frequency tolerance.

• kwargs – kwargs

generate_sql(station_id: int, frequency_channel: int, **kwargs: Any) → tuple[str, tuple]

Return an sql query with parameterized inputs.

Parameters:

• station_id – The id of the station we want a solution for.

• frequency_channel – The current frequency channel.

• kwargs – kwargs

Returns:

a sql query.

class SelectPreferred(logger: Logger, *args: Any, **kwargs: Any)

A SelectPreferred policy preferred.

The most recent solution within a defined frequency tolerance with the field preferred == TRUE is chosen. If none defined as preferred, no solution is returned.

For solutions linked to a calibration job (job_id IS NOT NULL), the preferred flag on calibration_job is authoritative. For standalone solutions (job_id IS NULL), the preferred flag on calibration_per_channel itself is used as a fallback.

>>>    "SELECT cpc.frequency_channel, "
>>>    "cpc.calibration_path, cpc.creation_time, cpc.station_id, "
>>>    "cpc.preferred, cpc.solution "
>>>    "FROM calibration_per_channel AS cpc "
>>>    "LEFT JOIN calibration_job AS cj ON cj.id = cpc.job_id "
>>>    f"WHERE cpc.station_id = {station_id} "
>>>    f"AND cpc.frequency_channel >= {frequency_channel-self.frequency_tolerance} "
>>>    f"AND cpc.frequency_channel <= {frequency_channel+self.frequency_tolerance} "
>>>    "AND (cj.preferred = TRUE "
>>>    "     OR (cpc.job_id IS NULL AND cpc.preferred = TRUE)) "
>>>    "ORDER BY cpc.creation_time DESC "
>>>    "LIMIT 1"

configure(frequency_tolerance: int = 0, **kwargs: Any) → None

Configure the SelectPreferred policy.

Parameters:

• frequency_tolerance – the absolute frequency tolerance.

• kwargs – kwargs

generate_fit_params_sql(station_id: int) → tuple[str, tuple]

Return a sql query for retrieving per-antenna phase fit parameters.

Parameters:

station_id – The id of the station we want fit params for.

Returns:

a sql query and parameters tuple.

generate_sql(station_id: int, frequency_channel: int, **kwargs: Any) → tuple[str, tuple]

Return an sql query with parameterized inputs.

Parameters:

• station_id – The id of the station we want a solution for.

• frequency_channel – The current frequency channel.

• kwargs – kwargs

Returns:

a sql query.

class SelectionManager(logger: Logger)

A SelectionManager.

__init__(logger: Logger) → None

Initialise a new SelectionManager.

Parameters:

logger – A logger for information purposes.

check_calibration_id(connection: Any, cal_id: str) → bool

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

Parameters:

• connection – the database connection.

• cal_id – the cal ID to check.

Returns:

True if the cal ID is already in the database

get_calibration_ids(connection: Any) → list[str]

Get all calibration_ids.

Parameters:

connection – the database connection.

Returns:

A list of all calibration_ids

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

Get per-antenna phase fit parameters from the database.

Returns the raw phase and phase_gradient arrays so that the caller can reconstruct gains for any set of frequency channels without making repeated round-trips to the store.

Parameters:

• connection – the database connection.

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

• calibration_id – optional unique identifier for the calibration; when supplied the preferred-job selection policy is bypassed.

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.

get_solution(connection: Any, **kwargs: Any) → list[float]

Get a solution using the loaded SelectionPolicy.

This will return a solution from a query set by the loaded SelectionPolicy

Parameters:

• connection – the database connection.

• kwargs – kwargs to feed the selection policy.

Returns:

the solution or an empty list.

Raises:

ValueError – then there is no configured policy to use.

get_solution_from_cal_id(connection: Any, channel_id: int, calibration_id: str) → list[float]

Get a solution from the calibration and station ids.

Parameters:

• connection – the database connection.

• channel_id – The channel id.

• calibration_id – The calibration id.

Returns:

the solution or an empty list.

property has_policy: bool

Return whether policy is defined.

Returns:

True if policy is not None.

property policy: str

Return the policy.

Returns:

the selection policy.

property policy_description: str

Return details about the policy description.

Returns:

information about the selection policy.

update_policy(policy: SelectPreferred | SelectClosestInRange) → None

Set a new policy.

Parameters:

policy – the sql to use.