ska_oso_tmcsim

The tmcsim package holds code related to simulation of TMC CentralNode and TMC SubarrayNode Tango devices.

centralnode

Simulates the behaviour of a TMC CentralNode for the purpose of integration tests.

ska_oso_tmcsim.centralnode.get_centralnode_trl(domain: str) → str

Get the TRL for a TMC CentralNode.

Returns pre-ADR-9 TRLs if the Tango domain is an old-style ‘ska_mid’ or ‘ska_low’ domain.

@param domain: Tango domain @return: full TRL for the TMC CentralNode

class ska_oso_tmcsim.centralnode.CentralNode(*args: Any, **kwargs: Any)

Simulates the bare minimum TMC CentralNode device server functionality required for OSO/TMC integration tests.

domain

The domain part of the device FQDN, eg TANGO_HOST/domain/family/member

init_device()

Simulate CentralNode device initialisation.

AssignResources(cdm_str)

Assign resources to a subarray.

ReleaseResources(cdm_str)

Release resources from a subarray.

cbfnode

Simulates the CBF subarray’s processorsReadyPercent attribute, providing input for the Subarray Quality Monitor (SQM). The SQM is a Transducer device that subscribes to processorsReadyPercent and exposes a derived readyToScan attribute (true when processorsReadyPercent == 100). In this programme increment (PI30), readyToScan depends solely on processorsReadyPercent, but additional inputs will be added in future PIs.

ska_oso_tmcsim.cbfnode.get_cbf_trl(domain: str, subarray_id: int) → str

Get the TRL for a CBF simulator device.

Parameters:

• domain – Tango domain (e.g. “mid-tmc” or “low-tmc”)

• subarray_id – Subarray ID

Returns:

Full TRL for the CBF simulator

class ska_oso_tmcsim.cbfnode.CbfSimulator(*args: Any, **kwargs: Any)

Lightweight CBF subarray simulator that publishes processorsReadyPercent via change events, with runtime control for happy/sad path testing.

Auto-triggers: subscribes to SubArrayNode.obsState change events and starts the ramp when obsState transitions to READY, resets when it transitions to IDLE or EMPTY. Explicit StartRamp()/Reset() calls override the auto-trigger.

SubarrayNodeTRL

Full TRL of the SubArrayNode to subscribe to for auto-trigger.

RampDuration

Duration in seconds for StartRamp() to ramp from 0 to 100.

init_device()

Simulate CBF device initialisation.

delete_device()

Tear down event subscriptions and cancel any running ramp.

processorsReadyPercent() → int

Get the current percentage of processors ready.

StartRamp()

Start ramping processorsReadyPercent from 0 to 100.

SetFailed()

Cancel the ramp, leaving processorsReadyPercent at its current value.

SetFailedOnNextRamp(percentage: int)

Make the next ramp stop at the given percentage instead of reaching 100.

Cleared after the ramp stops or by Reset/SetFailed.

Reset()

Cancel any in-progress ramp and reset processorsReadyPercent to 0.

SimulateObsState(obsstate_value: int)

Simulate receiving an obsState change event for testing.

The in-process Tango test context does not deliver change events across devices, so this command exercises the same callback logic directly.

obsstatestatemachine

obsstatemachine encodes the ADR-8 Subarray state model into a

class ska_oso_tmcsim.obsstatestatemachine.ObsStateMachineMixin(*args, **kwargs)

This mixin extends the default statemachine MachineMixin to allow setting of the state machine’s initial state. This specialisation only handles obsStates and expects the initial state to be set via a Tango device property.

state_field_name: str = 'state'

The model’s state field name that will hold the state value.

state_machine_name: str = None

A fully qualified name of the class, where it can be imported.

state_machine_attr: str = 'statemachine'

Name of the model’s attribute that will hold the machine instance.

initial_state_attr: str = 'initial_state'

Name of the device property that holds the desired initial state machine state.

class ska_oso_tmcsim.obsstatestatemachine.BaseObsStateMachine(model: Any | None = None, state_field: str = 'state', start_value: Any | None = None, rtc: bool = True, allow_event_without_transition: bool = False, listeners: List[object] | None = None)

Simple state machine for tracking SubArrayNode transitions as defined in ADR-8.

https://confluence.skatelescope.org/x/bIdIBg

is_release_all(cdm_json)

Guard to detect whether a JSON string includes the term ‘release_all’.

The end state for ReleaseResources is different depending on whether the payload contains ‘release_all’ or not. This guard is used to find the desired end state.

Parameters:

cdm_json – JSON to analyse

Returns:

True if ‘release_all’ in JSON.

class ska_oso_tmcsim.obsstatestatemachine.ObsStateStateMachine(model: Any | None = None, state_field: str = 'state', start_value: Any | None = None, rtc: bool = True, allow_event_without_transition: bool = False, transition_timing: dict[str, int | float] | None = None)

State machine that encodes the transitions in the ADR-8 subarray state model.

set_to_fail_after(*states: State)

Primes a transition to FAULT after a state sequence matching the input state sequence.

Parameters:

states – state sequence to match

on_enter_state(state)

Perform actions that should occur on entering a new state.

This template function hooks into python-statemachine and is called whenever a new state is entered.

Parameters:

state – new state

after_transition(event)

Perform actions that should occur after a state transition.

This template function hooks into python-statemachine and is called whenever a state transition has happened and the state has changed.

Parameters:

