Writing your first SKA Tango device using ska-tango-base

This page will guide you through the steps to writing your SKA Tango device based on the ska-tango-base package. This tutorial assumes you are already familiar with Tango, i.e. that you have come across Tango devices, attributes and commands before.

To serve as an example, the device we are going to build for this tutorial is called FileStats. The device will be configurable with a file path and will report statistics about the file at that path via Tango attributes. The device will also support two commands to manipulate the size of the configured file in order to demonstrate how to add “fast” and “long-running” commands.

The intention is for the example device to not require any specific domain knowledge to understand the business logic of the device and we can focus on learning what we need to know to build a SKA Tango device.

Step 0: Setup your environment

To follow along with this tutorial set up your development environment and install ska-tango-base package from the SKAO repository, along with the tutorial dependencies:

$ python3 -m pip --require-virtualenv install --extra-index-url https://artefact.skao.int/repository/pypi-all/simple ska-tango-base[tutorial]

Step 1: Decide on a Tango interface

The SKA Control Model defines several kinds of Tango devices that make up the control system. The ska-tango-base repository provides the following series of <X>Interface base classes that correspond to these kinds of Tango devices:

• BaseInterface

• ControllerInterface

• ObsInterface

• SubarrayInterface

Each interface defines a set of Tango commands and attributes that the particular kind of SKA Tango device should implement. Each interface is abstract in the sense that subclasses of this interface are expected to provide the behaviour of these commands and attributes by overriding methods. The interfaces form the following inheritance hierarchy with the BaseInterface, unsurprisingly, at the base of the interfaces:

For our FileStats device, we only need the BaseInterface. As this interface is the base of all the others, all SKA Tango devices need to implement it and the steps we follow here are going to be applicable to all SKA Tango devices.

Once you have selected the interface class that is appropriate for your device, you should inherit from this interface, in our case we have the following, which should be saved as file_stats_device.py:

Step 1: A BaseInterface device

import ska_tango_base as stb


class FileStats(stb.base.BaseInterface):
    """Tango device that exposes file statistics as attributes."""

Tip

If, for whatever reason, your device does not need to implement any of the Tango commands/attributes provided by the BaseInterface, it is recommended to at least use SKADevice. This will setup logging so that it can be consumed by the SKA infrastructure and also provides standard SKA attributes for version information. These will help your device fit into the SKA ecosystem.

Step 2: Starting our SKA Tango device

Now is a good time to get into the habit of launching our Tango device and connecting to it to see how it behaves.

Tip

While automated testing is an important practice when developing software, the purpose of this tutorial is to learn how to use ska-tango-base and there is nothing better for learning than getting your hands dirty with a Tango device. So, here we are going to focus on manual testing.

We need to be able to launch a Tango device server with our Tango device. A common way to do this is to add a __name__ == "__main__" block to our file_stats_device.py script, like so:

Step 2: Running our device server

import ska_tango_base as stb
import tango


class FileStats(stb.base.BaseInterface):
    """Tango device that exposes file statistics as attributes."""


if __name__ == "__main__":
    tango.server.run((FileStats,))

Now we can launch the device server in -nodb mode. Here we are specifying the port 12345 and creating a single FileStats device:

$ python file_stats_device.py tutorial -nodb -ORBendPoint giop:tcp::12345 -dlist tut/fs/1
1|2026-02-10T12:22:16.789Z|INFO|MainThread|set_logging_level|ska_device.py#224|tango-device:tut/fs/1|Logging level set to LoggingLevel.INFO on Python and Tango loggers
1|2026-02-10T12:22:16.789Z|INFO|MainThread|update_logging_handlers|logging.py#379|tango-device:tut/fs/1|Logging targets set to ['tango::logger']
Ready to accept request

Finally, from a separate itango terminal, we can create a tango.DeviceProxy to the running device. With this tango.DeviceProxy we can query what commands and attributes are defined for this device:

In [1]: dp = Device("tango://localhost:12345/tut/fs/1#dbase=no")

In [2]: dp.ping()
Out[2]: 346

In [3]: dp.get_command_list()
Out[3]: ['GetVersionInfo', 'Init', 'State', 'Status']

In [4]: dp.get_attribute_list()
Out[4]: ['buildState', 'versionId', 'loggingLevel', 'loggingTargets', 'adminMode', 'commandedState', 'healthState', 'healthInfo', 'State', 'Status']

Many of these attributes are provided by the BaseInterface and will require us to add code to our device to implement the behaviour of these attributes.

We can also read the device state and notice that we are “stuck” in DevState.INIT:

In [5]: dp.State()
Out[5]: <DevState.INIT: 9>

This is because the BaseInterface requires us to override the init_device() method.

Step 3: Overriding the init_device() method

The init_device() method is called automatically by Tango when the device is created and whenever a Tango client calls the built-in Init() command. The device is created at device server startup but also when a client calls either of the DevRestart() or RestartServer() admin device commands.

The SKA Control Model recommends that a Tango device pushes a state=DevState.INIT change event at the start of device initialisation and then pushes another event at the end of initialisation so that clients can be notified that the device is re-initialising. In order to achieve this, the ska_tango_base.base.base_interface.BaseInterface requires subclasses to override the init_device() method. When you override the BaseInterface.init_device() method, it is important that you:

• call super().init_device() at the start; and

• call init_completed() at the end

Warning

super().init_device() does much more than push the initial change event during initialisation, it is very important this is always called when you override init_device().

While we are at it, we should also provide information so that clients can query the device to determine what version of the software the device is running. This is achieved by setting _version_id and _build_state python attributes.

As is common practice at SKAO, we are going to store the version information in a release.py module, which for this tutorial you should save in the same directory as your file_stats_device.py module.

With this release module, we can add the version information to our init_device() method:

Step 3: Adding init_device()

import ska_tango_base as stb
import tango
import release


class FileStats(stb.base.BaseInterface):
    """Tango device that exposes file statistics as attributes."""

    def init_device(self) -> None:
        """Initialise the device."""
        super().init_device()

        self._version_id = release.version
        self._build_state = f"{release.name} {release.version}: {release.description}"

        self.init_completed()


if __name__ == "__main__":
    tango.server.run((FileStats,))

Now, after we restart the file_stats_device.py script, we can query the version information from the device via the versionId and buildState Tango attributes and via the GetVersionInfo() Tango command:

In [1]: dp = Device("tango://localhost:12345/tut/fs/1#dbase=no")

In [2]: dp.versionId
Out[2]: '0.1.0'

In [3]: dp.buildState
Out[3]: 'ska-tut-file-stats 0.1.0: A tutorial SKA Tango device.'

In [4]: dp.GetVersionInfo()
Out[4]: ['FileStats, ska-tut-file-stats 0.1.0: A tutorial SKA Tango device.']

