Station calibrator subpackage

This subpackage implements station calibrator functionality for MCCS.

class CalibrationScheduleState(value)[source]: Operational state of the calibration scheduler.

class MccsStationCalibrator(*args: Any, **kwargs: Any)[source]

An implementation of a station calibrator Tango device for MCCS.

GenerateCalibrationSolution(**kwargs: Any) → ska_tango_base.type_hints.TaskFunctionType[source]

Perform a frequency sweep and generate solutions for recorded channels.

Parameters:: kwargs – keyword arguments unpacked from the JSON argument.
Returns:: a task function to be submitted to the task executor.

GetCalibration(frequency_channel: int, calibration_id: str | None = None) → list[float][source]

Get a calibration from the calibration store.

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"]
}

Parameters:

frequency_channel – the frequency channel to calibrate for.
calibration_id – optional unique identifier for calibration.

Returns:

a calibration from the calibration store.

GetFittedSolutions(frequency_channels: list[int], calibration_id: str | None = None) → list[float][source]

Get fitted calibration solutions for a set of frequency channels.

Fetches phase fit parameters from the calibration store once and reconstructs gains for all requested channels in a single call, avoiding per-channel round-trips from MccsStation.ApplyConfiguration.

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_GetFittedSolutions.json",
    "title": "MccsStationCalibrator GetFittedSolutions schema",
    "description": "Schema for MccsStationCalibrator's GetFittedSolutions command",
    "type": "object",
    "properties": {
        "frequency_channels": {
            "description": "List of frequency channel indices to compute fitted solutions for.",
            "type": "array",
            "items": {
                "type": "integer",
                "minimum": 0,
                "maximum": 511
            },
            "minItems": 1
        },
        "calibration_id": {
            "description": "Optional unique identifier for the calibration.",
            "type": "string"
        }
    },
    "required": ["frequency_channels"]
}

Parameters:

frequency_channels – channel indices to compute solutions for.
calibration_id – optional unique identifier for the calibration.

Returns:

flat float array of shape (n_freqs * n_ant * 8,).

StartCalibrationLoop(**kwargs: Any) → tuple[list[ska_control_model.ResultCode], list[str | None]][source]

Start the calibration loop.

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": {
        "calibration_id": {
            "description": "Unique identifier for the calibration. If not provided, a random identifier of the form `job_<8 hex chars>` is generated.",
            "type": [
                "string",
                "null"
            ]
        },
        "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. Default is 'FEKO_AAVS3_vogel_256_elem_50ohm_'.",
            "type": [
                "string",
                "null"
            ]
        },
        "jones_solve": {
            "description": "Whether to call the polarised solver after initial calibration. Default is False.",
            "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. Default is True.",
            "type": [
                "boolean",
                "null"
            ]
        },
        "masked_antennas": {
            "description": "List of antennas which should not be used for calibration. Default is an empty list (no antennas masked).",
            "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. Default is 'gsm'.",
            "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": [
                "boolean",
                "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 value 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. Defaults to 32 for channels below 320 and 64 for channels 320 and above. nside=32 corresponds to approximately 110 arcminutes and is adequate at low frequencies; nside=64 (approximately 55 arcminutes) is used at higher frequencies where the beam is narrower.",
            "type": [
                "integer",
                "null"
            ],
            "default": 32,
            "minimum": 1
        },
        "niter": {
            "description": "Number of iterations for solver. Default is 200.",
            "type": [
                "integer",
                "null"
            ],
            "default": 200,
            "minimum": 1
        },
        "first_channel": {
            "description": "The lowest channel in the loop. Default is 64 (50 MHz with an 800 MHz ADC clock and 512 channels).",
            "type": [
                "integer",
                "null"
            ],
            "minimum": 0,
            "maximum": 511
        },
        "last_channel": {
            "description": "The final channel to use in loop. Default is 448.",
            "type": [
                "integer",
                "null"
            ],
            "minimum": 0,
            "maximum": 511
        },
        "daq_mode": {
            "description": "The DAQ mode to use for acquiring data. This is expected to be one of 'TCC' or 'xGPU'. Default is 'TCC'.",
            "type": [
                "string",
                "null"
            ],
            "enum": [
                "TCC",
                "xGPU",
                null
            ]
        },
        "daq_config": {
            "description": "Any overrides to pass to DAQ for configuration purposes. Default is an empty dict.",
            "type": [
                "object",
                "null"
            ],
            "additionalProperties": true
        }
    },
    "required": [],
    "additionalProperties": false
}

Parameters:: kwargs – keyword arguments unpacked from the JSON argument.
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[str | None]][source]

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[source]

Initialise this device object.

Parameters:

args – positional args to the init
kwargs – keyword args to the init

calibrationLoopParams() → str[source]

Return the parameters of the calibration loop in a json format.

Returns:: the parameters of the calibration loop in a json format.

calibrationSchedule(value: str) → None[source]

Set the calibration schedule.

Parameters:: value – a JSON array of schedule entries, each with a "cron" key and optional calibration parameter overrides, or an empty string to clear the schedule.
Raises:: ValueError – if any entry is missing "cron", or contains an invalid cron expression.

