Reference Subarray Device

A reference implementation of an SKA subarray device.

It inherits from SKASubarray but provides schemas for some commands.

class ska_tango_base.testing.reference.reference_subarray_device.SKASubarray[source]

A generic subarray device for SKA.

SKASubarray inherits from the SKAObsDevice and AbstractSubarrayInterface, and expects a component manager to be provided by implementing the create_component_manager() method.

__init__(*args: Any, **kwargs: Any) → None[source]

Initialise a new instance.

Parameters:

  • args – positional arguments.

  • kwargs – keyword arguments.

class InitCommand[source]

A class for the SKASubarray’s init_device() “command”.

do(*args: Any, **kwargs: Any) → tuple[ResultCode, str][source]

Stateless hook for device initialisation.

Parameters:

  • args – positional arguments to the command. This command does not take any, so this should be empty.

  • kwargs – keyword arguments to the command. This command does not take any, so this should be empty.

Returns:

A tuple containing a return code and a string message indicating status. The message is for information purpose only.

on_new_shared_bus() → None[source]

Initialise signals.

create_component_manager() → ComponentManagerT[source]

Create and return a component manager for this device.

Raises:

NotImplementedError – because it is not implemented.

_component_state_changed(fault: bool | None = None, power: PowerState | None = None, obsfault: bool | None = None, resourced: bool | None = None, configured: bool | None = None, scanning: bool | None = None) → None[source]

Update the device about the component’s state.

Subclasses should ensure that this is called by the component_state_callback callback passed to their component manager.

Parameters:

  • fault – The new fault state of the device, or None if unchanged

  • power – The new power state of the device, or None if unchanged

read_assignedResources() → list[str][source]

Read the resources assigned to the device.

The list of resources assigned to the subarray.

Returns:

Resources assigned to the device.

read_configuredCapabilities() → list[str][source]

Read capabilities configured in the Subarray.

A list of capability types with no. of instances in use on this subarray; e.g. Correlators:512, PssBeams:4 PstBeams:4, VlbiBeams:0.

Returns:

A list of capability types with no. of instances used in the Subarray

class AssignResourcesCommand[source]

A class for SKASubarray’s AssignResources() command.

__init__(command_tracker: CommandTrackerProtocol, component_manager: SubarrayComponentManager, callback: Callable[[bool], None] | None = None, logger: Logger | None = None, schema: dict[str, Any] | None = None) → None[source]

Initialise a new instance.

Parameters:

  • command_tracker – the device’s command tracker

  • component_manager – the device’s component manager

  • callback – an optional callback to be called when this command starts and finishes.

  • logger – a logger for this command to log with.

  • schema – an optional JSON schema for the command argument.

class ReleaseResourcesCommand[source]

A class for SKASubarray’s ReleaseResources() command.

__init__(command_tracker: CommandTrackerProtocol, component_manager: SubarrayComponentManager, callback: Callable[[bool], None] | None = None, logger: Logger | None = None, schema: dict[str, Any] | None = None) → None[source]

Initialise a new instance.

Parameters:

  • command_tracker – the device’s command tracker

  • component_manager – the device’s component manager

  • callback – an optional callback to be called when this command starts and finishes.

  • logger – a logger for this command to log with.

  • schema – an optional JSON schema for the command argument.

class ReleaseAllResourcesCommand[source]

A class for SKASubarray’s ReleaseAllResources() command.

__init__(command_tracker: CommandTrackerProtocol, component_manager: SubarrayComponentManager, callback: Callable[[bool], None] | None = None, logger: Logger | None = None) → None[source]

Initialise a new instance.

Parameters:

  • command_tracker – the device’s command tracker

  • component_manager – the device’s component manager

  • callback – an optional callback to be called when this command starts and finishes.

  • logger – a logger for this command to log with.

class ConfigureCommand[source]

A class for SKASubarray’s Configure() command.

__init__(command_tracker: CommandTrackerProtocol, component_manager: SubarrayComponentManager, callback: Callable[[bool], None] | None = None, logger: Logger | None = None, schema: dict[str, Any] | None = None) → None[source]

Initialise a new instance.

Parameters:

  • command_tracker – the device’s command tracker

  • component_manager – the device’s component manager

  • callback – an optional callback to be called when this command starts and finishes.

  • logger – a logger for this command to log with.

  • schema – an optional JSON schema for the command argument.

class ScanCommand[source]

A class for SKASubarray’s Scan() command.

__init__(command_tracker: CommandTrackerProtocol, component_manager: SubarrayComponentManager, callback: Callable[[bool], None] | None = None, logger: Logger | None = None, schema: dict[str, Any] | None = None) → None[source]

Initialise a new instance.