We can also subscribe to change events for the state attribute and observe that events are pushed when we initialise the device:

In [5]: sid = dp.subscribe_event("state", tango.EventType.CHANGE_EVENT, tango.utils.EventCallback())
2026-02-10 12:26:05.196861 TUT/FS/1 STATE CHANGE SUBSUCCESS [ATTR_VALID] DISABLE

In [6]: dp.Init()

2026-02-10 12:26:10.296456 TUT/FS/1 STATE CHANGE UPDATE [ATTR_VALID] INIT
2026-02-10 12:26:10.296552 TUT/FS/1 STATE CHANGE UPDATE [ATTR_VALID] DISABLE

Notice here that the BaseInterface has placed us into DevState.DISABLE after we have called init_completed(). We will discuss what the SKA Control Model says about this state below, but it doesn’t sound like a state we want our device to be in.

Our device is in DevState.DISABLE because our device initially has an adminMode attribute with a value of OFFLINE. We need to move our adminMode to ONLINE to instruct our Tango device that we wish to use it.

Tip

The adminMode is a memorised Tango attribute, meaning its (write) value is saved to the Tango database whenever it is written to by a client. This means that if we restart the device while it is in AdminMode.ONLINE it will come back in AdminMode.ONLINE. However, in our case we are starting the device in -nodb mode so there is no database to save the value to and our device will always start in AdminMode.OFFLINE.

Let’s try setting the adminMode:

In [7]: dp.adminMode
Out[7]: <adminMode.OFFLINE: 1>

In [8]: dp.adminMode = "ONLINE"
PyDs_PythonError: NotImplementedError: 'change_control_level' method must be implemented by 'FileStats'. The parent 'BaseInterface' is an abstract base class.

(For more detailed information type: tango_error)

This doesn’t work because we need to implement the behaviour of the adminMode for our device by overriding the change_control_level() method.

Step 4: Implementing the adminMode attribute

When a device is in AdminMode.OFFLINE the SKA Control Model requires that the device does not communicate with its “component” and reports its state as DevState.DISABLE. Whereas, when a device is in AdminMode.ONLINE, it should be monitoring and controlling its component. This means when a client transitions our device from AdminMode.OFFLINE to AdminMode.ONLINE we need to start monitoring and controlling some component and the BaseInterface notifies us of this by calling the change_control_level() method which we are responsible for overriding in order to start monitoring.

But hang on, what’s a component?

For the SKA Control Model, the “component” is just whatever the Tango device is responsible for monitoring and controlling, for example, it could be a piece of hardware such as a motor, it could be some other subordinate Tango devices or it could be a piece of software responsible for doing some number crunching. It could be almost anything and depends on the device in question. For our FileStats device the system under control is whichever file we are monitoring on the file system.

Note

In other SKA documentation the “component” is often referred to as the “system under control”. In this tutorial we use the term “component” as historically this is the term that has been used in ska-tango-base and is reflected in the ska-tango-base API.

If we are going to start monitoring a file, we need to know which file to monitor. Let’s add a Tango device property for the file path, being sure to provide a default value to make it easier to test. Let’s also override the change_control_level() method to start monitoring this path when we are instructed:

Step 4: Implementing adminMode

class FileStats(stb.base.BaseInterface):
    """Tango device that exposes file statistics as attributes."""

    FilePath = tango.server.device_property(
        dtype="DevString", default_value="dummy", doc="Path of file to monitor"
    )

    def init_device(self) -> None:
        """Initialise the device."""
        super().init_device()

        self._version_id = release.version
        self._build_state = f"{release.name} {release.version}: {release.description}"

        self.init_completed()

    def change_control_level(self, control_level: stb.base.ControlLevel) -> None:
        """Change how the device is interacting with the system under control."""
        if control_level == stb.base.ControlLevel.NO_CONTACT:
            self.component_disconnected()
        elif control_level == stb.base.ControlLevel.FULL_CONTROL:
            self.component_on()
        else:
            raise ValueError(f"Unknown control_level {control_level}")

The change_control_level() method accepts a ControlLevel() parameter which we can use to determine whether we should start or stop monitoring our component. As we start and stop monitoring, we are responsible for updating the state of our Tango device to match the state that we find the component in. The BaseInterface uses an internal state machine (an instance of OpStateModel) to manage the device’s state. We can drive this state machine by calling the various component_<x>() methods which will “perform an action” on the state machine.

For our implementation of change_control_level(), we perform the component_disconnected() action transitioning the device to DevState.DISABLE when we are instructed to stop monitoring (ControlLevel.NO_CONTACT). When we are instructed to monitor the component (ControlLevel.FULL_CONTROL) we transition to DevState.ON via the component_on() action to signal that we are ready to gather some file statistics.

Note

The SKA Control Model defines the DevState.ON, DevState.STANDBY and DevState.OFF states in terms of the monitored power consumption of the component. As there is no power consumption that we can monitor for a file, our FileStats device should just use the DevState.ON state and ignore the others. The SKA Control Model also defines optional On(), Standby() and Off() commands to transition the power consumption of the component. Again our FileStats device should ignore these commands as they are not applicable. Similarly, our FileStats should ignore the commandedState attribute as it does not implement commands to transition the state.

Let’s restart our file_stats_device.py script and check that our Tango device transitions to DevState.ON now that we have implemented the adminMode:

In [1]: dp = Device("tango://localhost:12345/tut/fs/1#dbase=no")

In [2]: sid = dp.subscribe_event("state", tango.EventType.CHANGE_EVENT, tango.utils.EventCallback())
2026-02-10 15:09:10.926130 TUT/FS/1 STATE CHANGE SUBSUCCESS [ATTR_VALID] DISABLE

In [3]: dp.adminMode = "ONLINE"

2026-02-10 15:09:13.383190 TUT/FS/1 STATE CHANGE UPDATE [ATTR_VALID] ON

Great! We are now ready to add our own attributes to expose metadata about the configured file and start “communicating” with our component.

Step 5: Implementing our own attributes

We are going to implement some attributes to expose the file’s size, owner, mode (permissions) and last modified time. We will expose most of these as strings in a similar format to the ls -l command.

To separate concerns somewhat, let’s implement a Metadata class that is responsible for querying the file system for information about the file and converting the data into a string when required.

Step 5a: Adding Metadata

import os
import pwd
import grp
import time
import stat
import logging


