Linked Electronic Communication Core (LECC) is a modular Python framework for moving messages across heterogeneous protocols — HTTP, TCP, UDP, MQTT, UART, I2C and whatever you add yourself — with automatic emulation of missing links and redundant delivery through Maskpert.
It draws inspiration from Ray Kurzweil's The Age of Spiritual Machines, which imagines a shared electronic communication language between machines. LECC is a small, practical step in that direction: one message envelope, many wires, and a delivery layer that keeps working when the wires do not.
- 🌐 Multi-protocol — HTTP, TCP, UDP, MQTT, UART/serial, I2C and an in-process memory bus.
- 🪶 Zero runtime dependencies — the core runs on the standard library alone. Hardware and broker support are opt-in extras.
- 👺 Maskpert — fan-out delivery, priority confirmation, a bounded and optionally persistent rescue spool, and automatic retry when links recover.
- 🎭 Automatic emulation — a link that cannot start (no broker, no serial device, port in use) degrades to an emulator instead of taking the process down, and is promoted back to the real transport as soon as it becomes reachable.
- ⚡ Resilience built in — per-transport circuit breakers, exponential backoff with jitter, loop prevention via hop trails, and bounded queues everywhere.
- 🔭 Observable — structured logging (text or JSON), per-transport metrics, a
/healthand/metricsendpoint, and alecc doctorcommand. - 🧩 Extensible — a new protocol is three methods and one decorator.
- ✅ Tested — a full suite that exercises real loopback sockets, not mocks.
pip install -e . # core, zero dependencies
pip install -e '.[mqtt]' # + MQTT support (paho-mqtt)
pip install -e '.[serial]' # + UART support (pyserial)
pip install -e '.[i2c]' # + I2C support (smbus2, Linux)
pip install -e '.[yaml]' # + YAML config (PyYAML)
pip install -e '.[all,dev]' # everything + test toolingRequires Python 3.9 or newer. The core is platform-independent; I2C requires Linux, and UART requires a serial device.
from lecc import LECCConfig, LECCCore, Message
config = LECCConfig.from_mapping({
"transports": {
"http": {"port": 5000, "host": "127.0.0.1"},
"tcp": {"port": 65433, "host": "127.0.0.1"},
"udp": {"port": 65434, "host": "127.0.0.1"},
},
"maskpert": {"priority": ["http", "tcp"]},
})
with LECCCore(config) as core:
@core.on_message
def handle(message: Message) -> None:
print(f"{message.source} -> {message.data}")
report = core.send({"temp": 21.5}, destination="tcp")
print(report.summary()) # delivered via tcp, confirmed by tcp
core.broadcast("hello everyone")lecc run --echo # start the core, log every inbound message
lecc run -c config.yaml # run from a config file
lecc send "hello" --destination tcp # send one message and exit
lecc send '{"data":{"n":1}}' --json # send a structured payload
lecc status # JSON snapshot of every transport
lecc doctor # diagnose deps, ports and link health
lecc transports # list available transport kinds
lecc config # print the effective configurationlecc doctor is the fastest way to understand what your machine can actually do:
Optional dependencies
[ok] MQTT
[--] UART — install with: pip install 'lecc[serial]'
[--] I2C — install with: pip install 'lecc[i2c]'
[ok] YAML config
Transports
[ok] http 127.0.0.1:5000 healthy
[ok] tcp 127.0.0.1:65433 healthy
[em] mqtt 127.0.0.1:1883 emulated
[em] uart /dev/ttyUSB0 emulated
4 configured, 0 degraded
Maskpert 👺 (Masked Expert) is the delivery layer. Given a message and a set of links of varying health, it maximises the odds of arrival:
- Candidate selection — a directed message goes to its destination; a broadcast fans out across every healthy link. Links are skipped when they are the message's own source, when the hop trail shows the message has already been there (loop prevention), or when their circuit breaker is open.
- Priority ordering — links named in
maskpert.priorityare tried first. - Fan-out — the message is sent across every remaining candidate. Each link receives its own copy with its own hop trail.
- Confirmation — delivery is confirmed when a priority link accepted the message, and unconfirmed when only secondary links did. Both are reported.
- Rescue — if no link accepted it, the message is spooled. A background loop retries the
spool every
rescue_intervalseconds, so anything queued during an outage is delivered as soon as a link returns. The spool is bounded, age-limited, de-duplicated by message id, and can be persisted to disk so it survives a restart.
report = core.send("critical reading")
report.ok # True if any link accepted it
report.delivered # ['http', 'tcp']
report.failed # {'udp': 'udp: sendto failed: ...'}
report.skipped # {'mqtt': 'unavailable'}
report.confirmed_by # 'http'
report.spooled # True if it went to the rescue queueWhen a transport cannot start — no MQTT broker running, no /dev/ttyUSB0, an unknown protocol
kind — LECC substitutes an emulator that keeps the transport's name, accepts traffic and
buffers it. Your routing rules, logs and metrics stay meaningful, and the process keeps
running. A background health loop retries the real transport; the moment it works, the
emulator is retired and its buffered messages are flushed through the real link.
Set emulate_on_failure: false on a transport if you would rather it be absent than faked.
Configuration comes from a Python mapping, a JSON file, a YAML file, or LECC_* environment
variables (LECC_CONFIG points at a file; LECC_LOG_LEVEL, LECC_WORKERS, etc. override
individual fields). See examples/config.example.yaml for a
fully commented reference.
log_level: INFO
log_format: text # or "json" for log shippers
workers: 4 # inbound handler threads
auto_forward: false # true turns the core into a protocol bridge
transports:
http:
port: 5000
host: 127.0.0.1
url: http://127.0.0.1:5000/api/data
mqtt:
host: 127.0.0.1
port: 1883
topic: lecc/telemetry
emulate_on_failure: true
maskpert:
priority: [http, tcp]
fanout: true
rescue_interval: 5.0
spool_path: ./spool.jsonl
retry:
max_attempts: 3
base_delay: 0.5
multiplier: 2.0
jitter: 0.1
circuit_breaker:
failure_threshold: 5
reset_timeout: 30.0Invalid configuration is rejected at load time with a clear message — including port collisions between two transports.
Implement _start, _stop and _send, then register the class. _send must raise on
failure; that is the signal Maskpert uses to fall back.
from lecc.exceptions import TransportUnavailable
from lecc.message import Message
from lecc.transports import Transport, register
@register("carrier-pigeon")
class PigeonTransport(Transport):
def _start(self) -> None:
self._loft = open_loft(self.config.options["loft"])
self._spawn(self._read_loop, "reader") # tracked, joined on stop
def _stop(self) -> None:
self._loft.close()
def _send(self, message: Message) -> None:
if not self._loft.has_pigeons():
raise TransportUnavailable(self.name, "no pigeons available")
self._loft.dispatch(message.to_bytes())
def _read_loop(self) -> None:
while not self._closing.wait(0.1):
for raw in self._loft.arrivals():
self._emit(Message.from_json(raw))Retries, circuit breaking, metrics, threading and shutdown are inherited. See
examples/custom_transport.py for a runnable version.
┌──────────────────────────────┐
inbound ──────────► │ LECCCore │
(any transport) │ worker pool → your handlers │
└───────────────┬──────────────┘
│ dispatch
┌───────▼────────┐
│ Maskpert │ fan-out, confirmation,
│ rescue spool │ loop prevention
└───────┬────────┘
┌───────────┬───────────┬───┴───────┬───────────┐
┌──▼──┐ ┌───▼──┐ ┌───▼──┐ ┌───▼──┐ ┌───▼───┐
│ HTTP│ │ TCP │ │ UDP │ │ MQTT │ │ UART │ each with a
└─────┘ └──────┘ └──────┘ └──────┘ └───────┘ circuit breaker,
│ │ │ │ │ retry policy and
└──────────┴───────────┴───────────┴───────────┘ emulation fallback
| Module | Responsibility |
|---|---|
lecc/message.py |
Immutable message envelope, hop trails, serialisation |
lecc/config.py |
Declarative config, validation, retry and breaker policies |
lecc/transports/ |
Protocol implementations behind one Transport contract |
lecc/maskpert.py |
Redundant delivery, confirmation, rescue spool |
lecc/core.py |
Lifecycle, worker pool, routing, health and promotion loops |
lecc/resilience.py |
Circuit breaker state machine, metrics |
lecc/cli.py |
run, send, status, doctor, transports, config |
- Framing — TCP and UART use newline-delimited JSON, so messages of any size survive segment boundaries. UDP carries one message per datagram and rejects payloads above 65,507 bytes rather than truncating them silently.
- HTTP —
POST /api/dataingests messages;GET /healthreturns a full system snapshot (503 when degraded, so it works as a container readiness probe);GET /metricsreturns counters. - MQTT — needs a broker (e.g. Mosquitto) on the configured host. Without one, the link is emulated.
- I2C — Linux only, write-oriented, payloads chunked to fit SMBus's 32-byte transactions.
- Bridging —
auto_forward: truere-publishes every inbound message to the other transports, turning a LECC instance into a protocol bridge. Hop trails prevent loops between bridged instances.
pip install -e '.[all,dev]'
pytest # full suite
pytest --cov=lecc # with coverage
ruff check . && ruff format --check .
mypy leccThe suite runs against real loopback sockets and a real HTTP server, with no network access and no hardware required.
Contributions are welcome:
- Fork the repository.
- Create a branch (
git checkout -b feature/new-protocol). - Add tests for your change and make sure
pytest,ruffandmypypass. - Open a pull request.
See CONTRIBUTING.md for details and CHANGELOG.md for the release history.
Distributed under the MIT License. See LICENSE.txt.
| Name | Role | GitHub | |
|---|---|---|---|
 |
Alejo | Author & Development | @alejomalia |
Inspired by The Age of Spiritual Machines by Ray Kurzweil.