babbelfish

A lightweight async service framework used by every sensor service in this workspace. Provides two base classes plus convention-driven YAML loading on top of heisskleber transports.

The goal is to let a service implementation be a single file: one @dataclass for configuration, one subclass of Service with an async runner(), and nothing else. Everything else — signal handling, sender wiring, config-file parsing — is handled here.

Quick tour

from dataclasses import dataclass

from babbelfish import Service, ServiceConf


@dataclass
class MyServiceConf(ServiceConf):
    topic: str = "my-topic"
    poll_interval: float = 1.0


class MyService(Service):
    def __init__(self, config: MyServiceConf) -> None:
        self.config = config
        self.senders = list(config.senders.values())
        super().__init__(config)

    async def runner(self) -> None:
        while True:
            payload = {"epoch": ..., "value": 42}
            await asyncio.gather(
                *[s.send(payload, topic=self.config.topic) for s in self.senders]
            )
            await asyncio.sleep(self.config.poll_interval)


if __name__ == "__main__":
    config = MyServiceConf.from_file("config.yaml")
    service = MyService(config)
    service.start()
    asyncio.run(service.task)

Example config.yaml:

name: "my-service"
topic: "/my-topic"
poll_interval: 0.5

output:
  mqtt:
    host: localhost
    port: 1883
  zmq:
    host: 127.0.0.1
    port: 5555
  file:
    rollover: 3600
    name_fmt: "mine_%Y%m%dT%H%M%S.csv"
    directory: "/tmp"
    format: "csv"

ServiceConf

Dataclass base for service configuration. Subclass it, add your own fields, and call from_file(path) to load a YAML config.

Built-in fields

Class methods

• from_file(path) — load a YAML file and delegate to from_dict.
• from_dict(data) — pop output: / input: sections, instantiate the matching heisskleber transports via their registry, drop unknown top-level keys, and return an instance of the subclass.

output: / input: sections

The first word of each key (before the first _) selects which heisskleber transport is used. Multiple instances of the same protocol are supported by suffixing:

output:
  mqtt:          # first MQTT sender
    host: primary
  mqtt_backup:   # second MQTT sender
    host: secondary
  zmq:
    host: 127.0.0.1
    port: 5555

The sub-dict under each key is passed to the corresponding heisskleber config class (MqttConf, ZmqConf, FileConf, …). Unknown protocols raise ValueError at load time.

After loading, config.senders is a dict[str, Sender] keyed by the original label ("mqtt", "mqtt_backup", "zmq", …). Services typically access them as a plain list: list(config.senders.values()).

Subclassing for nested structures

ServiceConf.from_dict silently drops unknown keys. If your config has nested structure (lists of dicts, typed sub-objects, …), override from_dict in your subclass — see msb_can.service.CanConf for an example that parses a channels: list into typed ChannelConf instances.

Service

Abstract base class for async services. Handles signal registration, logger setup, and task lifecycle.

Required override

@abstractmethod
async def runner(self) -> None:
    ...

runner is expected to be an infinite loop that produces and publishes data. It runs as an asyncio.Task created by start().

Override hooks (all optional)

stop_hook() is where hardware teardown belongs — close serial ports, disconnect I²C, stop driver threads, etc.

Signal handling

Service.__init__ installs handlers for SIGINT and SIGTERM that:

1. Cancel every other task on the event loop.
2. Gather them to surface any exceptions.
3. Log Program crashed successfully. and stop the loop.

Derived services must not install their own signal handlers — the base class owns the loop-level handlers.

Logger

self.logger is logging.getLogger(config.name). Pair this with a logging.yaml that configures a logger whose key matches config.name:

loggers:
  my-service:          # matches config.name
    level: INFO
    handlers: [console]
    propagate: no

Installation

Within the workspace:

[tool.uv.sources]
babbelfish = { workspace = true }

[project]
dependencies = [
    "babbelfish>=0.1.3",
    # …
]

Outside the workspace, install from PyPI:

uv pip install babbelfish

Or pin to a specific version:

uv pip install "babbelfish>=0.1.4"

Dependencies

• heisskleber — provides the Sender / Receiver transport abstractions and their registries. All services in this workspace publish via heisskleber only; paho-mqtt / pyzmq are never imported directly from a service.

Running the tests

uv run --no-sync pytest packages/babbelfish/tests/