class Metadata:
    """File metadata."""

    def __init__(self, path: str, logger: logging.Logger) -> None:
        self.path = os.path.abspath(path)
        self.logger = logger
        self.data: os.stat_result | None = None
        self.timestamp: float | None = None

    def reset(self) -> None:
        """Reset metadata to None."""
        self.data = None
        self.timestamp = None

    def refresh(self) -> None:
        """Refresh metadata."""
        self.data = os.stat(self.path)
        self.timestamp = time.time()

    @property
    def size(self) -> int | None:
        """Return the size of the file in bytes."""
        if self.data is not None:
            return self.data.st_size
        return None

    @property
    def mode(self) -> str | None:
        """
        Return the mode (permission) of the file.

        :return: mode in the same format as ``ls -l``.
        """
        if self.data is not None:
            return stat.filemode(self.data.st_mode)
        return None

    @property
    def owner(self) -> str | None:
        """
        Return the owner of the file.

        :return: has the format "<user>:<group>"
        """
        if self.data is not None:
            user = pwd.getpwuid(self.data.st_uid)
            group = grp.getgrgid(self.data.st_gid)
            return f"{user.pw_name}:{group.gr_name}"
        return None

    @property
    def last_modified_time(self) -> str | None:
        """
        Return the time the file was last modified.

        :return: last modified time in the ``ctime`` format
        """
        if self.data is not None:
            return time.ctime(self.data.st_mtime)
        return None

Separating application specific logic out from the Tango device is generally a good idea to allow unit testing of this functionality and to allow the Tango device to focus on providing a Tango interface. Here, we separated the file system query (i.e. the os.stat() call in refresh()) from the data conversion (i.e. the various properties). We also allow a disengaged state (where data is None) so that the Tango device can support DevState.DISABLE and not “communicate” with its component.

Note

Historically, ska-tango-base has forced Tango devices to follow this pattern by requiring them to implement application specific logic in a “component manager”. While this design pattern is a good one in the abstract, experience has shown that the exact line where concern should be separated is application specific and drawing this line in a foundational library such as ska-tango-base is too opinionated. It is better for each application to decide how to implement this design pattern for their specific case. ska-tango-base still supports the component manager approach via the SKABaseDevice class and its subclasses, however, it is not recommended to use this approach for new devices.

Now let’s use this Metadata class to implement our attributes:

Step 5b: Using Metadata

class FileStats(stb.base.BaseInterface):
    """Tango device that exposes file statistics as attributes."""

    FilePath = tango.server.device_property(
        dtype="DevString", default_value="dummy", doc="Path of file to monitor"
    )

    def init_device(self) -> None:
        """Initialise the device."""
        super().init_device()

        self._version_id = release.version
        self._build_state = f"{release.name} {release.version}: {release.description}"
        self.metadata = Metadata(self.FilePath, self.logger)

        self.init_completed()

    def change_control_level(self, control_level: stb.base.ControlLevel) -> None:
        """Change how the device is interacting with the system under control."""
        if control_level == stb.base.ControlLevel.NO_CONTACT:
            self.metadata.reset()
            self.component_disconnected()
        elif control_level == stb.base.ControlLevel.FULL_CONTROL:
            self.component_on()
        else:
            raise ValueError(f"Unknown control_level {control_level}")

    def read_attr_hardware(self, attr_list: list[int]) -> None:
        """Prepare for attribute read."""
        _ = attr_list
        if self.get_state() != tango.DevState.DISABLE:
            self.metadata.refresh()

    @tango.server.attribute(dtype=int)
    def size(self) -> stb.type_hints.ReadAttrType[int]:
        """Read the size of the file in bytes."""
        return self._to_read_attr_type(self.metadata.size, 0)

    @tango.server.attribute(dtype=str)
    def lastModifiedTime(self) -> stb.type_hints.ReadAttrType[str]:
        """Return the time the file was last modified."""
        return self._to_read_attr_type(self.metadata.last_modified_time, "")

    @tango.server.attribute(dtype=str)
    def owner(self) -> stb.type_hints.ReadAttrType[str]:
        """Return the owner of the file."""
        return self._to_read_attr_type(self.metadata.owner, "")

    @tango.server.attribute(dtype=str)
    def mode(self) -> stb.type_hints.ReadAttrType[str]:
        """Return the mode (permissions) of the file."""
        return self._to_read_attr_type(self.metadata.mode, "")

    def _to_read_attr_type(
        self, value: typing.Any, default: typing.Any
    ) -> stb.type_hints.ReadAttrType[typing.Any]:
        if value is None:
            return default, time.time(), tango.AttrQuality.ATTR_INVALID

        return value, self.metadata.timestamp, tango.AttrQuality.ATTR_VALID

Here, we are using read_attr_hardware() to refresh the metadata so that a client will receive data from the same os.stat() call if they read multiple attributes in the same read_attributes() call. We use the same timestamp for each attribute to make it clearer that they have come from the same os.stat() call. We are also using a helper method _to_read_attr_type() to ensure that we report the attribute quality factor as invalid if we do not have any data.

Let’s restart the file_stats_device.py script, and have a go at querying file metadata with our Tango device:

In [1]: dp = Device("tango://localhost:12345/tut/fs/1#dbase=no")

In [2]: print(dp.size)
None

In [3]: dp.adminMode = "ONLINE"

In [4]: dp.size
PyDs_PythonError: FileNotFoundError: [Errno 2] No such file or directory: '/home/tri/skao/ska-tango-base/docs/src/tutorials/create-tango-device/dummy'

(For more detailed information type: tango_error)

In [5]: !touch dummy

In [6]: dp.size
Out[6]: 0

In [7]: !head -c128 /dev/urandom >> dummy

In [8]: dp.size
Out[8]: 128

In [9]: dp.owner
Out[9]: 'tri:tri'

In [10]: dp.mode
Out[10]: '-rw-r--r--'

In [11]: dp.lastModifiedTime
Out[11]: 'Tue Feb 10 15:26:38 2026'

In [12]: print(f"{dp.mode} {dp.owner} {dp.size} {dp.lastModifiedTime}")
-rw-r--r-- tri:tri 128 Tue Feb 10 15:26:38 2026

Great! A client can request the file metadata, but unfortunately, we cannot subscribe to change events for these attributes:

In [13]: sid = dp.subscribe_event("size", tango.EventType.CHANGE_EVENT, tango.utils.EventCallback())
API_AttributePollingNotStarted: The polling (necessary to send events) for the attribute size is not started
(For more detailed information type: tango_error)

The standard way to solve this problem with Tango would be to enable polling, which can be done by configuring the Tango device. However, there are two issues with this approach which means we avoid using the built-in Tango polling at SKAO:

1. The Tango polling loop does not cope well when running in a resource constrained environment such as a kubernetes cluster. See Tango device server polling loop considered harmful.

2. Polling introduces latency so it should be avoided if possible. If the component is capable of generating events itself then this should be preferred.

Fortunately, the watchfiles python package (installed as a tutorial dependency) can provide events whenever our file changes allowing us to manually push Tango events with new metadata. This avoids the need to use polling for this device.

Tip

If your component does not support generating events itself, you should use Poller rather than the built in Tango polling loop to avoid the issue 1 above. Team Wombat is working with the wider Tango community to improve the Tango polling loop so that we can use it at SKA.

