Reference Subarray Device

A reference implementation of an SKA subarray device.

It inherits from SKASubarray but provides schemas for some commands.

class SKASubarray(*args: Any, **kwargs: Any)[source]

Implements the SKA SubArray device.

class InitCommand(device: tango.server.Device, logger: Logger | None = None, validator: ArgumentValidator | None = None)[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.

class AssignResourcesCommand(command_tracker: CommandTracker, component_manager: SubarrayComponentManager, callback: Callable[[bool], None] | None = None, logger: Logger | None = None, schema: dict[str, Any] | None = None)[source]: A class for SKASubarray’s AssignResources() command.

class ReleaseResourcesCommand(command_tracker: CommandTracker, component_manager: SubarrayComponentManager, callback: Callable[[bool], None] | None = None, logger: Logger | None = None, schema: dict[str, Any] | None = None)[source]: A class for SKASubarray’s ReleaseResources() command.

class ReleaseAllResourcesCommand(command_tracker: CommandTracker, component_manager: SubarrayComponentManager, callback: Callable[[bool], None] | None = None, logger: Logger | None = None)[source]: A class for SKASubarray’s ReleaseAllResources() command.

class ConfigureCommand(command_tracker: CommandTracker, component_manager: SubarrayComponentManager, callback: Callable[[bool], None] | None = None, logger: Logger | None = None, schema: dict[str, Any] | None = None)[source]: A class for SKASubarray’s Configure() command.

class ScanCommand(command_tracker: CommandTracker, component_manager: SubarrayComponentManager, callback: Callable[[bool], None] | None = None, logger: Logger | None = None, schema: dict[str, Any] | None = None)[source]: A class for SKASubarray’s Scan() command.

class EndScanCommand(command_tracker: CommandTracker, component_manager: SubarrayComponentManager, callback: Callable[[bool], None] | None = None, logger: Logger | None = None)[source]: A class for SKASubarray’s EndScan() command.

class EndCommand(command_tracker: CommandTracker, component_manager: SubarrayComponentManager, callback: Callable[[bool], None] | None = None, logger: Logger | None = None)[source]: A class for SKASubarray’s End() command.

class AbortCommand(command_tracker: CommandTracker, component_manager: SubarrayComponentManager, callback: Callable[[bool], None], logger: Logger | None = None)[source]

A class for SKASubarray’s Abort() command.

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

Stateless hook for Abort() command functionality.

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.

class ObsResetCommand(command_tracker: CommandTracker, component_manager: SubarrayComponentManager, callback: Callable[[bool], None] | None = None, logger: Logger | None = None)[source]: A class for SKASubarray’s ObsReset() command.

class RestartCommand(command_tracker: CommandTracker, component_manager: SubarrayComponentManager, callback: Callable[[bool], None] | None = None, logger: Logger | None = None)[source]: A class for SKASubarray’s Restart() command.

create_component_manager() → ComponentManagerT[source]

Create and return a component manager for this device.

Raises:: NotImplementedError – because it is not implemented.

init_command_objects() → None[source]: Set up the command objects.

activationTime() → float

Read the time of activation in seconds since Unix epoch.

Returns:: Time of activation in seconds since Unix epoch.

assignedResources() → list[str]

Read the resources assigned to the device.

The list of resources assigned to the subarray.

Returns:: Resources assigned to the device.

configuredCapabilities() → list[str]

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

is_AssignResources_allowed() → bool[source]

Return whether the AssignResource command may be called in the current state.

Raises:: StateModelError – command not permitted in observation state
Returns:: whether the command may be called in the current device state

AssignResources(argin: str) → tuple[list[ResultCode], list[str]]

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.

is_ReleaseResources_allowed() → bool[source]

Return whether the ReleaseResources command may be called in current state.

Raises:: StateModelError – command not permitted in observation state
Returns:: whether the command may be called in the current device state

ReleaseResources(argin: str) → tuple[list[ResultCode], list[str]]

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

is_ReleaseAllResources_allowed() → bool[source]

Return whether ReleaseAllResources may be called in the current device state.

Raises:: StateModelError – command not permitted in observation state
Returns:: whether the command may be called in the current device state

ReleaseAllResources() → tuple[list[ResultCode], list[str]]

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

is_Configure_allowed() → bool[source]

Return whether Configure may be called in the current device state.

Raises:: StateModelError – command not permitted in observation state
Returns:: whether the command may be called in the current device state

Configure(argin: str) → tuple[list[ResultCode], list[str]]

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

is_Scan_allowed() → bool[source]

Return whether the Scan command may be called in the current device state.

Raises:: StateModelError – command not permitted in observation state
Returns:: whether the command may be called in the current device state

Scan(argin: str) → tuple[list[ResultCode], list[str]]

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

is_EndScan_allowed() → bool[source]

Return whether the EndScan command may be called in the current device state.

Raises:: StateModelError – command not permitted in observation state
Returns:: whether the command may be called in the current device state

EndScan() → tuple[list[ResultCode], list[str]]

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

is_End_allowed() → bool[source]

Return whether the End command may be called in the current device state.

Raises:: StateModelError – command not permitted in observation state
Returns:: whether the command may be called in the current device state

End() → tuple[list[ResultCode], list[str]]

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

is_Abort_allowed() → bool[source]

Return whether the Abort command may be called in the current device state.

Raises:: StateModelError – command not permitted in observation state
Returns:: whether the command may be called in the current device state

Abort() → tuple[list[ResultCode], list[str]]

Abort any long-running command such as Configure() or 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

is_ObsReset_allowed() → bool[source]

Return whether the ObsReset command may be called in the current device state.

Raises:: StateModelError – command not permitted in observation state
Returns:: whether the command may be called in the current device state

ObsReset() → tuple[list[ResultCode], list[str]]

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

is_Restart_allowed() → bool[source]

Return whether the Restart command may be called in the current device state.

Raises:: StateModelError – command not permitted in observation state
Returns:: whether the command may be called in the current device state

Restart() → tuple[list[ResultCode], list[str]]

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

main(*args: str, **kwargs: str) → int[source]

Entry point for module.

Parameters:

args – positional arguments
kwargs – named arguments

Returns:

exit code