Calibration Store Device

This module provides a Tango device for a Calibration Store.

class MccsCalibrationStore(*args: Any, **kwargs: Any)

An implementation of the CalibrationStore Tango device.

CheckCalibrationId(cal_id: str) → bool

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]

Get all calibration_ids from the database.

Returns:

All calibration_ids from the database.

GetSolution(argin: str) → list[float]

Get a calibration solution from the database.

Parameters:

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

Returns:

a calibration solution from the database.

class GetSolutionCommand(*args: Any, **kwargs: Any)

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
        },
        "station_id": {
            "description": "the id of the station we want a solution for.",
            "type": "number"
        },
        "calibration_id": {
            "description": "the unique identifer for the calibration.",
            "type": "string"
        }
    },
    "required": [
        "station_id"
    ]
}

__init__(component_manager: CalibrationStoreComponentManager, logger: Logger | None = None) → None

Initialise a new instance.

Parameters:

• component_manager – the device’s component manager

• logger – a logger for this command to log with.

do(*args: Any, **kwargs: Any) → list[float]

Implement MccsCalibrationStore.GetSolution() command.

Parameters:

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

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

Returns:

a calibration solution from the database.

class InitCommand(*args: Any, **kwargs: Any)

A class for MccsCalibrationStore’s Init command.

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

do(*args: Any, **kwargs: Any) → tuple[ska_control_model.ResultCode, str]

Initialise the attributes and properties of the MccsCalibrationStore.

Parameters:

• args – positional args to the component manager method

• kwargs – keyword args to the component manager method

Returns:

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

StoreCalibrationJob(argin: str) → tuple[list[ska_control_model.ResultCode], list[str]]

Store a frequency sweep in the database.

Parameters:

argin – json-dictionary of sweep and metadata.

Returns:

tuple of result code and message.

class StoreCalibrationJobCommand(*args: Any, **kwargs: Any)

Class for handling the StoreCalibrationJob() 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_StoreCalibrationJob.json",
    "title": "MccsCalibrationStore StoreCalibrationJob schema",
    "description": "Schema for MccsCalibrationStore's StoreCalibrationJob command",
    "type": "object",
    "properties": {
        "sweep_id": {
            "description": "The id of the sweep to store a job for",
            "type": "integer"
        },
        "initial_mask": {
            "description": "Masked antennas passed in initially",
            "items": {
                "type": "integer"
            },
            "maxItems": 256
        },
        "calibration_parameters": {
            "description": "The parameters the calibration job was run with",
            "type": "string"
        },
        "user_friendly_name": {
            "description": "The input name of the calibration job",
            "type": "string"
        },
        "preferred": {
            "description": "Whether or not this job is to be used as the default.",
            "type": "boolean"
        }
    },
    "required": [
        "sweep_id"
    ]
}

__init__(component_manager: CalibrationStoreComponentManager, logger: Logger | None = None) → None

Initialise a new instance.

Parameters:

• component_manager – the device’s component manager

• logger – a logger for this command to log with.

do(*args: Any, **kwargs: Any) → tuple[list[ska_control_model.ResultCode], list[str]]

Implement MccsCalibrationStore.StoreCalibrationJob() command.

Parameters:

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

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

Returns:

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

StoreFrequencySweep(argin: str) → tuple[list[ska_control_model.ResultCode], list[str]]

Store a frequency sweep in the database.

Parameters:

argin – json-dictionary of sweep and metadata.

Returns:

tuple of result code and message.

class StoreFrequencySweepCommand(*args: Any, **kwargs: Any)

Class for handling the StoreFrequencySweep() 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_StoreFrequencySweep.json",
    "title": "MccsCalibrationStore StoreFrequencySweep schema",
    "description": "Schema for MccsCalibrationStore's StoreFrequencySweep command",
    "type": "object",
    "properties": {
        "station_id": {
            "description": "The id of the station to store a solution for",
            "type": "integer",
            "minimum": 1,
            "maximum": 512
        },
        "channel_start": {
            "description": "Start frequency channel of the sweep",
            "type": "integer",
            "minimum": 0,
            "maximum": 511,
            "units": "channel"
        },
        "channel_stop": {
            "description": "Stop frequency channel of the sweep",
            "type": "integer",
            "minimum": 0,
            "maximum": 511,
            "units": "channel"
        },
        "path_to_data": {
            "description": "The path where the frequency sweep is stored",
            "type": "string"
        },
        "start_time": {
            "description": "The time at which the sweep was started",
            "type": "string",
            "units": "ISO 8601 timestamp"
        },
        "nof_samples": {
            "description": "The number of samples correlated by the DAQ",
            "type": "integer"
        },
        "initial_static_delays": {
            "description": "The initial static delays used when the sweep was taken, in TPM order.",
            "items": {
                "type": "number"
            },
            "minItems": 32,
            "maxItems": 512
        }
    },
    "required": [
        "station_id",
        "channel_start",
        "channel_stop"
    ]
}

