Metadata-Version: 2.4
Name: python_zmq_framework
Version: 0.2.0
Summary: Python node library for a flow-based, language-agnostic node runtime over ZeroMQ (Node-RED without the UI) — the Python counterpart to ruby_zmq_framework.
Author-email: Paul Daniel <paulgdan@gmail.com>
License: MIT License
        
        Copyright (c) 2026 Paul Daniel
        
        Permission is hereby granted, free of charge, to any person obtaining a copy
        of this software and associated documentation files (the "Software"), to deal
        in the Software without restriction, including without limitation the rights
        to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
        copies of the Software, and to permit persons to whom the Software is
        furnished to do so, subject to the following conditions:
        
        The above copyright notice and this permission notice shall be included in all
        copies or substantial portions of the Software.
        
        THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
        IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
        FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
        AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
        LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
        OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
        SOFTWARE.
        
Project-URL: Homepage, https://github.com/pgdaniel/python_zmq_framework
Project-URL: Flow runtime (Ruby), https://github.com/pgdaniel/ruby_zmq_framework
Project-URL: Changelog, https://github.com/pgdaniel/python_zmq_framework/blob/master/CHANGELOG.md
Keywords: zeromq,zmq,pubsub,flow-based-programming,message-bus,node-red
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.8
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: System :: Distributed Computing
Classifier: Topic :: System :: Networking
Requires-Python: >=3.8
Description-Content-Type: text/markdown
License-File: LICENSE.txt
Requires-Dist: pyzmq>=25.0
Dynamic: license-file

# python_zmq_framework

**Python node library for a flow-based, language-agnostic runtime — Node-RED
without the UI.** The reference runtime lives in
[`ruby_zmq_framework`](https://github.com/pgdaniel/ruby_zmq_framework); this
library lets a Python process join that flow as a full citizen, or run
entirely standalone with no Ruby involved.

- **Nodes** are independent OS processes. Each one does one job, lives in one
  file, and knows nothing about any other node — not their ports, not their
  names, not their language.
- **Wires** are pub/sub topics carrying JSON, over ZeroMQ.
- **The graph is data**: `flow.yml` (in the Ruby repo) is the only artifact
  that knows the topology. `bin/flowctl` reads it, computes the wiring, and
  runs every node — Python or Ruby — with that wiring in its environment.
- **The contract is one page**: [`PROTOCOL.md`](PROTOCOL.md) is everything a
  node in any language needs to join.

Only dependency is `pyzmq`. Nothing here requires Ruby to install or run —
`boot()` also works standalone with no environment set, for local iteration
or running a single node on its own.

## Install

```bash
pip install -e .
# or, without installing the package:
pip install -r requirements.txt
```

## Writing a node

```python
from python_zmq_framework import FrameworkNode, boot

class RpmSmoother(FrameworkNode):
    def __init__(self, bus):
        super().__init__(bus)
        self.window = []

    def handle_message(self, topic, payload):
        self.window = (self.window + [payload["rpm"]])[-5:]
        self.broadcast("engine_data_smooth", {"rpm": sum(self.window) / len(self.window)})

RpmSmoother_instance = boot(RpmSmoother)
```

Note what's absent: no ports, no peers, no `.subscribe()` calls. Wiring
comes entirely from environment variables — `BUS_PORT`, `BUS_PEERS`,
`BUS_SUBSCRIBES`, `NODE_NAME` — set by whatever launches the process (see
[`PROTOCOL.md`](PROTOCOL.md)). A flow manifest entry in the Ruby repo's
`flow.yml` computes and injects them:

```yaml
  rpm_smoother:
    cmd: python3 ../python_zmq_framework/nodes/rpm_smoother.py
    subscribes: [engine_data]
    publishes: [engine_data_smooth]
```

Run standalone (no environment needed — it binds an ephemeral port) to poke
at a node in isolation: `python3 nodes/dbc_decoder.py`.

Every node automatically heartbeats every 5 seconds.
`FrameworkNode.handle_message` is `@abstractmethod`, so Python's `ABC`
machinery raises `TypeError` the moment you try to instantiate a subclass
that forgot it — before `__init__` runs at all, so no heartbeat thread can
leak from a failed construction. That's exactly the fast, specific feedback
an iterating LLM agent needs.

## How It Works

1. **`ZeroMQBus`** binds a PUB socket (pass `0` for an OS-assigned
   ephemeral port, readable back via `.port`) and connects a SUB socket to
   every peer. Peers may be ints (loopback ports), `"host:port"` strings,
   or full ZeroMQ endpoints; pass `bind_host="0.0.0.0"` to accept peers
   from other machines. There is no dynamic discovery — every node's
   address has to be listed in every peer's `peer_ports` up front (or
   computed for you by `bin/flowctl`). Messages published on a bus are
   also delivered synchronously to subscribers on that same bus.
2. **`FrameworkNode`** is the base class for a node on the bus. Subclass
   it and implement `handle_message(topic, payload)`. You get an
   automatic `heartbeat` broadcast (`node_name`, `status`, `timestamp`)
   every 5 seconds for free. `node_name` defaults to the class name; set
   `self.node_name = "..."` before calling `super().__init__()` to give
   distinct instances distinct identities.
3. **`boot(NodeClass)`** builds a node entirely from the environment
   contract in `PROTOCOL.md` — the same helper the Ruby gem exposes as
   `RubyZmqFramework.boot`.
4. **Resilience:** the listener survives anything the network throws at
   it — non-framework frame layouts and malformed JSON are dropped with a
   warning, and each subscriber's `handle_message` is caught individually
   so one raising handler can't starve the others or kill the listener.
   Handlers on one bus never run concurrently.
5. **Clean shutdown:** `bus.close()` stops the listener and releases the
   sockets; `node.stop_heartbeat()` ends the heartbeat thread. Stop your
   node first, then close the bus — publishing on a closed bus raises
   `ZeroMQBusError`.
6. **Thread safety:** a ZeroMQ PUB socket must not be written to
   concurrently from multiple threads. The heartbeat thread and whatever
   thread calls `broadcast()`/`publish()` both use the same socket, so
   `ZeroMQBus.publish` takes a lock around every send.

## Demo nodes

```bash
pip install -r requirements.txt
python3 nodes/dbc_decoder.py        # broadcasts engine_data
python3 nodes/telemetry_logger.py   # logs whatever BUS_SUBSCRIBES names
```

`nodes/dbc_decoder.py` is a stand-in for a real `cantools`-based DBC
decoder — a pure producer. `nodes/telemetry_logger.py` is a consumer: it
calls no `.subscribe()` itself, so wiring it to `engine_data` is purely an
environment-variable exercise:

```bash
BUS_PORT=5561 NODE_NAME=dbc_decoder python3 nodes/dbc_decoder.py &
BUS_PORT=5562 BUS_PEERS=127.0.0.1:5561 BUS_SUBSCRIBES=engine_data,heartbeat \
  NODE_NAME=logger python3 nodes/telemetry_logger.py
```

## Tests

```bash
python -m unittest discover -s tests
```

## Interop with the Ruby repo (optional)

The wire protocol and environment contract are shared — see
[`PROTOCOL.md`](PROTOCOL.md) — so a Python node and a Ruby node
interoperate automatically whenever both are wired into the same
`flow.yml` (in [`ruby_zmq_framework`](https://github.com/pgdaniel/ruby_zmq_framework)),
or by hand:

```python
bus = ZeroMQBus(5561, peer_ports=[5558])  # 5558 = a Ruby StateRegistry node
```

This is entirely optional — nothing here requires the Ruby gem to be
installed or running, and `boot()` works with only Python processes present.
