Calibration Store Device
This module provides a Tango device for a Calibration Store.
- class MccsCalibrationStore(*args: Any, **kwargs: Any)[source]
An implementation of the CalibrationStore Tango device.
- CheckCalibrationId(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
- GetCalibrationIds() list[str][source]
Get all calibration_ids from the database.
- Returns:
All calibration_ids from the database.
- GetFittedSolutions(station_id: int, frequency_channels: list[int], calibration_id: str | None = None) list[float][source]
Return fitted calibration solutions for a set of frequency channels.
Fetches the per-antenna phase fit parameters from the store and reconstructs complex gains for every requested frequency in a single call, avoiding per-channel round-trips from MccsStation.ApplyConfiguration.
The return array has shape
(n_freqs, n_ant, 8)flattened in frequency-major order, where the 8 elements per antenna are[xx_re, xx_im, xy_re, xy_im, yx_re, yx_im, yy_re, yy_im].- Parameters:
station_id – the id of the station to get fit params for
frequency_channels – channel indices to compute solutions for
calibration_id – optional unique identifier for the calibration
- Returns:
flat float array of fitted gains, empty if no params found
- GetLatestPreferredJobID(station_id: int) str[source]
Return the user_friendly_name of the most recent preferred job for a station.
- Parameters:
station_id – the station ID to query for.
- Returns:
the user_friendly_name of the most recent preferred job for the station, or an empty string if no preferred job exists.
- GetSolution(station_id: int, frequency_channel: int, calibration_id: str | None = None) list[float][source]
Get a solution for the provided frequency and station id.
- Parameters:
station_id – the id of the station to get the solution for
frequency_channel – the requested frequency channel
calibration_id – optional unique identifer for the calibration
- Returns:
the solution or an empty list
- MarkCalibrationJobPreferred(user_friendly_name: str) tuple[list[ska_control_model.ResultCode], list[str]][source]
Mark a calibration job as preferred.
The calibration job is identified by its unique
user_friendly_name. Returns FAILED if no job with that name exists in the database.- Parameters:
user_friendly_name – the unique name of the calibration job to mark.
- Returns:
a tuple containing the ResultCode and a message.
- StoreCalibrationJob(sweep_id: int, initial_mask: int | None = None, calibration_parameters: str | None = None, user_friendly_name: str | None = None, preferred: bool | None = None) tuple[list[ska_control_model.ResultCode], list[str]][source]
Store a calibration job in the database.
- Parameters:
sweep_id – the frequency sweep ID this job refers to.
initial_mask – masked antennas passed in initially
calibration_parameters – parameters the calibration job was run with
user_friendly_name – input name of the calibration job
preferred – whether or not this job is to be used as the default
- Returns:
tuple of result code and message
- StoreFrequencySweep(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 – Optional kwargs. * path_to_data: The path where the frequency sweep is stored * start_time: The time at which the sweep was started * nof_samples: The number of samples correlated by the DAQ * initial_static_delays: The static delays used when the sweep was taken
- Returns:
tuple of result code and message
- StorePhaseFitParams(job_id: int, phase: list[float] | None = None, phase_gradient: list[float] | None = None, phase_fit_cost: list[float] | None = None, phase_fit_covariance: list[list[list[float]]] | None = None, universally_masked_antennas: list[int] | None = None) 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.
phase – phase offset per antenna per Jones element
phase_gradient – delay per antenna per Jones element
phase_fit_cost – fit cost per antenna per Jones element
phase_fit_covariance – 2x2 covariance matrices per antenna per Jones element
universally_masked_antennas – TPM-indexed antenna indices that were masked in every observed channel
- Returns:
tuple of result code and message
- StoreSolution(solution: list[float], frequency_channel: int, station_id: int, calibration_path: str, acquisition_time: int, preferred: bool = False, corrcoeff: list[float] | None = None, residual_max: list[float] | None = None, residual_std: list[float] | None = None, xy_phase: 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) tuple[list[ska_control_model.ResultCode], list[str]][source]
Store the provided solution in the database.
- Parameters:
solution – Solution dataset
frequency_channel – Frequency channel to store for
station_id – The id of the station to store for
calibration_path – Path to the correlation matrix file
acquisition_time – Time the data was acquired
preferred – Mark the solution as preferred
corrcoeff – List of correlation coefficients
residual_max – List of residual visibility max abs deviation
residual_std – List of residual visibility std deviation
xy_phase – Estimated xy-phase error
n_masked_initial – Initial number of masked antennas
n_masked_final – Final number of masked antennas
lst – Apparent sidereal time at station
galactic_centre_elevation – Elevation of the galactic centre
sun_elevation – Elevation of the sun (degrees)
sun_adjustment_factor – Sun adjustment factor
masked_antennas – Antennas masked as bad
job_id – Optional reference to a calibration job
solution_type – Optional type of the solution
- Returns:
the solution or an empty list
- UpdateFrequencySweep(sweep_id: int, station_id: int | None = None, channel_start: int | None = None, channel_stop: int | None = None, path_to_data: str | None = None, start_time: str | None = None, nof_samples: int | None = None, initial_static_delays: list[float] | None = None) tuple[list[ska_control_model.ResultCode], list[str]][source]
Update a frequency sweep in the database.
- Parameters:
sweep_id – the sweep to update.
station_id – the id of the station to update
channel_start – start frequency channel
channel_stop – stop frequency channel
path_to_data – path to where the frequency sweep is stored
start_time – time at which the sweep was started
nof_samples – no of samples correlated by the DAQ
initial_static_delays – initial static delays when sweep was taken
- Returns:
tuple of result code and message
- UpdateSelectionPolicy(policy_name: str, frequency_tolerance: int | None = None) tuple[list[ska_control_model.ResultCode], list[str]][source]
Update the selection policy.
- Parameters:
policy_name – name of the policy
frequency_tolerance – absolute tolerance on the frequency
- Returns:
a tuple containing the ResultCode and a message
- __init__(*args: Any, **kwargs: Any) None[source]
Initialise this device object.
- Parameters:
args – positional args to the init
kwargs – keyword args to the init
- create_component_manager() CalibrationStoreComponentManager[source]
Create and return a component manager for this device.
- Returns:
a component manager for this device.
- lastJobID() int[source]
Return the last job ID stored in the database.
- Returns:
the last job ID stored in the database.
- lastSweepID() int[source]
Return the last sweep ID stored in the database.
- Returns:
the last sweep ID stored in the database.