knx-telegram-store

A standalone, host-agnostic Python library for KNX telegram persistence.

Features

• Canonical Data Model: A unified model for KNX telegrams shared between Home Assistant and SpectrumKNX.
• Pluggable Backends:
  • In-Memory: Fast, deque-based storage with full filtering support.
  • SQLite: Lightweight persistent storage with SQL-based filtering.
  • PostgreSQL: Full-scale storage. TimescaleDB is used automatically when the extension is available (hypertable partitioning + native compression); otherwise the store runs on plain PostgreSQL with identical semantics.
• Unified Query Model: Powerful declarative filtering including time-delta context windows and pagination.
• Stats & Maintenance: get_stats() reports count, covered time range and on-disk size; evict_older_than() supports dry runs; optimize() reclaims disk space (VACUUM).
• Read-Only Mode: Open a SQLite store owned and written by another process (e.g. Home Assistant's KNX telegram store) without running migrations or allowing writes.
• Concurrent Access: Writing SQLite stores use WAL journaling and a busy timeout, so a single writer and multiple (cross-process) readers coexist safely.
• Capability Flags: store.capabilities declares what a backend supports (supports_optimize, supports_size_stats, read_only, …) so hosts can gate UI instead of hardcoding backends.
• Log Container Format: formats.ets_xml streams the KNX CommunicationLog XML container (ETS6 group-monitor exports, Gira IP-Router data-logger dumps) to/from raw cEMI frames — constant memory, no protocol decoding, stdlib-only.
• Zero Runtime Dependencies: Core library (model, interface, in-memory) has no dependencies.
• Automated Schema Management: SQL backends handle their own creation and upgrades.

Installation

pip install knx-telegram-store

For SQL support:

pip install knx-telegram-store[sqlite]
pip install knx-telegram-store[postgres]

Usage

from datetime import datetime
from knx_telegram_store import StoredTelegram, TelegramQuery
from knx_telegram_store.backends.memory import MemoryStore


async def main():
    store = MemoryStore(max_size=1000)
    await store.initialize()

    telegram = StoredTelegram(
        timestamp=datetime.now(),
        source="1.1.1",
        destination="1/1/1",
        telegramtype="GroupValueWrite",
        direction="Incoming",
        value=22.5,
        unit="°C",
    )

    await store.store(telegram)

    query = TelegramQuery(destinations=["1/1/1"])
    result = await store.query(query)

    for t in result.telegrams:
        print(f"{t.timestamp}: {t.source} -> {t.destination} | {t.value} {t.unit}")

    await store.close()

Write-buffered stores

BufferedSqliteStore, BufferedPostgresStore and BufferedMemoryStore wrap the backend of the same name in an in-memory write buffer. store() and store_sync() only append to it and never touch the database; a periodic task started by start() drains it with store_many(), and so do flush(), close() and the reads that need to see every write (get_last_unique_telegrams(), get_stats(), query(..., flush_first=True)). The buffer holds at most max_buffer_size telegrams and drops the oldest beyond that.

from knx_telegram_store import BufferedSqliteStore

store = BufferedSqliteStore("/data/telegrams.db", flush_interval=600, max_buffer_size=120_000)
store.store_sync(telegram)  # allowed before initialize(): buffered, written by the first flush
await store.initialize()
store.start()
...
await store.close()  # flushes what is left, then closes the backend

Telegrams may be handed to the store before initialize() has run, or while it is running, and are written by the first flush after it. That lets a host keep recording while a long startup pass (see needs_migration()) or a retried initialization is under way. close() on a store whose initialize() never completed has no schema to write to, so it drops the buffer with a warning and only closes the backend; flush() and the flushing reads above are the caller's to avoid until initialize() has returned.

Stats, purging and space reclamation

from datetime import UTC, datetime, timedelta
from knx_telegram_store.backends.sqlite import SqliteStore

store = SqliteStore("/data/telegrams.db", retention_days=90)
await store.initialize()

stats = await store.get_stats()
print(
    f"{stats.telegram_count} telegrams, {stats.size_bytes} bytes, {stats.oldest_timestamp} .. {stats.newest_timestamp}"
)

cutoff = datetime.now(UTC) - timedelta(days=30)
would_delete = await store.evict_older_than(cutoff, dry_run=True)  # preview only
deleted = await store.evict_older_than(cutoff)

# Deleting rows does not shrink the database on disk by itself:
if store.capabilities.supports_optimize:
    await store.optimize()  # VACUUM — blocks writers, can take a while on large DBs

Read-only access to a shared store

Another process (e.g. Home Assistant's KNX integration) owns and writes the database; you only want to read it:

store = SqliteStore("/homeassistant/.storage/knx/telegrams.db", read_only=True)
await store.initialize()  # never runs DDL/migrations against a foreign schema

if await store.needs_migration():
    ...  # schema is older/newer than this library version — surface a warning

result = await store.query(TelegramQuery(limit=100))
await store.store(telegram)  # raises KnxTelegramStoreException — writes rejected

The file is opened with SQLite's mode=ro, so writes are impossible at the driver level. capabilities.read_only is True and supports_optimize is False in this mode. Writing stores enable WAL journaling, which makes this single-writer/multi-reader setup safe across processes.

Backups

A host that copies the SQLite file while the store is running (Home Assistant does, for its backups) calls begin_backup() first and end_backup() once the copy is done. The writing store checkpoints the WAL into the main file, leaving the -wal file empty, and holds a write lock until end_backup(), so the file does not change under the copy. Readers, including read-only stores in other processes, are not affected.

Between the two calls a buffered store keeps incoming telegrams in its buffer (bounded by max_buffer_size) and writes them when the backup has ended. Flushes and flush_first queries return without writing instead of waiting; store_many(), evictions, clear() and optimize() wait for end_backup(). Closing a store during a backup releases the lock. The calls are no-ops for the memory store and the PostgreSQL store, whose data the host does not copy.

await store.begin_backup()
try:
    ...  # copy telegrams.db
finally:
    await store.end_backup()

SQLite file format

Since 0.15 the SQLite backend stores timestamps as integer microseconds since the Unix epoch (UTC) instead of SQLAlchemy's 26-character text format, which makes the file about a quarter smaller. A database written by an earlier version is converted in place on the first initialize() of a writing store: the pass rewrites every row, is reported through needs_migration() so hosts can run that start without a timeout, and ends with a VACUUM, which needs free disk of about the size of the database and blocks writers while it runs. A converted file cannot be read by library versions before 0.15, so a process that only reads another one's database has to be on 0.15 before the owner is. Read-only stores never convert; opened on an unconverted file they report needs_migration() as True until the writing process has upgraded it. Since 0.15.1 they query such a file correctly in the meantime, time ranges included, and carry on when the owner converts it underneath them.

Validating a config / connection

Before triggering an expensive operation such as a migration, you can validate that a store is reachable. Both checks return a structured ConnectionCheckResult (ok, kind, message, detail) instead of raising.

from knx_telegram_store import ConnectionErrorKind
from knx_telegram_store.backends.sqlite import SqliteStore
from knx_telegram_store.backends.postgres import PostgresStore

# Static, side-effect-free config validation (before constructing a store):
#  - SQLite: sync — checks the file is writeable or can be created
result = SqliteStore.check_config("/data/telegrams.db")
#    (with read_only=True: checks the file exists and is readable instead)
result = SqliteStore.check_config("/data/telegrams.db", read_only=True)
#  - Postgres: async — actually connects to verify user/password/host/port/database
result = await PostgresStore.check_config("postgresql://user:pw@host:5432/knx")

if not result.ok:
    print(f"[{result.kind}] {result.message}")  # e.g. [auth] Authentication failed ...

# Live probe of an already-constructed store (no migrations, no schema changes):
store = SqliteStore("/data/telegrams.db")
result = await store.check_connection()
if result.kind is ConnectionErrorKind.OK:
    await store.initialize()

PostgreSQL and TimescaleDB

PostgresStore works against any PostgreSQL server. At initialize() it probes pg_available_extensions: when TimescaleDB is available, the telegrams table becomes a hypertable (existing rows are migrated in place via migrate_data => TRUE) and native compression is configured — chunks are compressed by a background policy once they age past compress_after_days (default 7, None disables compression). Without the extension everything runs on plain PostgreSQL tables; queries, retention and stats behave identically.

store = PostgresStore("postgresql://user:pw@host:5432/knx", retention_days=90, compress_after_days=7)
await store.initialize()
print(store.timescale_enabled)  # True / False (None before initialize())

check_config() / check_connection() succeed on both server types; the result message states which mode will be used.

Integration tests

The Postgres backend has an integration test suite that runs against real servers — a TimescaleDB container and a stock PostgreSQL container — so both the hypertable/compression path and the plain fallback are exercised. With Docker installed:

./scripts/run_integration_tests.sh            # full suite
./scripts/run_integration_tests.sh -k compression  # subset

The script starts both containers (docker-compose.test.yml), waits for them to become healthy, runs pytest -m integration tests/integration, and tears the containers down afterwards. To run tests manually, e.g. against your own servers:

docker compose -f docker-compose.test.yml up -d --wait
export KNX_TEST_TIMESCALE_DSN=postgresql://knx:knxtest@localhost:5433/knx
export KNX_TEST_PG_DSN=postgresql://knx:knxtest@localhost:5434/knx
pytest -m integration tests/integration -v
docker compose -f docker-compose.test.yml down -v

Tests for an unset DSN variable are skipped, so you can also point a single variable at an existing server. The same suite runs in CI against both containers on every push.

Reading / writing telegram log files

formats.ets_xml handles the KNX CommunicationLog XML container (namespace http://knx.org/xml/telegrams/01) produced by ETS6 exports and Gira data loggers. It operates on raw cEMI frames — no protocol decoding, no xknx dependency — so any consumer can stream large logs with constant memory.

from knx_telegram_store.formats import iter_communication_log, write_communication_log

# Incremental read (file path or binary stream, e.g. a zip entry):
for record in iter_communication_log("2026_03_05_TP1.xml"):
    print(record.timestamp, record.service, record.raw_data.hex())
    #      aware UTC        "L_Data.ind"   cEMI frame as logged

# Streaming write (records may be a generator; ETS6-compatible output):
with open("export.xml", "w", encoding="utf-8") as fh:
    count = write_communication_log(records, fh, connection_name="My Export")

A Gira-style <!-- timezone offset +01:00 hour --> comment is honored, and ETS's 7-digit fractional seconds are normalized to microseconds.

MCP tools

knx_telegram_store.mcp provides host-agnostic tool functions for exposing the store to AI agents over the Model Context Protocol. They are plain async functions over a TelegramStore, with frozen, JSON-serialisable dataclass inputs/outputs (timestamps are ISO-8601 UTC strings) and no dependency on any MCP SDK or web framework — each consumer wraps them into its own transport.

from dataclasses import asdict
from knx_telegram_store.mcp import query_telegrams, QueryTelegramsInput

result = await query_telegrams(store, QueryTelegramsInput(destinations=["1/1/1"], limit=100))
payload = asdict(result)  # ready to return as an MCP tool result

Available: query_telegrams, get_last_values, get_store_stats, get_store_capabilities, count_telegrams.

License

MIT