Station calibrator subpackage

This subpackage implements station calibrator functionality for MCCS.

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

An implementation of a station calibrator Tango device for MCCS.

GetCalibration(argin: str) → list[float]

Get a calibration from the calibration store.

Parameters:

argin – json-dictionary of field conditions.

Returns:

a calibration from the calibration store.

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

Class for handling the GetCalibration() 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/MccsStationCalibrator_GetCalibration.json",
    "title": "MccsStationCalibrator GetCalibration schema",
    "description": "Schema for MccsStationCalibrator's GetCalibration command",
    "type": "object",
    "properties": {
        "frequency_channel": {
            "description": "Freqency channel to calibrate for",
            "type": "integer",
            "minimum": 0,
            "maximum": 511
        }
    },
    "required": ["frequency_channel"]
}

__init__(component_manager: StationCalibratorComponentManager, 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 MccsStationCalibrator.GetCalibration() 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 from the calibration store.

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

A class for MccsStationCalibrator’s Init command.

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

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

Initialise the attributes and properties of the MccsStationCalibrator.

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.

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

Start the calibration loop.

Parameters:

argin – json-dictionary of calibration configuration.

Returns:

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

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

Class for handling the StartCalibrationLoop() 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/MccsStationCalibrator_StartCalibrationLoop.json",
    "title": "MccsStationCalibrator StartCalibrationLoop schema",
    "description": "Schema for MccsStationCalibrator's StartCalibrationLoop command",
    "type": "object",
    "properties": {
        "eep_filebase": {
            "description": "Start of EEP filenames before the frequency specification. For X/phi this gives eep_path/eep_filebase<freq>MHz_Xpol_phi.npy",
            "type": [
                "string",
                "null"
            ]
        },
        "jones_solve": {
            "description": "Whether to call the polarised solver after initial calibration.",
            "type": [
                "boolean",
                "null"
            ]
        },
        "adjust_solar_model": {
            "description": "Whether to adjust the sky model after initial calibration to limit differences between galactic and solar flux scales",
            "type": [
                "boolean",
                "null"
            ]
        },
        "masked_antennas": {
            "description": "List of antennas which should not be used for calibration",
            "type": [
                "array",
                "null"
            ],
            "items": {
                "type": "number"
            }
        },
        "skymodel": {
            "description": "Either a model visibility xarray to calibrate against or the name of a sky model type to generate. Supported sky models are `gsm` and `sun`. The `gsm` model combines components of the PyGDSM GlobalSkyModel16 Galactic sky model that are above the horizon with a point-source solar model, while sun uses only the point-source solar model.",
            "type": [
                "string",
                "null"
            ]
        },
        "min_uv": {
            "description": "Minimum baseline length used in calibration (in metres). The default value is zero.",
            "type": [
                "number",
                "null"
            ]
        },
        "refant": {
            "description": "Antenna to use for phase referencing. Should be one of the zero-based indices used in the visibility dataset. Default is 1. Both polarisations are phase referenced against the X polarisation of refant, as a non-zero XY-phase is expected and solved for.",
            "type": [
                "integer",
                "null"
            ],
            "minimum": 0
        },
        "ignore_eeps": {
            "description": "If True, EEP files will not be imported and station beam patterns and rotations will be ignored when modelling visibilities. Sky model components will be added with equal amplitude in XX and YY and zeros amplitude in XY and YX. Default is False.",
            "type": [
                "integer",
                "null"
            ]
        },
        "gain_threshold": {
            "description": "Fractional threshold used to reject antennas with low gains. Any antenna with either an X or Y gain amplitude less than the median is flagged as bad. The default values is 0.25.",
            "type": [
                "number",
                "null"
            ]
        },
        "station_config_path": {
            "type": [
                "array"
            ],
            "items": {
                "type": "string",
                "pattern": "^[^\\s]+$"
            },
            "minItems": 2,
            "maxItems": 2,
            "uniqueItems": true
        },
        "nside": {
            "description": "Healpix nside at which to generate sky model. Default of 32 corresponds to approximately 110 arcminutes. This is okay at low frequencies but may need to be increased at higher frequencies.",
            "type": [
                "integer",
                "null"
            ],
            "default": 32,
            "minimum": 1
        },
        "niter": {
            "description": "Number of iterations for solver.",
            "type": [
                "integer",
                "null"
            ],
            "default": 200,
            "minimum": 1
        },
        "first_channel": {
            "description": "The lowest channel in the loop.",
            "type": [
                "integer",
                "null"
            ],
            "minimum": 0,
            "maximum": 511
        },
        "last_channel": {
            "description": "The final channel to use in loop.",
            "type": [
                "integer",
                "null"
            ],
            "minimum": 0,
            "maximum": 511
        }
    },
    "required": [
        "station_config_path"
    ]
}