__init__(component_manager: CalibrationStoreComponentManager, logger: Logger | None = None) → None

Initialise a new instance.

Parameters:

• component_manager – the device’s component manager

• logger – a logger for this command to log with.

do(*args: Any, **kwargs: Any) → tuple[list[ska_control_model.ResultCode], list[str]]

Implement MccsCalibrationStore.StoreFrequencySweep() command.

Parameters:

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

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

Returns:

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

StoreSolution(argin: str) → tuple[list[ska_control_model.ResultCode], list[str]]

Store a solution in the database.

Parameters:

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

Returns:

tuple of result code and message.

class StoreSolutionCommand(*args: Any, **kwargs: Any)

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,
            "units": "channel"
        },
        "station_id": {
            "description": "The id of the station to store a solution for",
            "type": "integer",
            "minimum": 1,
            "maximum": 512
        },
        "preferred": {
            "description": "Mark a solution as preferred.",
            "type": "boolean"
        },
        "solution": {
            "description": "The solution dataset. This is ordered by tpm_idx (i.e 8 values for each tpm_id * 16 + (adc_channel // 2 )).",
            "type": "array",
            "items": {
                "type": "number"
            }
        },
        "corrcoeff": {
            "description": "List of Pearson product-moment correlation coefficients for each polarisation of the calibrated visibilities and the model.",
            "type": "array",
            "items": {
                "type": "number"
            }
        },
        "residual_max": {
            "description": "List of residual visibility maximum absolution deviation per polarisation (K).",
            "type": "array",
            "items": {
                "type": "number"
            },
            "minItems": 4,
            "maxItems": 4
        },
        "residual_std": {
            "description": "List of residual visibility standard deviation per polarisation (K).",
            "type": "array",
            "items": {
                "type": "number"
            },
            "minItems": 4,
            "maxItems": 4
        },
        "xy_phase": {
            "description": "Estimated xy-phase error.",
            "type": "number",
            "units": "Degrees"
        },
        "n_masked_initial": {
            "description": "Initial number of masked antennas.",
            "type": "integer"
        },
        "n_masked_final": {
            "description": "Final number of masked antennas.",
            "type": "integer"
        },
        "lst": {
            "description": "Apparent sidereal time at station.",
            "type": "number",
            "units": "Degrees"
        },
        "galactic_centre_elevation": {
            "description": " Elevation of the galactic centre.",
            "type": "number",
            "units": "Degrees"
        },
        "sun_elevation": {
            "description": "Elevation of the sun (degrees).",
            "type": "number",
            "units": "Degrees"
        },
        "sun_adjustment_factor": {
            "type": "number"
        },
        "masked_antennas": {
            "type": "array",
            "description": "Antenna masked as bad. This is currently eep indexed.",
            "items": {
                "type": "integer"
            },
            "minItems": 0,
            "maxItems": 256
        },
        "acquisition_time": {
            "description": "Time the data was acquired",
            "type": "number",
            "units": "Time since the Unix epoch"
        },
        "calibration_path": {
            "description": "The path to the correlation matrix file this solution was determined from.",
            "type": "string"
        },
        "calibration_id": {
            "description": "A unique identifier for calibrations to quickly assign them.",
            "type": "string"
        }
    },
    "required": [
        "frequency_channel",
        "station_id",
        "calibration_path",
        "acquisition_time",
        "solution"
    ]
}

__init__(component_manager: CalibrationStoreComponentManager, logger: Logger | None = None) → None

Initialise a new instance.

Parameters:

• component_manager – the device’s component manager

• logger – a logger for this command to log with.

do(*args: Any, **kwargs: Any) → tuple[list[ska_control_model.ResultCode], list[str]]

Implement MccsCalibrationStore.StoreSolution() command.

Parameters:

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

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

Returns:

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

UpdateFrequencySweep(argin: str) → tuple[list[ska_control_model.ResultCode], list[str]]

Update a frequency sweep in the database.

Parameters:

argin – json-dictionary of sweep and metadata.

Returns:

tuple of result code and message.

class UpdateFrequencySweepCommand(*args: Any, **kwargs: Any)

