Async Trinnov Altitude client for long-running integrations (Home Assistant primary target).
Version 2.x is a clean break from 1.x.
- No compatibility shims
- New lifecycle (
start/wait_synced/stop) - New state model (
client.state) - Optional command ACK handling
Read the migration guide: docs/MIGRATION_V2.md
pip install trinnov-altitudeimport asyncio
from trinnov_altitude.client import TrinnovAltitudeClient
async def main() -> None:
client = TrinnovAltitudeClient(host="192.168.1.90")
try:
await client.start()
await client.wait_synced(timeout=10)
await client.volume_set(-30.0)
await client.mute_on()
print(client.state.volume)
print(client.state.source)
finally:
await client.stop()
asyncio.run(main())await client.start()connects, bootstraps, and starts the read loop.await client.wait_synced()waits until welcome + catalogs + current indices are observed.await client.stop()stops listener and disconnects cleanly.
The control connection is a long-lived TCP push session, so a silent read is
ambiguous: a healthy link is quiet whenever nothing is changing, but a dead link
(an idle-killed half-open socket, or a processor whose control thread has wedged
while its TCP stack still ACKs) looks identical. To tell them apart, when the link
has been quiet for heartbeat_interval seconds the client sends a read-only probe
(get_current_state); if no traffic arrives within heartbeat_timeout, the link is
treated as dead and auto_reconnect kicks in. TcpTransport also enables
SO_KEEPALIVE as an OS-level backstop.
client = TrinnovAltitudeClient(
host="192.168.1.90",
heartbeat_interval=20.0, # seconds of quiet before a liveness probe (None disables)
heartbeat_timeout=5.0, # seconds to wait for the probe response before reconnecting
)Push messages remain the fast path for state updates. As a backstop for an
individual dropped notification, the client also requests the processor's
current state every reconcile_interval seconds. This schedule is independent
of ordinary push traffic, so unrelated messages cannot leave one cached field
stale indefinitely.
client = TrinnovAltitudeClient(
host="192.168.1.90",
reconcile_interval=30.0, # None disables periodic reconciliation
)The client parses raw messages first, then normalizes them into canonical state events. This keeps protocol quirks isolated and keeps the state reducer deterministic.
- Canonical identity:
CURRENT_PRESET <n>CURRENT_PROFILE <n>or index-onlyPROFILE <n>DECODER ... UPMIXER <mode>
- Optional catalogs:
- Presets via
LABELS_CLEAR+LABEL <n>: <name> - Sources via
PROFILES_CLEAR+PROFILE <n>: <name>
- Presets via
- Quirk profiles:
altitude_ciis selected whenIDENTSincludesaltitude_ci- In that profile,
META_PRESET_LOADED <n>is normalized as a source-change signal
Catalog messages may arrive late, be refreshed, or be absent. Consumers should not assume labels are always present.
def on_event(event, message):
if event == "connected":
...
elif event == "disconnected":
...
elif event == "received_message":
...
client.register_callback(on_event)Callback exceptions are isolated and logged (they do not crash the listener).
Use trinnov_altitude.adapter.AltitudeStateAdapter to convert mutable runtime state into immutable snapshots plus typed deltas/events:
snapshot: stable full-state view for coordinator datadeltas: field-level changes since previous snapshotevents: integration-friendly event stream (volume, mute, source, preset, etc.)
You can wire this directly through the client:
from trinnov_altitude.adapter import AltitudeStateAdapter
adapter = AltitudeStateAdapter()
def on_update(snapshot, deltas, events):
...
handle = client.register_adapter_callback(adapter, on_update)
# later: client.deregister_adapter_callback(handle)For Home Assistant coordinator/event-bus integration, use trinnov_altitude.ha_bridge:
coordinator_payload(snapshot)to_ha_events(events)build_bridge_update(snapshot, deltas, events)
You can use fire-and-forget commands (default) or explicit ACK waiting:
await client.volume_set(-20.0)
await client.command("volume -20", wait_for_ack=True, ack_timeout=2.0)uv sync --group dev
uv run ruff check .
uv run ruff format --check .
uv run ty check trinnov_altitude
uv run pytest -vOr use task wrappers:
task dev
task checkThe test suite includes a manual, read-only integration tier for validating behavior against real hardware.
- Marker:
integration_real - Opt-in gate:
TRINNOV_ITEST=1 - Target host:
TRINNOV_HOST=<ip-or-hostname> - Optional port override:
TRINNOV_PORT=44100 - If the device is offline/unreachable, tests are skipped.
These tests intentionally avoid mutating commands (no power/preset/source/volume state changes).
TRINNOV_ITEST=1 TRINNOV_HOST=192.168.30.3 task test:integration-realPyx is optional in this repo. You can keep publishing to PyPI/TestPyPI only.
- Install via Pyx: authenticate
uvwithPYX_API_KEYand configure your Pyx index URL inuv(uv add --index .../uv sync). - Publish to Pyx: run the
Releaseworkflow manually withtarget=pyxafter setting repository secretsPYX_API_KEYandPYX_PUBLISH_URL. - No dual-publish requirement: use Pyx when you need private/internal package distribution or policy control.
- Merge conventional-commit changes to
master. - Wait for the
release-pleaseworkflow to open/update a release PR. - Review and merge the release PR (this updates
CHANGELOG.mdand__version__). - Release Please creates the GitHub Release and tag.
- The
Releaseworkflow publishes artifacts to PyPI automatically for published releases. - For TestPyPI or Pyx-only publishing, run
Releasemanually withworkflow_dispatch.
- Migration guide: docs/MIGRATION_V2.md
- Maintainer runbook: docs/MAINTAINERS.md
- Protocol reference used for implementation: docs/Altitude Protocol.pdf (v1.15, 2019-04-19)