Parameters:

  • command_tracker – the device’s command tracker

  • component_manager – the device’s component manager

  • callback – an optional callback to be called when this command starts and finishes.

  • logger – a logger for this command to log with.

  • schema – an optional JSON schema for the command argument.

class EndScanCommand[source]

A class for SKASubarray’s EndScan() command.

__init__(command_tracker: CommandTrackerProtocol, component_manager: SubarrayComponentManager, callback: Callable[[bool], None] | None = None, logger: Logger | None = None) → None[source]

Initialise a new instance.

Parameters:

  • command_tracker – the device’s command tracker

  • component_manager – the device’s component manager

  • callback – an optional callback to be called when this command starts and finishes.

  • logger – a logger for this command to log with.

class EndCommand[source]

A class for SKASubarray’s End() command.

__init__(command_tracker: CommandTrackerProtocol, component_manager: SubarrayComponentManager, callback: Callable[[bool], None] | None = None, logger: Logger | None = None) → None[source]

Initialise a new instance.

Parameters:

  • command_tracker – the device’s command tracker

  • component_manager – the device’s component manager

  • callback – an optional callback to be called when this command starts and finishes.

  • logger – a logger for this command to log with.

class ObsResetCommand[source]

A class for SKASubarray’s ObsReset() command.

__init__(command_tracker: CommandTrackerProtocol, component_manager: SubarrayComponentManager, callback: Callable[[bool], None] | None = None, logger: Logger | None = None) → None[source]

Initialise a new instance.

Parameters:

  • command_tracker – the device’s command tracker

  • component_manager – the device’s component manager

  • callback – an optional callback to be called when this command starts and finishes.

  • logger – a logger for this command to log with.

class RestartCommand[source]

A class for SKASubarray’s Restart() command.

__init__(command_tracker: CommandTrackerProtocol, component_manager: SubarrayComponentManager, callback: Callable[[bool], None] | None = None, logger: Logger | None = None) → None[source]

Initialise a new instance.

Parameters:

  • command_tracker – the device’s command tracker

  • component_manager – the device’s component manager

  • callback – an optional callback to be called when this command starts and finishes.

  • logger – a logger for this command to log with.

init_command_objects() → None[source]

Set up the command objects.

execute_AssignResources(argin: str) → DevVarLongStringArrayType[source]

Assign resources to this subarray.

To modify behaviour for this command, modify the do() method of the command class.

Parameters:

argin – the resources to be assigned

Returns:

A tuple containing a result code and a string message. If the result code indicates that the command was accepted, the message is the unique ID of the task that will execute the command. If the result code indicates that the command was not excepted, the message explains why.

execute_ReleaseResources(argin: str) → DevVarLongStringArrayType[source]

Delta removal of assigned resources.

To modify behaviour for this command, modify the do() method of the command class.

Parameters:

argin – the resources to be released

Returns:

A tuple containing a result code and the unique ID of the command

execute_ReleaseAllResources() → DevVarLongStringArrayType[source]

Remove all resources to tear down to an empty subarray.

To modify behaviour for this command, modify the do() method of the command class.

Returns:

A tuple containing a result code and the unique ID of the command

execute_Configure(argin: str) → DevVarLongStringArrayType[source]

Configure the capabilities of this subarray.

To modify behaviour for this command, modify the do() method of the command class.

Parameters:

argin – JSON-encoded string with the scan configuration”, configuration specification

Returns:

A tuple containing a result code and the unique ID of the command

execute_Scan(argin: str) → DevVarLongStringArrayType[source]

Start scanning.

To modify behaviour for this command, modify the do() method of the command class.

Parameters:

argin – JSON-encoded string with the per-scan configuration

Returns:

A tuple containing a result code and the unique ID of the command

execute_EndScan() → DevVarLongStringArrayType[source]

End the scan.

To modify behaviour for this command, modify the do() method of the command class.

Returns:

A tuple containing a result code and the unique ID of the command

execute_End() → DevVarLongStringArrayType[source]

End the scan block.

To modify behaviour for this command, modify the do() method of the command class.

Returns:

A tuple containing a result code and the unique ID of the command

execute_ObsReset() → DevVarLongStringArrayType[source]

Reset the current observation process.

To modify behaviour for this command, modify the do() method of the command class.

Returns:

A tuple containing a result code and the unique ID of the command

execute_Restart() → DevVarLongStringArrayType[source]

Restart the subarray. That is, deconfigure and release all resources.

To modify behaviour for this command, modify the do() method of the command class.

Returns:

A tuple containing a result code and the unique ID of the command

ska_tango_base.testing.reference.reference_subarray_device.main(*args: str, **kwargs: str) → int[source]

Entry point for module.

Parameters:

  • args – positional arguments

  • kwargs – named arguments

Returns:

exit code