__init__(component_manager: StationCalibratorComponentManager, 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[Optional[str]]]

Implement MccsStationCalibrator.StartCalibrationLoop() 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.

StopCalibrationLoop() → tuple[list[ska_control_model.ResultCode], list[Optional[str]]]

Stop the calibration loop.

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() → StationCalibratorComponentManager

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.

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

A component manager for a station calibrator.

__init__(logger: Logger, station_name: str, station_id: int, calibration_store_name: str, calibration_solver_trl: str, communication_state_callback: Callable[[ska_control_model.CommunicationStatus], None], component_state_callback: Callable[[...], None]) → None

Initialise a new instance.

Parameters:

• logger – the logger to be used by this object.

• station_name – the name of this calibrator’s station

• station_id – the id of the station.

• calibration_store_name – the name of this calibrator’s calibration store

• calibration_solver_trl – the trl of the solver device.

• communication_state_callback – callback to be called when the status of the communications channel between the component manager and its component changes.

• component_state_callback – callback to be called when the component state changes.

get_calibration(channel: int) → list[float]

Get a calibration from the calibration store.

Parameters:

channel – the frequency channel to calibrate for

Raises:

ValueError – if the outside temperature has not been read yet

Returns:

A list of calibration values

start_calibration_loop(kwargs: dict[str, Any]) → tuple[list[ska_control_model.ResultCode], list[str | None]]

Start the calibration loop.

Parameters:

kwargs – kwargs to define calibration.

Returns:

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

start_communicating() → None

Establish communication with the StationCalibrator components.

stop_calibration_loop() → tuple[list[ska_control_model.ResultCode], list[str | None]]

Stop the calibration loop from running.

Returns:

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

stop_communicating() → None

Break off communication with the StationCalibrator components.

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

A health model for a station calibrator.

At present this uses the base health model; this is a placeholder for a future, better implementation.

• Station calibrator component manager
  • StationCalibratorComponentManager
    • StationCalibratorComponentManager.__init__()
    • StationCalibratorComponentManager.get_calibration()
    • StationCalibratorComponentManager.start_calibration_loop()
    • StationCalibratorComponentManager.start_communicating()
    • StationCalibratorComponentManager.stop_calibration_loop()
    • StationCalibratorComponentManager.stop_communicating()
• Station calibrator device
  • MccsStationCalibrator
    • MccsStationCalibrator.GetCalibration()
    • MccsStationCalibrator.GetCalibrationCommand
      • MccsStationCalibrator.GetCalibrationCommand.__init__()
      • MccsStationCalibrator.GetCalibrationCommand.do()
    • MccsStationCalibrator.InitCommand
      • MccsStationCalibrator.InitCommand.do()
    • MccsStationCalibrator.StartCalibrationLoop()
    • MccsStationCalibrator.StartCalibrationLoopCommand
      • MccsStationCalibrator.StartCalibrationLoopCommand.__init__()
      • MccsStationCalibrator.StartCalibrationLoopCommand.do()
    • MccsStationCalibrator.StopCalibrationLoop()
    • MccsStationCalibrator.__init__()
    • MccsStationCalibrator.create_component_manager()
    • MccsStationCalibrator.init_command_objects()
  • main()
• Station calibrator health model
  • StationCalibratorHealthModel