hapPyTango

A lightweight, asyncio-based Tango client for Python 3, built on fnorb (a pure-Python CORBA ORB). No PyTango, no C++ Tango library, no compilation.

This is a modern Python 3 reimplementation of the original gevent-based hapPyTango: gevent greenlets become asyncio coroutines, and events use Tango's ZMQ event system instead of the old (dropped) CORBA notifd events.

What it is

A device's Tango commands and attributes become awaitable members on a Device proxy:

dev = await hapPyTango.connect("sys/tg_test/1")
await dev.DevString("hello")   # call a command
await dev.double_scalar()      # read an attribute
await dev.double_scalar(3.14)  # write an attribute

All I/O runs on the host application's asyncio event loop through fnorb.aio, so communication with Tango devices is non-blocking and cooperative. It is not a complete Tango client like PyTango - it is small (the client is a few hundred lines) but functional as-is.

hapPyTango started as a fun week-end project: an alternative Tango client for Python that would be lightweight, easy to install (especially on Windows - no compilation, no C++ library needed), easy to use, Pythonic, and able to take advantage of cooperative multi-tasking. The original relied on gevent greenlets; this version uses asyncio coroutines, so any host application running an asyncio event loop benefits from asynchronous communication with Tango devices and can weave complex device sequences into its own concurrency.

Install

Requires Python >= 3.11, fnorb, and pyzmq (for events).

Using pixi (recommended - also provides real Tango servers to test against):

pixi install
pixi run python -c "import hapPyTango"

Or with pip (fnorb must be importable - install it first from its repo):

pip install git+https://github.com/mguijarr/fnorb
pip install -e .

Usage

Connecting

connect() returns a ready-to-use Device. A device is named domain/family/member; prefix it with host:port/ to point at a specific Tango host, and append #dbase=no to talk to a device that runs without the Tango database.

import asyncio
import hapPyTango

async def main():
    # Any of these forms works:
    #   "sys/tg_test/1"                              (via $TANGO_HOST + database)
    #   "host:port/sys/tg_test/1"                    (explicit host, database)
    #   "host:port/sys/tg_test/1#dbase=no"           (no database)
    dev = await hapPyTango.connect("sys/tg_test/1")
    print("connected to", dev.name)
    await dev.close()

asyncio.run(main())

Commands and attributes

After connect(), every command and attribute the device exposes is available as an awaitable member. Calling with no argument reads, calling with one argument writes:

# Commands are awaitable methods; the argument (if any) is the command input:
print(await dev.DevString("hello"))
print(await dev.State(), await dev.Status())

# Attributes: no argument reads, one argument writes:
print(await dev.double_scalar())      # read
await dev.double_scalar(3.14)         # write
print(await dev.double_spectrum())    # a list, for SPECTRUM attributes

# Read from the device's polling cache instead of the hardware:
print(await dev.long_scalar(from_cache=True))

There are explicit forms too, handy when the member name is dynamic:

await dev.command_inout("DevString", "hello")
await dev.read_attribute("double_scalar")
await dev.write_attribute("double_scalar", 3.14)

When calling a command or reading/writing an attribute, several keyword arguments are accepted:

• timeout - timeout in seconds (default: 3).
• from_cache - return the value from the device's polling cache instead of the hardware (default: False).

Reads take two more:

• return_value - return the attribute value, or the complete AttributeValue object (default: True).
• return_set_value - also return the set value, as a (read_value, set_value) pair (default: False).

Error handling

Tango DevFailed / MultiDevFailed errors are translated into TangoError:

try:
    await dev.throw_exception(...)
except hapPyTango.TangoError as exc:
    print("device refused:", exc)

Events

Subscribe to Tango ZMQ events. The callback may be a plain or a coroutine function; keep the returned subscription and unsubscribe() to stop it.

def on_change(event):
    if event.errors:
        print("event error:", event.errors[0].desc)
    else:
        print(event.event_name, "->", event.value)

sub = await dev.subscribe_event("double_scalar", on_change, event_type="change")
...
await sub.unsubscribe()

event_type is one of "change", "periodic", "archive", "user_event". The callback receives an EventData with .value, .quality, .event_name and .errors. A device only emits events for an attribute once polling / event criteria are configured for it - normally in the Tango database, or at runtime with the admin device's AddObjPolling command (see examples/experiment.py).

Runnable examples

The examples/ directory has complete scripts that run against a live TangoTest device:

• run_tangotest.sh - start a no-database TangoTest server on port 10000.
• demo.py - commands, attribute read/write and a short event subscription.
• experiment.py - the above plus enabling server-side polling and watching periodic events.

./examples/run_tangotest.sh &                       # terminal 1
pixi run python examples/experiment.py              # terminal 2

Interactive use

Because everything is a coroutine, hapPyTango is pleasant to drive from an asyncio REPL. ptpython supports top-level await, so you can talk to a device interactively - and since commands and attributes are real members, tab-completion and dir(dev) show you what the device offers:

