This is the multi-page printable view of this section. Click here to print.

Return to the regular view of this page.

Communication layer

ZeroMQ Hubs, clients, properties and configuration.

Quatm’s services communicate over ZeroMQ. The design keeps the timing path independent of rendering, and makes local and remote clients interchangeable.

Hubs and streams

Each data type has a Hub that collects and redistributes messages:

scalar data streams;
image streams;
command channels;
property updates;
messages and logs.

Clients connect to the Hubs they need. A client process does not need to know which other clients exist, and a Hub failure is visible and restartable.

Client types

Client	Purpose
`DataClient`	subscribe to scalar data series
`ImageClient`	subscribe to camera images
`CommandClient`	send commands to a controller
`MessageClient`	publish and receive operational messages
`Properties`	synchronise the distributed property tree

Properties

The property tree is the shared configuration and state model. Properties are read and written by name through a filesystem-like path; the Property Logger persists the tree and applies the persistence policy. GUI panels, web dashboards and experiment code all read the same values.

Remote property writes use a persistent publisher and explicit acknowledgement, so a remote action is not silently lost when a connection is slow.

Transport choices

IPC for processes on the same machine (lowest latency).
TCP for split control/analysis machines and remote clients.

Curve encryption is available for local and network channels. The transport layer also handles multipart framing, malformed-frame recovery and request timeouts.

Configuration

Endpoints, project root and service selection are read from the project’s configuration/ directory. A service can be launched by the Operations Console, by a project launcher, or directly as a Python module for debugging.

1 - Clients

Data/Image/Command clients: publish/subscribe communication (TCP/IPC), with connection targets set by the central configuration.

Overview

quatm/servers/clients.py provides the communication clients: the connection target is determined by the central configuration (Data / Image / Command Hub). GenericClient offers the standard send/receive interface and is suited to small data (<10k); use ImageClient for large payloads such as images.

Client classes

Class	Connection	Description
`GenericClient`	Determined by configuration	General send/receive (including numpy arrays), subscribe/unsubscribe
`DataClient`	Data Hub	Publish and subscribe ordinary data
`ImageClient`	Image Hub	Sending large image-scale data (send takes a numpy array directly)
`CommandClient`	Command Hub	Sends control commands to drivers (the command name is folded into the topic)
`NpEncoder`	—	JSON serialization of numpy types

Key methods (GenericClient)

Method	Description
`send(datadict, arr=None)`	Publish data (optionally with one numpy array)
`recv()`	Receive one frame: (topic, dict[, array]); returns None on timeout
`subscribe` / `unsubscribe`	Subscribe / unsubscribe a channel
`has_new_data`	Non-blocking check for new data
`close`	Close the PUB/SUB socket (idempotent)

Usage

from quatm.servers.clients import DataClient

dc = DataClient(name="pmt")
dc.send({"pmt": 1234}, arr=numpy_array)
topic, data = dc.recv()

2 - Property tree

Distributed hierarchical property tree: the local copy is synchronized bidirectionally through the Property Hub and Propertylogger.

Overview

quatm/servers/properties.py provides the property tree (Properties): it manages the configuration and properties of the control centre and maintains a synchronized copy of the property dictionary. A local set / get publishes to the Property Hub, and central changes are pushed back to the local copy. Properties are addressed like a file system (for example Drivers/xxx, or an absolute path starting with /).

Key methods

Method	Description
`set(key, value)`	Set a property and publish the change to the Property Hub
`get(key, default=...)`	Read a value (always returns a deep copy; creates the default if missing)
`delete(key)`	Delete an entry (including all its children) and broadcast the change
`changes()`	Return the keys changed since the previous call
`close` / `close_all`	Close sockets and background threads

PropertyAttribute exposes a single property of the property tree as an ordinary Python attribute (the host must have _props).

Usage

from quatm.servers.properties import Properties

props = Properties()
props.set("laser/power", 50e-3)
power = props.get("laser/power")

3 - Data management

Multi-stream data aggregation: aligned to each experiment run, missing streams zero-filled on timeout, then written to HDF5.

Overview

quatm/servers/datamgr.py provides data management (DataSummary): it aggregates the data streams of several input channels and assembles them into a complete dictionary for each run. Incoming data is first accumulated in incompleteData, and once the data of all subscribed channels has arrived the whole record is moved to completeData; if some data is still missing when the run ends and times out, the missing parts are filled with zeros.

Key methods

Method	Description
`initDataq`	(Re)subscribe to the channels configured in the properties and to the experiment start/end events
`recvData` / `processIncomingData`	Receive data and merge it into the most recent run record
`savetoFile`	Write the collected run dictionary to a structured HDF5 dataset
`loadFromFile`	Read a set of stored run dictionaries back from HDF5
`clear`	Clear all data collected so far
`run_forever`	Blocking loop that keeps receiving data

Usage

from quatm.servers.datamgr import DataSummary

summary = DataSummary(name="sum")
summary.initDataq()     # Subscribe to channels configured in properties
summary.run_forever()   # Keep receiving; savetoFile writes the run dictionary to .h5

4 - Messaging

Standardized message stream client: sends timestamped, source-tagged messages to the Message Hub.

Overview

quatm/servers/messageclient.py provides the messaging system: MessageClient is the standardized message stream client, sending messages with a timestamp and originating file to the Message Hub (PUB). Everyday code should generally not construct it directly but use the module-level send functions instead.

Send functions

Function	Description
`send_info(msg)`	Info level: enters the message stream as an ordinary notification
`send_warning(msg)`	Warning level: enters the message stream and raises a warning notification
`send_error(msg)`	Error level: enters the message stream and raises an error notification
`send_debug(msg)`	Debug level: shown only on the debug channel

Usage

from quatm.servers.messageclient import send_info, send_error

send_info("Experiment started")
send_error("Camera connection timed out")

5 - Configuration and property persistence

configreader + propertylogger: reading JSON configuration, persisting the property tree and monitoring active streams.

Modules

Module	Function
`configreader.ConfigReader`	Read-only wrapper: reads `configfile.json` (`getConfiguration()`)
`configreader.Config()`	Reads configfile.json and returns the configuration dictionary
`configreader.Properties()`	Reads the properties file `properties.json` and returns a dictionary
`propertylogger.run_propertylogger()`	Property tree persistence + INIT acknowledgement (Property Logger service)
`datalogger.DataStreamLogger`	Monitors active data streams and writes them into the property tree
`imagelogger.ImageStreamLogger`	Monitors active image streams and writes them into the property tree

Usage

from quatm.servers.configreader import ConfigReader

cfg = ConfigReader()
config = cfg.getConfiguration()   # Full configuration dictionary

6 - Peripheral service modules

Peripheral services such as InfluxDB, notification push, the image network server, the Web Hub and the XSUB/XPUB proxy.

Service list

Module	Function
`influxdb`	InfluxDB client wrapper: hardware status writes (`data2db`) and historical queries (`query_data`); can run as an RPC service
`notifications`	Notification push: configuration service, `push_message` push, `post_to_mattermost` compatibility entry point
`image_network_server`	Image network service: flushes the latest frame of each image stream to a Web figure
`webhub`	Server entry point of the Web remote control platform (`quatm.web`; takes host/port from the Servers section of configfile)
`xsub_xpub`	XSUB/XPUB forwarding server (`run_server`)
`experiment_status`	Experiment status monitoring / early-warning notifications
`propertylogger` / `datalogger` / `imagelogger`	Property tree persistence and monitoring of active data/image streams (see Configuration system)

Usage

from quatm.servers.notifications import push_message

push_message("Experiment finished")   # Sent through enabled push services (never raises)