Step 6: Implementing event driven monitoring

In order for our Tango device to get notified that the file we are monitoring has changed, we need a dedicated background thread to receive these notifications. Let’s start by adding a method, monitor_for_updates() to our Metadata class that can refresh the metadata whenever we receive a change notification from watchfiles:

Step 6a: Watching files

import watchfiles
import threading

# Disable watchfiles info logs as they can be quite noisy
logging.getLogger("watchfiles.main").setLevel(logging.WARNING)


class Metadata:
    def monitor_for_updates(self, stop_event: threading.Event) -> None:
        """Setup monitoring for file metadata changes."""
        parent = os.path.dirname(self.path)

        def only_filename(change: watchfiles.Change, name: str) -> bool:
            _ = change
            return os.path.abspath(name) == self.path

        # Monitor from parent directory to know if the file is created or deleted.
        # TODO: Support missing parent directory
        watch = watchfiles.watch(
            parent, watch_filter=only_filename, stop_event=stop_event
        )

        # Refresh after we start watch to ensure we are up-to-date and
        # don't miss a change
        self.refresh()

        self.logger.debug("Watching %s", parent)
        for changes in watch:
            self.logger.debug("Got changes=%s", changes)
            self.refresh()

Here, we setup the watch(), filtering out events not for our file, before refreshing the metadata and waiting for notifications. This ensures that we do not miss any changes. Once we get notified of changes, we don’t look at what the changes are and instead just refresh the metadata. We accept a stop_event parameter to pass on watch() so that the monitoring can be interrupted on request.

Note

As is outlined in the TODO, this implementation of monitor_for_updates() does not handle missing parent directories. If FileStats was a real SKA Tango device, this would need resolving, however, we have omitted this here to keep the implementation simple so that we can focus on the generally applicable behaviour of the Tango device.

Next, let’s spawn a thread from our Tango device to call monitor_for_updates() whenever we change control level:

Step 6b: Spawning a thread

class FileStats(stb.base.BaseInterface):
    """Tango device that exposes file statistics as attributes."""

    FilePath = tango.server.device_property(
        dtype="DevString", default_value="dummy", doc="Path of file to monitor"
    )

    def init_device(self) -> None:
        """Initialise the device."""
        super().init_device()

        self._version_id = release.version
        self._build_state = f"{release.name} {release.version}: {release.description}"
        self.metadata = Metadata(self.FilePath, self.logger)

        self.stop_event: threading.Event | None = None
        self.monitor_thread: threading.Thread | None = None

        self.init_completed()

    def delete_device(self) -> None:
        """De-initialise device."""
        self._stop_monitor_thread()
        super().delete_device()

    def change_control_level(self, control_level: stb.base.ControlLevel) -> None:
        """Change how the device is interacting with the system under control."""
        if control_level == stb.base.ControlLevel.NO_CONTACT:
            self._stop_monitor_thread()
            self.metadata.reset()
            self.component_disconnected()
        elif control_level == stb.base.ControlLevel.FULL_CONTROL:
            self.component_on()
            self._start_monitor_thread()
        else:
            raise ValueError(f"Unknown control_level {control_level}")

    def _start_monitor_thread(self) -> None:
        def target(device: FileStats) -> None:
            try:
                assert device.stop_event is not None
                device._status = f"Monitoring '{self.FilePath}'"
                device.metadata.monitor_for_updates(device.stop_event)
            except Exception as ex:
                device.logger.exception("File monitor raised unexpected exception")
                device.component_fault()
                device._status = f"Monitor thread crashed: {ex}"
                return
            device._status = f"Monitoring disabled"

        self.stop_event = threading.Event()
        self.monitor_thread = threading.Thread(target=target, args=(self,))
        self.monitor_thread.start()

    def _stop_monitor_thread(self) -> None:
        if self.stop_event is not None:
            assert self.monitor_thread is not None

            self.stop_event.set()
            self.monitor_thread.join()

            self.monitor_thread = None
            self.stop_event = None

Here, we make sure to also shutdown the thread in our delete_device() method. The thread target function wraps the call to monitor_for_updates() in a try/except statement defensively to ensure that any issues not handled by the monitor_for_updates() method are logged and the device is transitioned to the DevState.FAULT state. This state indicates that the device has encountered an error it is not going to recover from automatically and requires an operator to intervene. The operator can try to resolve this by calling the Init() to reinitialise the device. In general, requiring manual intervention like this should be considered a bug in the SKA Tango device, or at least that the SKA Tango device is not being used as intended, and should be avoided. However, it is good practice to be defensive in a scenario such as spawning a thread so that if a problem occurs it is reported and an operator knows that something fishy has occurred.

Warning

Any resources you allocate in your init_device() method (such as threads, sockets, open files, etc.) _must_ be cleaned up in a delete_device() method, otherwise the built-in Init() command will not work reliably. Also, when overriding the delete_device() method, you _must_ call super().delete_device() to clean up any resources a super class has allocated.

Initialising your SKA Tango device has more guidelines for managing the lifecycle of your device.

Tip

Whenever your SKA Tango device transitions to DevState.FAULT, you should provide a status describing what has failed. This can be done by assigning to the _status signal which will update the Tango built-in status attribute. You also need to make sure the status gets cleared when the device no longer has DevState.FAULT.

Now that we have a thread that can monitor our file, we need that thread to be able to push data to the Tango device to be sent on as Tango events. The BaseInterface class provides a signal bus mechanism to facilitate this. The Metadata class can emit signals on the bus which can then be listened for by the Tango device to push events for the corresponding attributes. The signal processing happens on a background thread managed by the BaseInterface. The advantage of this approach is that it decouples the Metadata from the Tango device and avoids our monitoring background thread from having to grab the Tango device lock at any point.

Let’s add some AttrSignal to our Metadata class and replace our existing attributes with ska_tango_base.software_bus.attribute_from_signal objects pointing to the Metadata signals:

Step 6c: Hooking up signals

class FileStats(stb.base.BaseInterface):
    """Tango device that exposes file statistics as attributes."""

    def on_new_shared_bus(self) -> None:
        """Create sharing observers."""
        super().on_new_shared_bus()
        self.metadata = Metadata(self.FilePath, self.logger)

    size = stb.software_bus.attribute_from_signal(
        "metadata.size", dtype=int, abs_change=1
    )
    lastModifiedTime = stb.software_bus.attribute_from_signal(
        "metadata.last_modified_time", dtype=str
    )
    owner = stb.software_bus.attribute_from_signal("metadata.owner", dtype=str)
    mode = stb.software_bus.attribute_from_signal("metadata.mode", dtype=str)