$ ptpython
>>> import hapPyTango
>>> dev = await hapPyTango.connect("localhost:10000/sys/tg_test/1#dbase=no")
>>> dir(dev)          # commands, attributes and methods are all members
['DevBoolean', 'DevDouble', 'DevLong', 'DevString', 'DevVoid', 'Init',
 'State', 'Status', 'SwitchStates', ...,
 'ampli', 'boolean_scalar', 'close', 'command_inout', 'connect',
 'double_image', 'double_scalar', 'double_spectrum', 'enum_scalar',
 'long_scalar', 'name', 'ping', 'read_attribute', 'short_scalar',
 'string_scalar', 'subscribe_event', 'throw_exception', 'wave',
 'write_attribute', ...]
>>> await dev.State()
RUNNING
>>> await dev.Status()
'The device is in RUNNING state.'
>>> await dev.DevDouble(3.14)      # a command echoes its argument
3.14
>>> await dev.DevString("hello")
'hello'
>>> await dev.double_scalar()      # TangoTest simulates the read value
2.0183516759952433

How it works

The interesting parts are:

Module	Responsibility
client.py	The Device proxy, the command/attribute wrappers, value conversion, and connection bootstrap.
events.py	ZMQ event subscription and CDR decoding of event messages.
Tango_aio/	The Tango CORBA stubs, generated from the vendored tango.idl (see below). Do not edit by hand — regenerate instead.

Regenerating the CORBA stubs

src/hapPyTango/Tango_aio/ is generated from tango.idl, a vendored copy of the official Tango V11 interface definition kept at the repository root. It is compiled with fnorb's IDL compiler:

$ python ../fnorb/tools/compile_idl.py tango.idl <output_dir>

Connection bootstrap

connect() builds a Device and calls Device.connect(), which:

1. Parses the name (_parse_name) into a Tango host, a domain/family/member name, and a no-database flag.
2. Resolves the object reference. With a database, _resolve_via_db calls DbImportDevice on the database object and gets back a stringified IOR plus the device's IDL revision. Without a database (#dbase=no), the reference is built by hand from a host:port/key corbaloc (_object_from_corbaloc), since fnorb.aio only stringifies real IORs; the revision is then probed with CORBA _is_a (_detect_impl).
3. Narrows the reference to the right typed stub - Device, Device_2, …, Device_7 (_narrow / _device_class). The revision decides which CORBA methods exist: e.g. command_inout vs command_inout_2, read_attributes vs read_attributes_3. Spectrum/image attributes need revision >= 3.
4. Introspects (_introspect): command_list_query and get_attribute_config populate a _members dict of _Command / _Attribute wrappers. State and Status are added as special no-argument commands if the device doesn't expose them as commands.
5. Starts a background keepalive task that pings the device every few seconds so the connection stays warm.

Device.__getattr__ looks names up in _members, which is what turns dev.double_scalar into the wrapper object you await.

Calling commands and attributes

_Command.__call__ wraps the Python argument in a CORBA Any of the command's input type and invokes the right command_inout* method under an asyncio.timeout. _Attribute.read / .write do the same for read_attributes* / write_attributes*, wrapping the whole call so that a Tango DevFailed becomes a TangoError.

The Any marshalling lives in two helpers:

• python2any(value, data_type, ...) wraps a Python value in a CORBA Any of the correct Tango IDL type. Attribute writes always use the array type (a scalar is wrapped as a one-element array), matching how Tango carries values.
• tango2python(any_value, scalar=...) unwraps an Any back to a plain Python value, unwrapping the one-element array Tango uses for scalars and splitting the mixed DevVarLongStringArray / DevVarDoubleStringArray types into (numbers, strings) tuples.

The _CMD_TYPE / _ATTR_TYPE tables map Tango's CmdArgType / AttributeDataType enum values to the scalar and array IDL type names used for this wrapping.

Events

Modern Tango (>= 8) delivers events over ZMQ, not CORBA. subscribe_event (in events.py):

1. Connects a proxy to the device's admin (dserver) device and calls ZmqEventSubscriptionChange, which returns the ZMQ endpoint and the topic string to listen on.
2. Opens a zmq.asyncio SUB socket, connects it to the endpoint, and subscribes to the exact topic (so ZMQ filters out other events on the same channel).
3. Runs a receive loop that decodes each event. Every event is a 4-part ZMQ message [topic, endianness, ZmqCallInfo, data]; the data frame is a 0xc0dec0de marker followed by an AttributeValue_4/5 marshalled as its own CDR stream. fnorb's typecodes unmarshal both, honouring the per-message byte order. The decoded EventData is handed to your callback (awaited if it is a coroutine).

unsubscribe() cancels the loop, closes the socket, and tells the admin device to stop sending.

Tests

pixi run python -m unittest discover -s tests

Unit tests cover value conversion, command/attribute dispatch, and event decoding without a live device. The client is additionally verified end-to-end against a real TangoTest device.

Manual end-to-end check against a real device

# no-database TangoTest on port 10000
pixi run TangoTest test -nodb -dlist sys/tg_test/1 -ORBendPoint giop:tcp::10000 &
pixi run python -c "
import asyncio, hapPyTango
async def main():
    dev = await hapPyTango.connect('localhost:10000/sys/tg_test/1#dbase=no')
    print(await dev.State(), await dev.double_scalar())
    await dev.close()
asyncio.run(main())
"

License

MIT - see LICENSE.