Examples

Sending Commands

A list of commands that can be sent can be found in the Input Commands section of the protocol documentation. Additional information on the user-level commands available can be found in the Operation Summary of the installer manual.

import asyncio
import contextlib

from nessclient import Client


async def main() -> None:
    host = "127.0.0.1"
    port = 65432
    client = Client(host=host, port=port)
    # Run background loops so responses are received while awaiting
    keepalive_task = asyncio.create_task(client.keepalive())
    try:
        # Send arming command via library abstraction
        await client.arm_away("1234")
        # Send panic command via library abstraction
        await client.panic("1234")
        # Send disarm command via library abstraction
        await client.disarm("1234")
        # Send aux control command for output 2 via library abstraction
        await client.aux_output(2)
        # Send custom command
        # In this instance, we are sending a status update command to view
        # output status
        await client.send_command("S15")
        # Send and await a status request to receive the decoded response
        # note: the background keepalive must be running for this to work
        resp = await client.send_command_and_wait("S14", timeout=5.0)
        print("Arming status response:", resp)
    finally:
        await client.close()
        keepalive_task.cancel()
        with contextlib.suppress(asyncio.CancelledError):
            await keepalive_task


if __name__ == "__main__":
    asyncio.run(main())

Listening for Events

import asyncio

from nessclient import Client, ArmingState, ArmingMode, BaseEvent


async def main() -> None:
    host = "127.0.0.1"
    port = 65432
    client = Client(host=host, port=port)

    @client.on_zone_change
    def on_zone_change(zone: int, triggered: bool) -> None:
        print("Zone {} changed to {}".format(zone, triggered))

    @client.on_state_change
    def on_state_change(state: ArmingState, _arming_mode: ArmingMode | None) -> None:
        print("Alarm state changed to {}".format(state))

    @client.on_event_received
    def on_event_received(event: BaseEvent) -> None:
        print("Event received:", event)

    try:
        await asyncio.gather(
            client.keepalive(),
            client.update(),
        )
    finally:
        client.close()


if __name__ == "__main__":
    asyncio.run(main())

Streaming Events

The nessclient API also exposes asynchronous streams as an alternative to callbacks. Instead of registering handlers, you can iterate over the following async generators:

• client.stream_events()

• client.stream_state_changes()

• client.stream_zone_changes()

• client.stream_aux_output_changes()

This approach fits naturally into asyncio applications and allows cooperative scheduling via await.

import asyncio

from nessclient import Client


async def main() -> None:
    client = Client(host="127.0.0.1", port=65432)
    events = client.stream_events()
    state_changes = client.stream_state_changes()
    zone_changes = client.stream_zone_changes()
    output_changes = client.stream_aux_output_changes()

    async def print_events() -> None:
        async for event in events:
            print("Event received:", event)

    async def print_state_changes() -> None:
        async for state, mode in state_changes:
            print("State changed:", state, mode)

    async def print_zone_changes() -> None:
        async for zone_id, triggered in zone_changes:
            print("Zone {} changed to {}".format(zone_id, triggered))

    async def print_output_changes() -> None:
        async for output_id, active in output_changes:
            print(f"Output {output_id} changed to {active}")

    try:
        await asyncio.gather(
            print_events(),
            print_state_changes(),
            print_zone_changes(),
            print_output_changes(),
            # Start keepalive, then trigger a one-off update for initial state.
            client.keepalive(),
            client.update(),
        )
    finally:
        await events.aclose()
        await state_changes.aclose()
        await zone_changes.aclose()
        await output_changes.aclose()
        await client.close()


if __name__ == "__main__":
    asyncio.run(main())