class Metadata(stb.software_bus.SharingObserver):
    """File metadata."""

    size_signal = stb.software_bus.AttrSignal[int](name="size")
    mode_signal = stb.software_bus.AttrSignal[str](name="mode")
    owner_signal = stb.software_bus.AttrSignal[str](name="owner")
    last_modified_time_signal = stb.software_bus.AttrSignal[str](
        name="last_modified_time"
    )

Here, we moved the initialisation of the Metadata object to the on_new_shared_bus() method as it is now a SharingObserver and we need it to be available to share the bus with it.

Finally, let’s actually emit values on each of these signals whenever we refresh() or reset() the metadata:

Step 6d: Pushing signals

class FileStats(stb.base.BaseInterface):
    """Tango device that exposes file statistics as attributes."""

    def read_attr_hardware(self, attr_list: list[int]) -> None:
        """Prepare for attribute read."""
        _ = attr_list
        if self.get_state() != tango.DevState.DISABLE:
            self.metadata.refresh()
            with self.allow_internal_threads():
                self.shared_bus.wait_for_thread()


class Metadata(stb.software_bus.SharingObserver):
    """File metadata."""

    def reset(self) -> None:
        """Reset metadata to None."""
        self.data = None
        self.timestamp = None

        self.logger.debug(f"Resetting metadata")
        self._emit_signals()

    def refresh(self) -> None:
        """Refresh metadata."""
        self.data = os.stat(self.path)
        self.timestamp = time.time()

        self.logger.debug("Refreshing metadata: %s", self.data)
        self._emit_signals()

    def _emit_signals(self) -> None:
        if self.timestamp is not None:
            self.size_signal = (
                typing.cast(int, self.size),
                self.timestamp,
                tango.AttrQuality.ATTR_VALID,
            )
            self.mode_signal = (
                typing.cast(str, self.mode),
                self.timestamp,
                tango.AttrQuality.ATTR_VALID,
            )
            self.owner_signal = (
                typing.cast(str, self.owner),
                self.timestamp,
                tango.AttrQuality.ATTR_VALID,
            )
            self.last_modified_time_signal = (
                typing.cast(str, self.last_modified_time),
                self.timestamp,
                tango.AttrQuality.ATTR_VALID,
            )
        else:
            self.size_signal = None
            self.mode_signal = None
            self.owner_signal = None
            self.last_modified_time_signal = None

Here, we assign None to each signal if we do not have any data. This will instruct the ska_tango_base.software_bus.attribute_from_signal objects to clear any data they have stored and return a ATTR_INVALID result if a client reads the attribute.

We also call BusProtocol.wait_for_thread in our read_attr_hardware() method to ensure that the signals emitted by the call to refresh() have been processed before we read any attributes as the ska_tango_base.software_bus.attribute_from_signal will not return the new values until the corresponding events have been pushed.

Warning

When an attribute is updated via a signal, the actual update happens asynchronously on a background thread. If the order of operations is important, you must ensure that all attribute updates happen on this background thread by using, for example, attribute_from_signal; or you must ensure that the updates are synchronised with the thread by using wait_for_thread() as we do in our read_attr_hardware().

Now, after we have restarted the file_stats_device.py script, we can receive events from the Tango device as the file changes:

In [1]: dp = Device("tango://localhost:12345/tut/fs/1#dbase=no")

In [2]: sid = dp.subscribe_event("size", tango.EventType.CHANGE_EVENT, tango.utils.EventCallback())
2026-02-10 20:37:04.617333 TUT/FS/1 SIZE CHANGE SUBSUCCESS [ATTR_INVALID] None

In [3]: sid2 = dp.subscribe_event("lastModifiedTime", tango.EventType.CHANGE_EVENT, tango.utils.EventCallback())
2026-02-10 20:37:13.877250 TUT/FS/1 LASTMODIFIEDTIME CHANGE SUBSUCCESS [ATTR_INVALID] None

In [4]: dp.adminMode = "ONLINE"

2026-02-10 20:37:23.244046 TUT/FS/1 SIZE CHANGE UPDATE [ATTR_VALID] 128
2026-02-10 20:37:23.244046 TUT/FS/1 LASTMODIFIEDTIME CHANGE UPDATE [ATTR_VALID] Tue Feb 10 15:26:38 2026
In [5]: !head -c128 /dev/urandom >> dummy

2026-02-10 20:37:33.753628 TUT/FS/1 SIZE CHANGE UPDATE [ATTR_VALID] 256
2026-02-10 20:37:33.753628 TUT/FS/1 LASTMODIFIEDTIME CHANGE UPDATE [ATTR_VALID] Tue Feb 10 20:37:33 2026

The monitoring side of our Tango device is now almost complete, but there are two attributes provided by BaseInterface that we have not yet implemented: healthState and healthInfo. Let’s try reading them now:

In [6]: dp.healthState
Out[6]: <healthState.FAILED: 2>

In [7]: dp.healthInfo
Out[7]: ('Device implementation has not provided a health report',)

The healthInfo attribute has told us that our device implementation needs to provide a health report. So let’s do that.

Step 7: Providing health reports

The healthState of a SKA Tango device should be used to indicate whether the Tango device is capable of performing its functionality. The healthInfo attribute consists of a list of strings used to describe why we are in a non-OK health state. Together, these two attributes are called the health report of the device and it is our job to provide health reports via the report_health() method. For our FileStats device, we should use this method to communicate any issues encountered by the monitoring thread, notably, if we are unable to call os.stat() on the configured file.

In order for our Metadata class to asynchronously send exceptions to the Tango device, let’s introduce another signal, stat_failed, and then hook it up to the refresh() and reset() methods:

Step 7a: Pushing exceptions

class Metadata(stb.software_bus.SharingObserver):
    """File metadata."""

    stat_failed = stb.software_bus.Signal[Exception | None]()

    def reset(self) -> None:
        """Reset metadata to None."""
        self.data = None
        self.timestamp = None
        self.stat_failed = None

        self.logger.debug(f"Resetting metadata")
        self._emit_signals()

    def refresh(self) -> None:
        """Refresh metadata.

        If refreshing fails, the metadata is set to None.
        """
        try:
            self.data = os.stat(self.path)
            self.timestamp = time.time()
            self.stat_failed = None
        except OSError as ex:
            self.data = None
            self.timestamp = None
            self.stat_failed = ex

        self.logger.debug("Refreshing metadata: %s", self.data)
        self._emit_signals()

Then, let’s add a signal listener to the Tango device that can update the healthState whenever this new signal is emitted:

Step 7b: Receiving exceptions

class FileStats(stb.base.BaseInterface):
    """Tango device that exposes file statistics as attributes."""

    @stb.software_bus.listen_to_signal("metadata.stat_failed")
    def __on_stat_failed(self, ex: Exception | None) -> None:
        if ex is None:
            self.report_health(scm.HealthState.OK, [])
        else:
            self.report_health(scm.HealthState.FAILED, [f"stat('{self.FilePath}'): {ex}"])