create_component_manager() → StationCalibratorComponentManager[source]

Create and return a component manager for this device.

Returns:: a component manager for this device.

init_device() → None[source]: Initialise the device.

isLoopRunning() → bool[source]

Return whether or not the calibration loop is currently running.

Returns:: whether or not the calibration loop is currently running.

is_GenerateCalibrationSolution_allowed(request_type: LRCReqType | None = ska_tango_base.long_running_commands.LRCReqType.ENQUEUE_REQ) → bool[source]

Return whether the GenerateCalibrationSolution command is allowed.

Parameters:: request_type – The request type
Returns:: True if command is allowed.
Raises:: CmdNotAllowedError – if the calibration loop or schedule is running.

jobsInQueue() → int[source]

Return number of correlator files left to process.

Returns:: number of correlator files left to process.

latestPreferredJobId() → str[source]

Return the user_friendly_name of the most recent preferred calibration job.

Returns:: the user_friendly_name of the most recent preferred calibration job, or an empty string if no preferred job exists.

scheduleState(value: CalibrationScheduleState) → None[source]

Pause or resume the calibration scheduler.

Parameters:: value – PAUSED to suspend firing, RUNNING to resume. EMPTY is ignored — that state is managed automatically by writing an empty calibrationSchedule.

scheduledSweeps() → str[source]

Return all calibration runs scheduled within the next 24 hours.

Each item in the returned array has a "scheduled_time" (ISO 8601 UTC) and any per-run parameter overrides from the schedule entry.

Returns:: a JSON array of upcoming runs, sorted by time, or "[]" if no schedule is set.

class NumpyEncoder(*, skipkeys=False, ensure_ascii=True, check_circular=True, allow_nan=True, sort_keys=False, indent=None, separators=None, default=None)[source]

Converts numpy types to JSON.

default(obj: Any) → Any[source]

Dereturn the encoded value.

Parameters:: obj – the object.
Returns:: the encoded value.

class StationCalibratorComponentManager(*args: Any, **kwargs: Any)[source]

A component manager for a station calibrator.

__init__(logger: Logger, station_name: str, station_id: int, calibration_store_name: str, calibration_solver_trl: str, station_config_path: list[str], communication_state_callback: Callable[[ska_control_model.CommunicationStatus], None], component_state_callback: Callable[[...], None], channels_to_ignore: list[int] | None = None, event_serialiser: ska_low_mccs_common.EventSerialiser | None = None) → None[source]

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.
station_config_path – TMData reference [source, yaml_path] for the station configuration, used as the default when not supplied in a command.
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.
channels_to_ignore – channel indices with known RFI to skip when solving.
event_serialiser – the event serialiser to be used by this object.

property calibration_schedule: list[dict[str, Any]]

Return the active schedule entries.

Returns:: the list of active schedule entries.

cleanup() → None[source]

Cleanup resources held by the component manager.

This includes cleaning up resources held by all sub-component managers.

do_generate_calibration_solution(task_callback: Callable | None = None, task_abort_event: Event | None = None, **kwargs: Any) → None[source]

Generate the calibration solution.

Parameters:

task_callback – optional callback
task_abort_event – optional abort event
kwargs – see schema for keyword argument definitions

Raises:

ValueError – if unexpected result code received

get_calibration(frequency_channel: int, calibration_id: str | None = None) → list[float][source]

Get a calibration from the calibration store.

Parameters:

frequency_channel – the frequency channel to calibrate for
calibration_id – Unique identifier for calibration.

Returns:

A list of calibration values

get_fitted_solutions(frequency_channels: list[int], calibration_id: str | None = None) → list[float][source]

Get fitted calibration solutions for a set of frequency channels.

Calls GetFittedSolutions on the calibration store once, returning gains reconstructed from the stored phase fit parameters for every requested channel in a single round-trip.

Parameters:

frequency_channels – channel indices to compute solutions for.
calibration_id – optional unique identifier for the calibration.

Returns:

flat float array of shape (n_freqs * n_ant * 8,).

property latest_preferred_job_id: str

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

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

property schedule_state: CalibrationScheduleState

Return the current scheduler state.

Returns:: the current scheduler state.

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

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[source]: Establish communication with the StationCalibrator components.

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

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[source]: Break off communication with the StationCalibrator components.

class StationCalibratorHealthModel(*args: Any, **kwargs: Any)[source]

A health model for a station calibrator.

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

fit_gains(channels: ndarray, phases: ndarray) → tuple[ndarray, ndarray, ndarray, ndarray][source]

Fit a linear delay model to the observed phase data.

For each antenna and polarisation, fits phase = offset + 2π·f·delay over the supplied channels. The resulting parameters can be evaluated at any frequency, including channels the correlator skipped.

Parameters:

channels – 1-D array of channel indices that were actually observed.
phases – 2-D array of phases, shape [n_channels, n_ant * 4], where the inner axis layout per antenna is [xx, xy, yx, yy].

Returns:

A tuple of (offsets, delays, cov_matrices, costs), each shaped for 256 * 4 Jones elements with XX at stride-4 index 0 and YY at stride-4 index 3.