CSP.LMC Subarray Consistency Policy

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

Bases: ScanConsistencyPolicy

Evaluate CSP subsystem consistency during scan-related states.

The policy validates whether CSP subsystems are in observation states compatible with the current scan situation.

When a retained Scan command context with schedule metadata is available, the policy evaluates the snapshot using scheduled-scan gates. Before scan entry, the CSP Subarray remains READY until all valid successful planned devices are SCANNING. After the CSP Subarray has entered SCANNING, it remains SCANNING until all valid successful planned devices have returned READY.

When no active scan schedule is available, the policy falls back to the legacy observed-state behaviour. In that mode, scan consistency checks are run when the aggregated state is SCANNING or when the system appears to collapse unexpectedly from SCANNING to EMPTY or IDLE.

This class does not interpret command lifecycle state. In particular, it does not decide whether a Scan, EndScan, or Abort command is in progress or completed. Those decisions are handled by the command-aware policy and the observation supervisor.

Parameters:

  • logger – Logger used to report consistency warnings.

  • apply_inconsistency_policy – Whether detected hard inconsistencies should force a FAULT decision.

  • required_subsystems – Default subsystem families required for a valid scan. The set is refined at runtime according to 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, retained_scan_context: CommandContext | None = None, reconcile_timeout: bool | None = False, refreshed_after_reconcile: bool | None = False) → Decision

Evaluate CSP scan consistency for the current observation snapshot.

If a Scan lifecycle context with schedule metadata is available, the snapshot is evaluated with the scheduled-scan start and exit gates. Otherwise, the policy uses the legacy observed-state scan consistency rules.

Parameters:

  • aggregated – Current aggregated observation state.

  • obs_modes – Active observation modes reported for the subarray.

  • subsystems – Snapshot of subsystem states used for consistency evaluation.

  • previous_state – Previously published stable observation state.

  • command_context – Current command context, if a command is being reconciled.

  • retained_scan_context – Retained Scan context from the active or most recent scan lifecycle.

  • reconcile_timeout – Whether command reconciliation has timed out.

  • refreshed_after_reconcile – Whether subsystem snapshots have already been refreshed after reconciliation.

Returns:

Policy decision to apply, wait, or fault.

_resolve_scan_lifecycle_context(*, command_context: CommandContext | None, retained_scan_context: CommandContext | None) → CommandContext | None

Return the scan lifecycle context for consistency checks.

The active command context is only relevant when it is itself a Scan command. During EndScan or other commands, scan consistency must continue to use the retained Scan context captured when the scan started.

Abort has precedence over any retained Scan lifecycle. While Abort is active, subsystem states such as ABORTING and ABORTED are expected and must not be evaluated as Scan consistency violations

_accepted_obs_states_for_observed_scan(*, command_context: CommandContext | None) → tuple[ska_control_model.ObsState, ...]

Return accepted observed states for legacy scan lifecycle checks.

During EndScan, selected devices may legitimately transition from SCANNING to READY at different times. Treat those mixed snapshots as transitional rather than as scan inconsistencies.

should_override_in_progress_command(context: CommandContext) → bool

Return whether scan lifecycle policy should override command progress.

Scheduled Scan commands are handled by this policy because their immediate observable state may legitimately differ from the Scan command target during the retained scan lifecycle. For example, the CSP Subarray may remain READY before scan entry, and may remain SCANNING while waiting for selected devices to return READY after scan execution.

Generic command progress reconciliation must therefore defer to the scan lifecycle policy whenever the current command is a Scan with retained schedule metadata.

_evaluate_scheduled_scan(*, aggregated: ska_control_model.ObsState, obs_modes: list[ska_control_model.ObsMode], subsystems: dict[str, ComponentSnapshot], previous_state: ska_control_model.ObsState | None = None, retained_scan_context: CommandContext, reconcile_timeout: bool | None = False, refreshed_after_reconcile: bool | None = False) → Decision

Evaluate scheduled Scan consistency using start and exit gates.

Start gate: The CSP Subarray remains READY until all valid successful planned devices are SCANNING.

Exit gate: Once the CSP Subarray has entered SCANNING, it remains SCANNING until all valid successful planned devices are READY.