Class for handling the UpdateFrequencySweep() 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_StoreFrequencySweep.json",
    "title": "MccsCalibrationStore StoreFrequencySweep schema",
    "description": "Schema for MccsCalibrationStore's StoreFrequencySweep command",
    "type": "object",
    "properties": {
        "sweep_id": {
            "description": "The id of the sweep to update",
            "type": "integer"
        },
        "station_id": {
            "description": "The id of the station to store a solution for",
            "type": "integer",
            "minimum": 1,
            "maximum": 512
        },
        "channel_start": {
            "description": "Start frequency channel of the sweep",
            "type": "integer",
            "minimum": 0,
            "maximum": 511,
            "units": "channel"
        },
        "channel_stop": {
            "description": "Stop frequency channel of the sweep",
            "type": "integer",
            "minimum": 0,
            "maximum": 511,
            "units": "channel"
        },
        "path_to_data": {
            "description": "The path where the frequency sweep is stored",
            "type": "string"
        },
        "start_time": {
            "description": "The time at which the sweep was started",
            "type": "string",
            "units": "ISO 8601 timestamp"
        },
        "nof_samples": {
            "description": "The number of samples correlated by the DAQ",
            "type": "integer"
        },
        "initial_static_delays": {
            "description": "The initial static delays used when the sweep was taken, in TPM order.",
            "items": {
                "type": "number"
            },
            "minItems": 32,
            "maxItems": 512
        }
    },
    "required": [
        "sweep_id"
    ]
}

__init__(component_manager: CalibrationStoreComponentManager, logger: Logger | None = None) → None

Initialise a new instance.

Parameters:

• component_manager – the device’s component manager

• logger – a logger for this command to log with.

do(*args: Any, **kwargs: Any) → tuple[list[ska_control_model.ResultCode], list[str]]

Implement MccsCalibrationStore.UpdateFrequencySweep() command.

Parameters:

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

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

Returns:

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

UpdateSelectionPolicy(argin: str) → tuple[list[ska_control_model.ResultCode], list[str]]

Update the selection policy.

Parameters:

argin –

A validated json dictionary containing the following mandatory keys:

• temperature_tolerance (float)

• frequency_tolerance (int)

Returns:

tuple of result code and message.

class UpdateSelectionPolicyCommand(*args: Any, **kwargs: Any)

Class for handling the UpdateSelectionPolicy() 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_UpdateSelectionPolicy.json",
    "title": "MccsCalibrationStore UpdateSelectionPolicy schema",
    "description": "Schema for MccsCalibrationStore's UpdateSelectionPolicy command",
    "type": "object",
    "properties": {
        "policy_name": {
            "description": "The name of the policy. Please see details in ReadTheDocs https://developer.skao.int/projects/ska-low-mccs/en/latest/api/calibration_store/selection_policy/index.html",
            "type": "string",
            "enum": [
                "preferred",
                "closest_in_range"
            ]
        },
        "frequency_tolerance": {
            "description": "The absolute tolerance on the frequency",
            "type": "integer",
            "units": "channelizer channel"
        }
    },
    "required": [
        "policy_name"
    ]
}

__init__(component_manager: CalibrationStoreComponentManager, logger: Logger | None = None) → None

Initialise a new instance.

Parameters:

• component_manager – the device’s component manager

• logger – a logger for this command to log with.

do(*args: Any, **kwargs: Any) → tuple[list[ska_control_model.ResultCode], list[str]]

Implement MccsCalibrationStore.UpdateSelectionPolicy() command.

Parameters:

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

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

Returns:

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

__init__(*args: Any, **kwargs: Any) → None

Initialise this device object.

Parameters:

• args – positional args to the init

• kwargs – keyword args to the init

create_component_manager() → CalibrationStoreComponentManager

Create and return a component manager for this device.

Returns:

a component manager for this device.

init_command_objects() → None

Initialise the command handlers for this device.

init_device() → None

Initialise the device.

lastJobID() → int

Return the last job ID stored in the database.

Returns:

the last job ID stored in the database.

lastSweepID() → int

Return the last sweep ID stored in the database.

Returns:

the last sweep ID stored in the database.

selectionPolicy() → str

Read the selectionPolicy used by this device.

Returns:

a JSON serialised dictionary with information in a format validated by UpdateSelectionPolicy.json.

selectionPolicyDescription() → str

Read the full selectionPolicy description in use.

Returns:

a string with information about the policy.

main(*args: str, **kwargs: str) → int

Entry point for module.

Parameters:

• args – positional arguments

• kwargs – named arguments

Returns:

exit code