CSP.LMC Subarray Consistency Policy

class ska_csp_lmc_common.observation.subarray_policy.CspScanConsistencyPolicy(*, logger: Logger | None = None, hard_fault_on_inconsistency: bool = True, required_subsystems: Set[str] | None = None)

Bases: ScanConsistencyPolicy

Consistency policy for CSP scanning operations.

This policy monitors subsystem health during active scans. It flags inconsistencies if a subsystem restarts or fails. If the inconsistency is non-recoverable, it triggers a transition to FAULT to protect system integrity and data quality.

Parameters:

  • hard_fault_on_inconsistency – Whether to trigger a hard fault if subsystems are inconsistent.

  • required_subsystems – Set of subsystem tokens (e.g., ‘cbf’) required for a valid scan. Defaults to {“cbf”, “pss”, “pst”}.

evaluate(aggregated: ska_control_model.ObsState, obs_modes: List[ska_control_model.ObsMode], subsystems: Dict[Component, ska_control_model.ObsState], previous_state: ska_control_model.ObsState | None = None) → Decision

Evaluate the consistency of subsystems against the aggregated state.

Parameters:

  • aggregated – The current overall observation state.

  • obs_modes – List of active observation modes.

  • subsystems – Dictionary of subsystem components and their states.

  • previous_state – The previously published ObsState.

Returns:

A Decision object containing the state and fault info.

_evaluate_scan_consistency(obs_modes: List[ska_control_model.ObsMode], subsystems: Dict[Component, ska_control_model.ObsState]) → Decision

Apply consistency rules for the SCANNING state.

Parameters:

  • obs_modes – List of active observation modes.

  • subsystems – Dictionary of subsystem Components and their states.

_refine_required_subsystems(obs_modes: List[ska_control_model.ObsMode]) → Set[str]

Remove the subsystems that are not mentioned in the ObsModes from the required subsystems set.

_filter_unassigned_subsystems(subsystems: Dict[Component, ska_control_model.ObsState]) → Dict[Component, ska_control_model.ObsState]

Remove the subsystem components that are not involved in the scanning.

_calculate_pst_threshold(subsystems: Dict[Component, ska_control_model.ObsState]) → int

Calculate a PST beam threhsold number, which represent the acceptable number of malfunctioning PST beams that degradate but not invalidate a PULSAR_TIMING scan.

_find_invalid_subsystems_for_scan(subsystems: Dict[Component, ska_control_model.ObsState], required_subs: Set[str], obs_modes: List[ska_control_model.ObsMode]) → List[Dict[str, Any]]

Identify required subsystems that are not in the SCANNING obsState.

Parameters:

subsystems – Dictionary of subsystem Components and their states.

Returns:

A list of dictionaries, each representing an invalid subsystem.

_classify_scan_inconsistency(fqdn: str, obs_state: ska_control_model.ObsState, obs_modes: List[ska_control_model.ObsMode]) → Dict[str, str]

Classify the type of inconsistency based on the subsystem’s obsState.

It returns a structured description of the failure mode.

Parameters:

  • fqdn – The fqdn of the component in unexpected obsState

  • obs_state – The unexpected obsState of the subsystem.

  • obs_modes – The active observation modes.

Returns:

A dictionary containing a ‘code’ and ‘description’ of the issue.

_build_scan_consistency_details(obs_modes: List[ska_control_model.ObsMode], invalid_subsystems: List[Dict[str, Any]]) → str

Build a detailed, human-readable fault message.

Example:

ObsMode:IMAGING. SCAN consistency check failed.
Causes:
- low-cbf/subarray/01 was unexpectedly restarted during the
  scan (state is EMPTY, should be SCANNING).
Parameters:

  • obs_modes – The active observation modes.

  • invalid_subsystems – The list of subsystems that failed validation.

Returns:

A formatted diagnostic message string.

_evaluate_decision_final_obsstate(invalid_subsystems: List[Dict[str, Any]]) → Tuple[ska_control_model.ObsState, bool]

Select the final CSP Subarray ObsState to be emitted.

_abc_impl = <_abc._abc_data object>
_is_protocol = False