CSP.LMC Subarray Consistency Policy

class ska_csp_lmc_common.observation.subarray_policy.CspScanConsistencyPolicy(*, logger: Logger | None = None, apply_inconsistency_policy: 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:

logger – Logger used to report consistency warnings.
apply_inconsistency_policy – 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”}. The set is refined at runtime based on the active observing modes.

evaluate(aggregated: ska_control_model.ObsState, obs_modes: List[ska_control_model.ObsMode], subsystems: Dict[str, ComponentSnapshot], previous_state: ska_control_model.ObsState | None = None, command_context: CommandContext | None = None, reconcile_timeout: bool | None = False) → Decision

Evaluate the consistency of subsystems against the aggregated state.

The consistency check normally runs only while the aggregated state is SCANNING. A special case is also handled for transitions from SCANNING to EMPTY or IDLE, because a restart of a critical subsystem may collapse the aggregated state before the inconsistency is classified.

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[str, ComponentSnapshot]) → Decision

Apply consistency rules for the SCANNING state

This method refines the set of required subsystems according to the active observing modes, filters out components that are not relevant to the current scan, evaluates subsystem inconsistencies and returns the final decision for the aggregated obsState.

Parameters:

obs_modes – List of active observation modes.
subsystems – Dictionary of subsystem Components and their states.

Returns:

A decision for the aggregated obsState.

_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, and returns the set of subsystems that are required for the current scan.

The default required subsystem set is adapted to the active observing modes so that only the subsystems that contribute to the current observation are validated.

Parameters:: obs_modes – Active observing modes.
Returns:: Set of subsystem tokens required for consistency checks.

_filter_unassigned_subsystems(subsystems: Dict[str, ComponentSnapshot]) → Dict[str, ComponentSnapshot]

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

In particular, PST beams that are not assigned to the current subarray are ignored, because their obsState must not influence the consistency evaluation of this scan.

Parameters:: subsystems – Mapping between subsystem components and their reported obsStates.
Returns:: Filtered subsystem mapping containing only components relevant to the current scan.

_calculate_pst_threshold(subsystems: Dict[str, ComponentSnapshot]) → 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.

The returned value represents the maximum number of failing PST beams that can be tolerated before the failure count is considered critical. The policy uses a strict ‘greater than threshold’ comparison, with a special-case behaviour for a single beam: if only one PST beam is present, a failure of that beam is enough to trigger a fault.

Parameters:: subsystems – Mapping between subsystem components and their reported obsStates.
Returns:: Failure threshold used for PST-beam fault aggregation.

_find_invalid_subsystems_for_scan(subsystems: Dict[str, ComponentSnapshot], 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.

Each inconsistent subsystem is classified and added to the returned list. PST beams are handled separately so that their impact can be aggregated according to the active observing modes: - in pure PULSAR_TIMING scans, the number of failing PST beams is compared with the configured threshold; - in commensal-mode scans, PST failures are reported as degraded-scan conditions and do not force the aggregated obsState to FAULT on their own.

Parameters:

subsystems – Dictionary of subsystem Components and their states.
required_subs – Subsystem tokens that are required for the current scan.
obs_modes – Active observing modes.

Returns:

A list of dictionaries, each representing an invalid subsystem.

_build_pst_threshold_exceeded_inconsistency(invalid_pst_beams: List[Dict[str, Any]]) → Dict[str, Any]

Build a synthetic inconsistency when failing PST beams exceed the threshold in PULSAR_TIMING-only scans.

Parameters:: invalid_pst_beams – Classified inconsistent PST beams collected during scan validation.
Returns:: Synthetic inconsistency dictionary used to force a FAULT decision when the threshold is exceeded.

_log_commensal_mode_pst_warning(invalid_pst_beams: List[Dict[str, Any]], obs_modes: List[ska_control_model.ObsMode]) → None

Log a warning for PST inconsistencies in commensal observing mode.

Parameters:

invalid_pst_beams – Classified inconsistent PST beams to report.
obs_modes – Active observing modes used to explain why scanning can continue.

_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 (obsState 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, Severity]

Derive the final aggregated obsState from classified inconsistencies.

The most severe inconsistency determines the final decision: - HIGH severity forces a transition to FAULT; - MEDIUM severity keeps the scan in SCANNING but marks it as degraded; - LOW severity keeps the scan in SCANNING.

Parameters:: invalid_subsystems – Classified inconsistent subsystems.
Returns:: Tuple containing the final obsState, the hard-fault flag and the highest detected severity.

_abc_impl = <_abc._abc_data object>

_is_protocol = False