Here, we are supplying an empty health_info list to report_health() when the health_state is OK. This is required, the health_info must be provided for DEGRADED and FAILED health states and must be empty for OK. Our FileStats device is relatively simple, so there is only ever one issue to include in the health report at a time. For more complex devices there may be multiple issues which should all be reported in the health info.

Note

The UNKNOWN should no longer be used.

And finally, let’s update the healthState in our monitor thread target function when we encounter any unexpected problem:

Step 7c: Update healthState

class FileStats(stb.base.BaseInterface):
    """Tango device that exposes file statistics as attributes."""

    def _start_monitor_thread(self) -> None:
        def target(device: FileStats) -> None:
            try:
                assert device.stop_event is not None
                device._status = f"Monitoring '{self.FilePath}'"
                device.metadata.monitor_for_updates(device.stop_event)
            except Exception as ex:
                device.logger.exception("File monitor raised unexpected exception")
                device.component_fault()
                self.report_health(scm.HealthState.FAILED, [f"Device in fault: {ex}"])
                device._status = f"Monitor thread crashed: {ex}"
                return
            device._status = f"Monitoring disabled"

Tip

Whenever a SKA Tango device transitions to DevState.FAULT it should also transition to HealthState.FAILED. The healthState should be used to indicate whether the device can fulfil its core functionality, if the Tango device has encountered an unexpected error (as indicated by DevState.FAULT) then it cannot fulfil its core functionality and therefore should report HealthState.FAILED. However, the converse is not true. DevState.FAULT should only be used if the SKA Tango device knows for certain that operator intervention is required for the Tango device itself to recover when the underlying component recovers.

For our FileStats device, the configured file being missing does not warrant DevState.FAULT because if the file appears (via miracle or otherwise) the Tango device will detect this and the healthState will recover to HealthState.OK. However, if the _parent_ directory was missing then DevState.FAULT is warranted because the monitor thread crashes if the parent directory is missing (because we did not implement this for simplicity’s sake) and if the parent directory is created, the Tango device will not notice. It requires an operator to call the Init() command after the directory has been created.

With all this in place, let’s restart the file_stats_device.py script and try out our new healthState:

In [1]: dp = Device("tango://localhost:12345/tut/fs/1#dbase=no")

In [2]: import ska_control_model as scm

In [3]: sid = dp.subscribe_event("healthState", tango.EventType.CHANGE_EVENT, lambda ev: print(scm.HealthState(ev.attr_value.value)))
HealthState.FAILED

In [4]: sid2 = dp.subscribe_event("healthInfo", tango.EventType.CHANGE_EVENT, tango.utils.EventCallback())
2026-03-27 13:18:24.534954 TUT/FS/1 HEALTHINFO CHANGE SUBSUCCESS [ATTR_VALID] ('Device implementation has not provided a health report',)

In [5]: dp.adminMode = "ONLINE"
2026-03-27 13:19:20.462533 TUT/FS/1 HEALTHINFO CHANGE UPDATE [ATTR_VALID] ()
HealthState.OK

In [6]: !rm dummy
2026-03-27 13:19:27.111892 TUT/FS/1 HEALTHINFO CHANGE UPDATE [ATTR_VALID] ("stat('dummy'): [Errno 2] No such file or directory: '/path/to/dummy'",)
HealthState.FAILED

In [7]: !touch dummy
2026-03-27 13:19:35.976478 TUT/FS/1 HEALTHINFO CHANGE UPDATE [ATTR_VALID] ()
HealthState.OK

With health reporting now implemented, our FileStats device can monitor the stats on a file and report problems according to the SKA Control Model. However, our Tango device does not perform any _control_ of its component. Let’s add some commands to explore how control is handled for SKA Tango devices.

Step 8: Implementing commands

At SKA we talk about two different kinds of Tango commands which we call fast commands and long running commands respectively. As the names suggest, so-called fast commands are standard Tango commands that complete quickly (the SKA CS Guidelines suggest <10ms as the cut off). When a client is invoking a fast command, it is assumed that the client will block to wait for a response, as such it is important that these commands really are fast.

However, it is not just the client that calls the command that suffers if a fast command is slow. Due to the use of the default “SYNC mode” used by SKA Tango devices, a lock is held while each Tango command is executed such that no other client can interact with the Tango device while the Tango command is executing. This means that a slow “fast command” locks the Tango device from _all_ other clients. Moreover, while in “SYNC mode” the Tango device is unable to push events while this command is executing.

To alleviate these issues with Tango commands that take too long, at SKA we have devised the “long running command” protocol, built on top of the Tango event system. The long running command mechanism allows the Tango device to release the lock while it services a command. This allows other clients to use the Tango device while this command is being executed. More information about long running commands can be found at Long running commands.

Our FileStats device is going to implement two commands Shrink() and Grow() to allow clients to manipulate the size of the file. Shrinking a file is a relatively quick operation as it just requires updating the metadata so we will implement Shrink() as a SKA fast command. Conversely, we are going to require that the Grow() command does not return until the data has been committed to disk (via fsync(2)), so this we will implement it as a long running command.

For both these commands, we are going to require that they really are shrinking or growing the file otherwise we will refuse to execute them. To support this we are going to need to know the size of the file before we grow or shrink it. This should be handled by our Metadata class and indeed we can query the current size of the file using the size property. However, if we cannot determine the size of file for whatever reason we just get back None and it would be nice to provide some details to the caller of the command if we cannot determine the size of the file. Let’s begin by adding a get_size_or_raise() method to the Metadata class that solves this problem for us:

Step 8a: Storing exceptions

class Metadata(stb.software_bus.SharingObserver):
    """File metadata."""

    stat_failed = stb.software_bus.Signal[Exception | None](stored=True)

    def __init__(self, path: str, logger: logging.Logger) -> None:
        self.path = os.path.abspath(path)
        self.logger = logger
        self.data: os.stat_result | None = None
        self.timestamp: float | None = None
        self.lock = threading.Lock()

    def reset(self) -> None:
        """Reset metadata to None."""
        with self.lock:
            self.data = None
            self.timestamp = None
            self.stat_failed = None

            self.logger.debug(f"Resetting metadata")
            self._emit_signals()

    def refresh(self) -> None:
        """Refresh metadata.

        If refreshing fails, the metadata is set to None.
        """
        with self.lock:
            try:
                self.data = os.stat(self.path)
                self.timestamp = time.time()
                self.stat_failed = None
            except OSError as ex:
                self.data = None
                self.timestamp = None
                self.stat_failed = ex

            self.logger.debug("Refreshing metadata: %s", self.data)
            self._emit_signals()

    def get_size_or_raise(self) -> int:
        """
        Return the size of the file.

        If the size is not available raises the exception that was generated
        when the last stat failed.

        :return: size of file in bytes
        :raises Exception: if size not available
        """

        with self.lock:
            size = self.size
            if size is not None:
                return size

            ex = self.stat_failed
            if ex is not None:
                raise ex

        raise ValueError(
            f"Unable to determine size of '{self.path}' for unknown reason."
        )