During scheduled-scan gate evaluation, READY and SCANNING are accepted transition states. A valid device is a successful planned device whose current state is still accepted by the scan consistency policy.

Runtime-failed devices are excluded from gate counting only after _evaluate_scan_consistency has confirmed that their state does not represent a hard fault.

_cbf_uncommanded_empty(*, previous_state: ObsState | None, command_context: CommandContext | None, subsystems: dict[str, ComponentSnapshot]) → bool

Return whether CBF unexpectedly moved to EMPTY.

This condition is detected only when no command context is active and the previous CSP Subarray ObsState was not EMPTY. In this situation, a CBF transition to EMPTY is considered uncommanded, because it was not caused by a command-driven release or recovery operation.

Parameters:

  • previous_state – Previous CSP Subarray ObsState, or None if no previous state is available.

  • command_context – Active command context, or None if no command is currently being evaluated.

  • subsystems – Mapping of subsystem FQDNs to their latest component snapshots.

Returns:

True if a CBF component is in EMPTY without an active command and the previous CSP Subarray ObsState was not EMPTY; False otherwise.

_evaluate_observed_scan_state(aggregated: ska_control_model.ObsState, accepted_obs_states: Collection[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, retained_scan_context: CommandContext | None = None) → Decision
_evaluate_scan_consistency(obs_modes: List[ska_control_model.ObsMode], subsystems: Dict[str, ComponentSnapshot], accepted_obs_states: Collection[ska_control_model.ObsState], final_state_if_consistent=None, retained_scan_context: CommandContext | None = None) → Decision

Apply scan consistency rules for the accepted observation states.

This method determines the devices that are relevant to the scan, evaluates whether their ObsState is included in accepted_obs_states, classifies subsystem inconsistencies, and returns the final policy decision.

When a retained Scan context is available, the check uses the requested observing modes, planned devices, and family rules captured when the Scan lifecycle started. Otherwise, it falls back to inferring required subsystems and family rules from the currently reported observing modes.

Parameters:

  • obs_modes – Active observation modes.

  • subsystems – Dictionary of subsystem component snapshots.

  • accepted_obs_states – ObsStates accepted as consistent for this check, for example SCANNING for legacy scans or READY/SCANNING for scheduled-scan transitions.

  • final_state_if_consistent – Optional final ObsState to return when no inconsistency is found.

  • retained_scan_context – Retained Scan context used for command aware scan consistency checks.

Returns:

Policy decision for the aggregated observation state.

_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.

_find_invalid_subsystems_for_scan(family_rule: Dict[str, FamilyRule], required_subs: Set[str], obs_modes: List[ska_control_model.ObsMode], accepted_obs_state: Collection[ska_control_model.ObsState] = (ska_control_model.ObsState.SCANNING,)) → List[Dict[str, Any]]

Identify required subsystems outside the accepted ObsStates.

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:

  • family_rule – Family-specific scan consistency rules.

  • required_subs – Required subsystem component snapshots keyed by device FQDN.

  • obs_modes – Active observing modes.

  • accepted_obs_state – ObsStates accepted as valid for this scan consistency check.

Returns:

A list of dictionaries, each representing an invalid subsystem.

_build_invalid_subsystem_info(*, component: ComponentSnapshot, obs_modes: List[ska_control_model.ObsMode]) → Dict[str, Any]

Classify a required subsystem that is outside accepted obsStates.

_resolve_invalid_pst_beams(*, family_rule: FamilyRule, invalid_pst_beams: List[Dict[str, Any]], obs_modes: List[ska_control_model.ObsMode], pulsar_timing_only: bool) → List[Dict[str, Any]]

Resolve PST inconsistencies according to active observing modes.

_resolve_pulsar_timing_invalid_pst_beams(*, family_rule: FamilyRule, invalid_pst_beams: List[Dict[str, Any]]) → List[Dict[str, Any]]

Apply threshold-based PST handling in PULSAR_TIMING-only scans.

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

Downgrade PST inconsistencies during commensal observing modes.

_abc_impl = <_abc._abc_data object>
_is_protocol = False