event – e.g. ‘configure’

class ska_oso_tmcsim.obsstatestatemachine.LoggingObserver

State machine observer that logs every transition. Useful for development and testing.

See https://python-statemachine.readthedocs.io/en/latest/observers.html

on_transition(event: str, source: State, target: State)

Logs state transitions as and when they occur.

subarraynode

Simulates the behaviour of a TMC SubArrayNode for integration testing.

ska_oso_tmcsim.subarraynode.get_subarraynode_trl(domain: str, subarray_id: int) → str

Get the TRL for a TMC SubArrayNode.

Returns pre-ADR-9 TRLs if the Tango domain is an old-style ‘ska_mid’ or ‘ska_low’ domain.

@param domain: Tango domain @param subarray_id: Subarray ID @return: full TRL for the SubArrayNode

class ska_oso_tmcsim.subarraynode.MethodCall(*args: Any, **kwargs: Any)

Simple dataclass for describing a method call on a Tango device server.

class ska_oso_tmcsim.subarraynode.SubArrayNode(*args: Any, **kwargs: Any)

Simulates the bare minimum TMC SubArrayNode device server functionality required for OSO/TMC integration tests.

state_machine_name: str = 'ObsStateStateMachine'

A fully qualified name of the class, where it can be imported.

state_machine_attr: str = 'statemachine'

Name of the model’s attribute that will hold the machine instance.

state_field_name: str = '_obsState'

The model’s state field name that will hold the state value.

initial_state_attr: str = 'initial_obsstate'

Name of the device property that holds the desired initial state machine state.

initial_obsstate

Initial obsState for state machine provided by the OSO test harnesses

history_limit

Maximum size for the call history. The oldest entries will be removed to stay below this limit.

init_device()

Simulate SubArrayNode device initialisation.

on_transition(event: str, source: State, target: State)

Template function used by the simulator to hook into statemachine transitions.

obsState() → ObsState

Get the current obsState of this SubArrayNode.

AssignResources(cdm_str)

Add resources to a subarray.

ReleaseResources(cdm_str)

Release resources from a subarray.

Configure(cdm_str)

Configure subarray resources.

Scan(cdm_str)

Perform a scan.

Abort(*args, **kwargs)

Abort the current operation.

End(*args, **kwargs)

Mark the end of the scheduling block scan sequence.

Restart(*args, **kwargs)

Restart the device.

History() → str

Get the history of commands and arguments received by this device.

The returned JSON string will a serialised list of JSON objects, one object for each call received by the device. See the MethodCall class for details of the object. The maximum size of the list will match the call history size limit, which is set by the history_limit device property.

Returns:

JSON command history

ClearHistory()

Clear the history of JSON arguments.

InjectFaultAfter(states_str)

Will cause the state to go to FAULT if the state machine passes through the states given in the arg.

Parameters:

states_str – string in the format “[‘IDLE’, ‘CONFIGURING’]”

InjectDelay(cmd_with_delay_str: str)

Will cause the state transition for the event to have a delay, to simulate a long-running command.

NOTE: THIS ISN’T THE SAME BEHAVIOUR AS REAL TMC. A command will always return quickly and it is the state transitions in a separate process where there might be a delay. Here the state transitions happen within the command before it returns.

Parameters:

cmd_with_delay_str – A serialised dict with the delay for each command, e.g. “{‘Configure’: 1, ‘AssignResources’: 0.5}

testharness

This module contains code related to running the OSO simulators in a test Tango context.

class ska_oso_tmcsim.testharness.TMCSimTestHarness(domain: str)

TMCSimTestHarness is an integration test harness for OSO’s TMC Simulator.

TMCSimTestHarness can populate a Tango test context with simulations of

• TMC CentralNode

• TMC SubArrayNode

• TMC Subarray Quality Monitor

• TMC MCCS subarray leaf node

• CSP Leaf node

• CSP CBF, for the Quality Monitor to react to

• SDP controller

add_central_node()

Make CentralNode available within the test context.

add_csp_leaf_node(subarray_id: int = 1)

Make CSP leaf node available within the test context.

Parameters:

subarray_id – The subarray ID for which the leaf node should be added. Defaults to 1.

add_sdp_controller_node(sdp_version=None)

Make an SDP controller node available within the test context.

add_subarray(subarray_id: int, initial_obsstate: ObsState = ObsState.EMPTY)

Add a subarray to the test harness.

The subarray obsState will be set to the default obsState of EMPTY unless overridden.

Parameters:

• subarray_id – The subarray ID for which the subarray should be added.

• initial_obsstate – The initial obsState of the subarray. Defaults to ObsState.EMPTY.

add_cbf_device(subarray_id: int, ramp_duration=None)

Make a CBF simulator device available within the test context.

Parameters:

• subarray_id – The subarray ID for the CBF simulator.

• ramp_duration – Optional ramp duration in seconds.

add_subarray_quality_monitor(subarray_id: int)

Make a SubarrayQualityMonitor (Transducer) available within the test context.

The device is added with no properties. At runtime, observing scripts configure it to subscribe to CBF processorsReadyPercent and derive readyToScan (processorsReadyPercent == 100).

Parameters:

subarray_id – The subarray ID for the quality monitor.

add_mccs_subarray_leaf_node(subarray_id: int)

Make a MCCS Subarray Leaf Node available within the test context.

Parameters:

subarray_id – The subarray ID for the leaf node.