Here, we have passed stored=True to the stat_failed signal so that the last value emitted is kept around and we can use it later. We have introduced a lock to ensure that the exception and the data are updated atomically.

Implementing a fast command

Armed with our new get_size_or_raise() method, let’s implement a Shrink() command using os.truncate():

Step 8b: Shrink

class FileStats(stb.base.BaseInterface):
    """Tango device that exposes file statistics as attributes."""

    def is_Shrink_allowed(self) -> bool:
        """Return true if the Shrink command is allowed."""
        return typing.cast(bool, self.get_state() != tango.DevState.DISABLE)

    @tango.server.command(dtype_out="DevVarLongStringArray")
    def Shrink(self, new_size: int) -> stb.type_hints.DevVarLongStringArrayType:
        """
        Shrink the file to the provided size.

        :param new_size: size to shrink to
        :return: ResultCode.OK, message
        :raises ValueError: if the new file size is greater than the current size or
                            the current size is unknown
        """

        current_size = self.metadata.get_size_or_raise()

        if new_size > current_size:
            raise ValueError(
                f"New size '{new_size}' is greater than current size '{current_size}'. "
                "Cannot shrink to new size."
            )

        os.truncate(self.FilePath, new_size)

        return [scm.ResultCode.OK], [f"File shrunk to size '{new_size}'"]

Here, we are returning a (ska_control_model.ResultCode, message) pair as this is recommended by the SKA CS Guidelines, however, we are using exceptions to report total failures. This is because clients already have to deal with exceptions thrown by Tango when e.g. there are network issues connecting to the device, it is often easier for them to handle an additional exception rather than having to check error codes as well. The ska_control_model.ResultCode should only be used by a command if it needs to communicate partial success. See Reporting failure of long-running commands for a longer discussion on reporting failures.

We are also providing an is_Shrink_allowed() method to disallow calling the command if our device is disabled. This is important to avoid modifying the file if our device has been put in AdminMode.OFFLINE.

Warning

It might be tempting to disallow this command while the healthState is not OK to avoid having to deal with the current size not being available. In this exact situation this might be reasonable because we know for certain it isn’t going to work, however, it is a bad idea for a Tango device to disobey a client based on the healthState. Similarly, we do not reject the command when we are in DevState.FAULT.

In general, your SKA Tango device should make its best effort to execute the commands it has been instructed to execute and to report back any problems it finds. The client already knows about the problem and might be trying to diagnose issues by invoking commands. We should not get in their way. Note, however, that the command still fails if we cannot determine the size of the file because it would be “dangerous” to shrink the file if we do not know that it is really a shrink operation.

This argument actually suggests that we should not use the Metadata object to determine the original size, but instead call os.stat() directly in the command. However, we chose the current implementation to highlight the stored=True option for the Signal data descriptor.

Implementing a long running command

When an asynchronous long running command is invoked on a SKA Tango device, it is sometimes necessary to cancel the command because, for example, the situation has changed so the command is no longer required or because the command was invoked in error. The SKA long running command mechanism provides an Abort() command to support command cancellation and this is implemented as a “co-operative abort”, in that each command is responsible for checking if it has been aborted while it is executed.

In order to support being aborted, our Grow() command will grow the file in chunks and check if the command has been aborted between writing each chunk. As the logic for doing this transfer is a little complicated, it would be a good idea to test this independently, let’s start by writing a chunked_transfer() function that supports being aborted:

Step 8c: Chunked transfer

def chunked_transfer(
    destination: typing.BinaryIO,
    source: typing.BinaryIO,
    chunk_size: int,
    total_bytes: int,
    *,
    progress_callback: stb.type_hints.ProgressCallbackType,
    task_abort_event: threading.Event,
) -> None:
    progress_callback(progress=0)
    written = 0
    last_progress = 0

    def transfer_chunk(size: int) -> None:
        nonlocal written, last_progress

        bytes = source.read(size)
        if len(bytes) != size:
            raise RuntimeError(
                f"Not enough data. Found {written + len(bytes)} bytes, "
                f"expected at least {total_bytes} bytes."
            )

        destination.write(bytes)
        destination.flush()
        os.fsync(destination.fileno())

        written += len(bytes)
        progress = int(100 * (written / total_bytes))
        if progress > last_progress:
            progress_callback(progress=progress)
            last_progress = progress

    n_complete_chunks = total_bytes // chunk_size
    remainder = total_bytes - n_complete_chunks * chunk_size

    for i in range(n_complete_chunks):
        transfer_chunk(chunk_size)

        if task_abort_event.is_set():
            raise stb.executor.TaskAborted()

    if remainder > 0:
        transfer_chunk(remainder)

    assert written == total_bytes

Here, we also accept a progress_callback which we call with a completion percentage as we transfer each chunk. As we will see later, this callback is provided by ska-tango-base’s long running command API and it is generally a good idea to provide this progress information if possible. If we notice that the task_abort_event has been set, we raise a TaskAborted exception to acknowledge that we have been aborted. This exception will be used by the long running command API later.

Now we are ready to add our long running command. In order to support long running commands, our device needs to inherit from LRCMixin in addition to the BaseInterface class. We then decorate our command function with long_running_command() as opposed to command(). This command method needs to return a task function that will then be scheduled on a background thread to perform the long running command. This task function can be built with the task() decorator. Let’s have a go:

Step 8d: Grow

import ska_tango_base.long_running_commands as stb_lrc


class FileStats(stb_lrc.LRCMixin, stb.base.BaseInterface):
    """Tango device that exposes file statistics as attributes."""

    def is_Grow_allowed(
        self,
        request_type: stb_lrc.LRCReqType = stb_lrc.LRCReqType.ENQUEUE_REQ,
    ) -> bool:
        """Return true if the Grow command is allowed."""
        if request_type == stb_lrc.LRCReqType.EXECUTE_REQ:
            return typing.cast(bool, self.get_state() != tango.DevState.DISABLE)

        return True

    Grow_SCHEMA: dict[str, stb.type_hints.JSONData] = {
        "$schema": "https://json-schema.org/draft/2020-12/schema",
        "$id": "artefact.skao.int/mylrc.schema.json",
        "title": "Grow schema",
        "description": "Validates the keyword arguments for the Grow command",
        "type": "object",
        "properties": {
            "new_size": {"type": "number"},
            "chunk_size": {"type": "number"},
            "source": {"type": "string"},
        },
    }

    @stb_lrc.long_running_command
    @stb.validators.validate_json_args
    def Grow(
        self, new_size: int, chunk_size: int, source: str
    ) -> stb.type_hints.TaskFunctionType:
        """
        Grow the file.

        The file will be grown in chunks with bytes from a given source.

        The source must be able to provide all of the required bytes.  If not
        enough bytes are available then the operation will fail and the file
        will be reset to its original size.

        This is a long running command.

        :param new_size: New size to grow to
        :param chunk_size: Size of the chunks
        :param source: path to source of bytes
        :return: ResultCode, message
        """

        @stb.executor.TaskExecutor.task
        def task(
            progress_callback: stb.type_hints.ProgressCallbackType,
            task_abort_event: threading.Event,
        ) -> tuple[scm.ResultCode, str]:
            try:
                initial_size = self.metadata.get_size_or_raise()
            except Exception as ex:
                return (scm.ResultCode.FAILED, f"{ex}")

            if new_size < initial_size:
                return (
                    scm.ResultCode.FAILED,
                    f"New size '{new_size}' is less than current size '{initial_size}'"
                    "Cannot grow file.",
                )

            to_transfer = new_size - initial_size
            source_file = open(source, "br")
            dest_file = open(self.FilePath, "ba")

            try:
                chunked_transfer(
                    dest_file,
                    source_file,
                    chunk_size,
                    to_transfer,
                    progress_callback=progress_callback,
                    task_abort_event=task_abort_event,
                )
            except Exception as ex:
                source_file.close()
                dest_file.close()
                os.truncate(self.FilePath, initial_size)
                if isinstance(ex, stb.executor.TaskAborted):
                    self.logger.error(
                        f"Grow aborted. Shrinking back to initial size '{initial_size}'"
                    )
                    raise
                else:
                    self.logger.exception(
                        f"Grow failed. Shrinking back to initial size '{initial_size}'"
                    )
                    return scm.ResultCode.FAILED, f"Chunked transfer failed: {ex}"

            return scm.ResultCode.OK, f"File size increased to {new_size}"

        return task

Here, we are using the ska_tango_base.validators.validate_json_args() to support multiple arguments for the command. The resulting Tango command will have a single string as an argument which is expected to hold a JSON string. This JSON string will be validated against the Grow_SCHEMA provided before the task is submitted to be executed. The client will receive an exception if the string is not valid JSON or does not match the schema.

Notice as well, that we are not raising exceptions from our task (except TaskAborted), but instead we now return a ResultCode. This is because there is no mechanism to transfer an exception to an LRC client once the task has been scheduled and so instead long running commands use error codes. Again, see Reporting failure of long-running commands for a more detailed discussion of this.

Let’s restart the file_stats_device.py script and try out our new commands. We can use the invoke_lrc() function to call our long running command and get notified of its progress:

In [1]: dp = Device("tango://localhost:12345/tut/fs/1#dbase=no")

In [2]: import json

In [3]: import ska_tango_base.long_running_commands as stb_lrc

In [4]: def make_callback(name):
   ...:     def callback(**kwargs):
   ...:         print(f"{name}: {kwargs}")
   ...:     return callback
   ...:

In [5]: dp.Shrink(0)
API_CommandNotAllowed: Command Shrink not allowed when the device is in DISABLE state
(For more detailed information type: tango_error)

In [6]: subs = stb_lrc.invoke_lrc(make_callback("Grow"), dp, "Grow", command_args=(json.dumps({"new_size": 4096, "chunk_size": 512, "source": "/dev/urandom"}),))

Grow: {'status': <TaskStatus.STAGING: 0>}
Grow: {'status': <TaskStatus.QUEUED: 1>}
Grow: {'status': <TaskStatus.REJECTED: 6>, 'result': [6, 'Command is not allowed']}
In [7]: dp.adminMode = "ONLINE"

In [8]: dp.Shrink(0)
Out[8]: [array([0], dtype=int32), ["File shrunk to size '0'"]]

In [9]: dp.size
Out[9]: 0

In [10]: subs = stb_lrc.invoke_lrc(make_callback("Grow"), dp, "Grow", command_args=(json.dumps({"new_size": 4096, "chunk_size": 512, "source": "/dev/urandom"}),))

Grow: {'status': <TaskStatus.STAGING: 0>}
Grow: {'status': <TaskStatus.QUEUED: 1>}
Grow: {'status': <TaskStatus.IN_PROGRESS: 2>}
Grow: {'progress': 0}
Grow: {'progress': 12}
Grow: {'progress': 25}
Grow: {'progress': 37}
Grow: {'progress': 50}
Grow: {'progress': 62}
Grow: {'progress': 75}
Grow: {'progress': 87}
Grow: {'progress': 100}
Grow: {'status': <TaskStatus.COMPLETED: 5>, 'result': [0, 'File size increased to 4096']}
In [11]: dp.size
Out[11]: 4096

Let’s also check that we can Abort() the long running command:

In [12]: subs = stb_lrc.invoke_lrc(make_callback("Grow"), dp, "Grow", command_args=(json.dumps({"new_size": 30 * 1024 * 1024, "chunk_size": 512, "source": "/dev/urandom"}),))

Grow: {'status': <TaskStatus.STAGING: 0>}
Grow: {'status': <TaskStatus.QUEUED: 1>}
Grow: {'status': <TaskStatus.IN_PROGRESS: 2>}
Grow: {'progress': 0}
Grow: {'progress': 1}
Grow: {'progress': 2}
Grow: {'progress': 3}
Grow: {'progress': 4}
Grow: {'progress': 5}
Grow: {'progress': 6}
Grow: {'progress': 7}
Grow: {'progress': 8}
Grow: {'progress': 9}
Grow: {'progress': 10}
Grow: {'progress': 11}
Grow: {'progress': 12}
Grow: {'progress': 13}
Grow: {'progress': 14}
Grow: {'progress': 15}
Grow: {'progress': 16}
Grow: {'progress': 17}
In [13]: subs2 = stb_lrc.invoke_lrc(make_callback("Abort"), dp, "Abort")

Abort: {'status': <TaskStatus.STAGING: 0>}
Abort: {'status': <TaskStatus.IN_PROGRESS: 2>}
Grow: {'status': <TaskStatus.ABORTED: 3>, 'result': [7, 'Task aborted']}
Abort: {'status': <TaskStatus.COMPLETED: 5>, 'result': [0, 'Abort completed OK']}

Great! This all seems to be working.

Conclusion

With this you now have built an SKA Tango device that can monitor and control a file on the file system. Along the way we have learnt:

• How to manage the various BaseInterface attributes

• How to setup monitoring of a component and push data to the Tango device using the signal bus

• How to create commands, be they long running or fast

Further reading

• Tango Best Practices: For some general advice about writing Tango devices

• How to use optional attributes: For if you need to support some of the optional attributes defined by the SKA Control Model

• Components and component managers: For if you are working with a device that uses a component manager

• How to invoke a long running command: For when you need to call some other Tango device’s long running command