Transport ¶

Returns:

Complete BVLL message bytes ready for UDP transmission.

bac_py.transport.bvll.decode_bvll(data)[source]¶

Decode a BVLL message from raw UDP datagram.

Parameters:: data (memoryview | bytes) – Raw UDP datagram bytes.
Return type:: BvllMessage
Returns:: Decoded BvllMessage.
Raises:: ValueError – If data is too short, BVLC type byte is invalid, declared length is inconsistent, or a function-specific payload (e.g. Forwarded-NPDU originating address) is truncated.

BACnet/IP¶

BACnet/IP transport using asyncio UDP per Annex J.

exception bac_py.transport.bip.BvlcNakError(result_code, source)[source]¶

Bases: Exception

Raised when a BVLC-Result NAK is received for a pending request.

Parameters:

result_code (int)
source (BIPAddress)

Return type:

None

class bac_py.transport.bip.BIPTransport(interface='0.0.0.0', port=47808, broadcast_address='255.255.255.255', *, multicast_enabled=False, multicast_address='239.255.186.192', multicast_ttl=32)[source]¶

Bases: object

BACnet/IP transport using asyncio UDP.

Provides send/receive for BACnet/IP datagrams wrapped in BVLL.

Parameters:

interface (str)
port (int)
broadcast_address (str)
multicast_enabled (bool)
multicast_address (str)
multicast_ttl (int)

async start()[source]¶

Bind UDP socket and start listening.

Return type:: None

async stop()[source]¶

Close UDP socket and stop BBMD/foreign device if attached.

Return type:: None

on_receive(callback)[source]¶

Parameters:: callback (Callable[[bytes, bytes], None]) – Called with (npdu_bytes, source_mac) for each received datagram containing an NPDU. source_mac is the 6-byte BACnet/IP MAC (4-byte IP + 2-byte port).
Return type:: None

send_unicast(npdu, mac_address)[source]¶

Send a directed message (Original-Unicast-NPDU).

Parses the 6-byte MAC inline to avoid BIPAddress object creation on the hot path.

Parameters:

npdu (bytes) – NPDU bytes to send.
mac_address (bytes) – 6-byte destination MAC (4-byte IP + 2-byte port).

Return type:

send_broadcast(npdu)[source]¶

Send a local broadcast (Original-Broadcast-NPDU).

If registered as a foreign device, uses Distribute-Broadcast-To-Network instead per Annex J.5.6. If a BBMD is attached, also forwards to BDT peers and registered foreign devices per Annex J.4.5.

Parameters:: npdu (bytes) – NPDU bytes to broadcast.
Return type:: None

property local_address: BIPAddress¶: The local BACnet/IP address of this transport.

property local_mac: bytes¶: The 6-byte MAC address of this port (4-byte IP + 2-byte port).

property max_npdu_length: int¶: Maximum NPDU length for BACnet/IP per Clause 6.

property bbmd: BBMDManager | None¶: The attached BBMD manager, or None if not configured.

async attach_bbmd(bdt_entries=None)[source]¶

Attach a BBMD manager to this transport.

Creates and starts a BBMDManager integrated with this transport. The BBMD intercepts incoming BVLC messages before they reach the normal receive path, and outgoing broadcasts are also forwarded to BDT peers and foreign devices.

Per Annex J.7.1 this allows a single device to combine BBMD and router functionality.

Parameters:: bdt_entries (list[BDTEntry] | None (default: None)) – Optional initial BDT entries. If None, the BBMD starts with an empty BDT.
Return type:: BBMDManager
Returns:: The attached BBMDManager instance.
Raises:: RuntimeError – If transport not started or BBMD already attached.

property foreign_device: ForeignDeviceManager | None¶: The attached foreign device manager, or None.

async attach_foreign_device(bbmd_address, ttl)[source]¶

Attach a foreign device manager to this transport.

Creates and starts a ForeignDeviceManager that will register with the specified BBMD and periodically re-register to maintain the registration.

Incoming BVLC-Result messages will be routed to the manager to track registration status.

Parameters:

bbmd_address (BIPAddress) – Address of the BBMD to register with.
ttl (int) – Time-to-Live for the registration in seconds.

Return type:

ForeignDeviceManager

Returns:

The attached ForeignDeviceManager instance.

Raises:

RuntimeError – If transport not started or foreign device already attached.

async read_bdt(bbmd_address, *, timeout=5.0)[source]¶

Read the Broadcast Distribution Table from a remote BBMD.

Sends a Read-Broadcast-Distribution-Table request and waits for the Read-BDT-Ack response.

Parameters:

bbmd_address (BIPAddress) – Address of the BBMD to query.
timeout (float (default: 5.0)) – Seconds to wait for a response.

Return type:

list[BDTEntry]

Returns:

List of BDTEntry instances from the remote BBMD.

Raises:

RuntimeError – If transport not started.
BvlcNakError – If the target is not a BBMD.
TimeoutError – If no response within timeout.

async write_bdt(bbmd_address, entries, *, timeout=5.0)[source]¶

Write a Broadcast Distribution Table to a remote BBMD.

Sends a Write-Broadcast-Distribution-Table request and waits for the BVLC-Result response.

Parameters:

bbmd_address (BIPAddress) – Address of the BBMD to configure.
entries (list[BDTEntry]) – BDTEntry instances to write.
timeout (float (default: 5.0)) – Seconds to wait for a response.

Return type:

BvlcResultCode

Returns:

The BvlcResultCode from the BBMD.

Raises:

RuntimeError – If transport not started.
TimeoutError – If no response within timeout.

async read_fdt(bbmd_address, *, timeout=5.0)[source]¶

Read the Foreign Device Table from a remote BBMD.

Sends a Read-Foreign-Device-Table request and waits for the Read-FDT-Ack response.

Parameters:

bbmd_address (BIPAddress) – Address of the BBMD to query.
timeout (float (default: 5.0)) – Seconds to wait for a response.

Return type:

list[FDTEntry]

Returns:

List of FDTEntry instances from the remote BBMD. The expiry field is set to 0.0 since it is not meaningful for remote entries; use the remaining property instead.

Raises:

RuntimeError – If transport not started.
BvlcNakError – If the target is not a BBMD.
TimeoutError – If no response within timeout.

async delete_fdt_entry(bbmd_address, entry_address, *, timeout=5.0)[source]¶

Delete a Foreign Device Table entry on a remote BBMD.

Sends a Delete-Foreign-Device-Table-Entry request and waits for the BVLC-Result response.

Parameters:

bbmd_address (BIPAddress) – Address of the BBMD.
entry_address (BIPAddress) – Address of the FDT entry to delete.
timeout (float (default: 5.0)) – Seconds to wait for a response.

Return type:

BvlcResultCode

Returns:

The BvlcResultCode from the BBMD.

Raises:

RuntimeError – If transport not started.
TimeoutError – If no response within timeout.

BBMD¶

BACnet/IP Broadcast Management Device (BBMD) per Annex J.4-J.5.

Provides BBMDManager for BDT/FDT management, broadcast forwarding between BACnet/IP subnets, and foreign device registration handling.

class bac_py.transport.bbmd.BDTEntry(address, broadcast_mask)[source]¶

Bases: object

Broadcast Distribution Table entry per Annex J.4.

Each entry contains the B/IP address of a BBMD peer and a 4-octet broadcast distribution mask used to compute the forwarding address.

Parameters:

address (BIPAddress)
broadcast_mask (bytes)

address: BIPAddress¶

broadcast_mask: bytes¶

encode()[source]¶

Encode to 10-byte wire format.

Return type:: bytes

classmethod decode(data)[source]¶

Decode from 10-byte wire format.

Return type:: BDTEntry
Parameters:: data (bytes | memoryview)

class bac_py.transport.bbmd.FDTEntry(address, ttl, expiry)[source]¶

Bases: object

Foreign Device Table entry per Annex J.5.2.

Tracks a registered foreign device with its TTL and the absolute time at which the entry expires.

Parameters:

address (BIPAddress)
ttl (int)
expiry (float)

address: BIPAddress¶

ttl: int¶

expiry: float¶

property remaining: int¶

Seconds remaining before this entry expires.

Capped at 65535 per J.5.2.1 (2-octet wire encoding).

class bac_py.transport.bbmd.BBMDManager(local_address, send_callback, local_broadcast_callback=None, broadcast_address=None, max_fdt_entries=128, max_bdt_entries=128, accept_fd_registrations=True, allow_write_bdt=False, global_address=None, bdt_backup_path=None, fdt_cleanup_interval=10.0)[source]¶

Bases: object

BACnet/IP Broadcast Management Device per Annex J.4-J.5.

Manages BDT (Broadcast Distribution Table) and FDT (Foreign Device Table), handles broadcast forwarding between BACnet/IP subnets, and processes foreign device registration requests.

The BBMD must be wired into a BIPTransport to intercept and process BVLC messages before they reach the application layer.

Parameters:

local_address (BIPAddress)
send_callback (Callable[[bytes, BIPAddress], None])
local_broadcast_callback (Callable[[bytes, BIPAddress], None] | None)
broadcast_address (BIPAddress | None)
max_fdt_entries (int)
max_bdt_entries (int)
accept_fd_registrations (bool)
allow_write_bdt (bool)
global_address (BIPAddress | None)
bdt_backup_path (Path | None)
fdt_cleanup_interval (float)

property bdt: list[BDTEntry]¶: Current Broadcast Distribution Table.

property fdt: dict[BIPAddress, FDTEntry]¶: Current Foreign Device Table.

property accept_fd_registrations: bool¶: Whether foreign device registrations are accepted.

property global_address: BIPAddress | None¶: Optional public/NAT address per Annex J.7.8.

set_bdt(entries)[source]¶

Set the Broadcast Distribution Table.

Parameters:: entries (list[BDTEntry]) – New BDT entries. Should include this BBMD’s own entry.
Return type:: None

async start()[source]¶

Start the FDT cleanup background task.

If a bdt_backup_path was configured and the file exists, the BDT is restored from it before starting.

Return type:: None

async stop()[source]¶

Stop the FDT cleanup background task.

Return type:: None

handle_bvlc(function, data, source, *, udp_source=None)[source]¶

Process a BVLC message directed at the BBMD.

Parameters:

function (BvlcFunction) – BVLC function code.
data (bytes) – Payload after BVLL header (for most functions) or full payload including originating address (for Forwarded-NPDU, which is pre-parsed by decode_bvll()).
source (BIPAddress) – For most functions this is the UDP source address. For FORWARDED_NPDU this is the originating address extracted from the BVLL header.
udp_source (BIPAddress | None (default: None)) – The actual UDP peer address. Only needed for FORWARDED_NPDU where source is the originating address. Used for BDT mask lookup to decide whether to re-broadcast the NPDU on the local wire.

Return type:

Returns:

True if the message was fully consumed by the BBMD and should not be delivered locally. False if the NPDU should also be processed by the normal receive path (this is the case for Original-Broadcast-NPDU and Forwarded-NPDU, which are forwarded and delivered locally).

Foreign Device¶

Foreign device registration manager per Annex J.5-J.6.

Provides ForeignDeviceManager for registering with a remote BBMD, periodic re-registration, and broadcast distribution via the BBMD.

class bac_py.transport.foreign_device.ForeignDeviceManager(bbmd_address, ttl, send_callback, local_address=None)[source]¶

Bases: object

Manages registration with a remote BBMD per Annex J.5-J.6.

A BACnet/IP device on a different subnet registers with a BBMD as a “foreign device” to receive broadcast traffic. This manager handles:

Initial registration with the BBMD
Periodic re-registration at TTL/2 intervals to prevent expiry
Tracking registration state (registered, failed)
Sending broadcasts via Distribute-Broadcast-To-Network

Usage:

fd_mgr = ForeignDeviceManager(
    bbmd_address=BIPAddress("192.168.1.1", 47808),
    ttl=60,
    send_callback=transport_send,
)
await fd_mgr.start()
# ...later...
await fd_mgr.stop()

Parameters:

bbmd_address (BIPAddress)
ttl (int)
send_callback (Callable[[bytes, BIPAddress], None])
local_address (BIPAddress | None)

property bbmd_address: BIPAddress¶: The BBMD address this device is registered with.

property ttl: int¶: The registration TTL in seconds.

property is_registered: bool¶: Whether the device is currently registered with the BBMD.

property last_result: BvlcResultCode | None¶: The last BVLC-Result code received from the BBMD.

async start()[source]¶

Start the registration loop.

Sends an initial registration and begins periodic re-registration at TTL/2 intervals.

Return type:: None

async stop()[source]¶

Stop the registration loop and deregister from the BBMD.

If the device is currently registered and a local address was provided, sends a Delete-Foreign-Device-Table-Entry to the BBMD so it can immediately remove the FDT entry rather than waiting for TTL + grace period expiry.

Return type:: None

handle_bvlc_result(data)[source]¶

Process a BVLC-Result received from the BBMD.

Should be called when a BVLC-Result is received from the BBMD address after a registration attempt.

Parameters:: data (bytes) – 2-octet result code payload.
Return type:: None

send_distribute_broadcast(npdu)[source]¶

Send a broadcast via the BBMD using Distribute-Broadcast-To-Network.

This is used by foreign devices instead of local broadcast. The BBMD will distribute the NPDU to all BDT peers and other registered foreign devices.

Parameters:: npdu (bytes) – NPDU bytes to broadcast.
Raises:: RuntimeError – If not registered with a BBMD.
Return type:: None

BACnet/IPv6 BVLL¶

BVLL (BACnet Virtual Link Layer) encoding and decoding for BACnet/IPv6 per Annex U.

class bac_py.transport.bvll_ipv6.Bvll6Message(function, data, source_vmac=None, dest_vmac=None, originating_address=None)[source]¶

Bases: object

Decoded BACnet/IPv6 BVLL message.

Parameters:

function (Bvlc6Function)
data (bytes)
source_vmac (bytes | None)
dest_vmac (bytes | None)
originating_address (BIP6Address | None)

function: Bvlc6Function¶

data: bytes¶

source_vmac: bytes | None¶

dest_vmac: bytes | None¶

originating_address: BIP6Address | None¶

bac_py.transport.bvll_ipv6.encode_bvll6(function, payload, *, source_vmac=None, dest_vmac=None, originating_address=None)[source]¶

Encode a complete BACnet/IPv6 BVLL message.

Uses a single pre-sized bytearray to avoid intermediate allocations.

Parameters:

function (Bvlc6Function) – BVLC6 function code.
payload (bytes) – NPDU payload bytes (or result/registration data).
source_vmac (bytes | None (default: None)) – 3-byte source VMAC (required for most functions).
dest_vmac (bytes | None (default: None)) – 3-byte destination VMAC (required for unicast and ACKs).
originating_address (BIP6Address | None (default: None)) – 18-byte originating IPv6 address (Forwarded-NPDU).

Return type:

Returns:

Complete BVLL message bytes ready for UDP transmission.

bac_py.transport.bvll_ipv6.decode_bvll6(data)[source]¶

Decode a BACnet/IPv6 BVLL message from raw UDP datagram.

Parameters:: data (memoryview | bytes) – Raw UDP datagram bytes.
Return type:: Bvll6Message
Returns:: Decoded Bvll6Message.
Raises:: ValueError – If data is too short, type byte is invalid, or payload is truncated.

BACnet/IPv6 Transport¶

BACnet/IPv6 transport using asyncio UDP per Annex U.

class bac_py.transport.bip6.VMACEntry(address, last_seen)[source]¶

Bases: object

A cached mapping from VMAC to IPv6 address.

Parameters:

address (BIP6Address)
last_seen (float)

address: BIP6Address¶

last_seen: float¶

class bac_py.transport.bip6.VMACCache(ttl=300.0, max_entries=4096)[source]¶

Bases: object

VMAC-to-IPv6 address resolution cache with TTL-based eviction.

Parameters:

ttl (float)
max_entries (int)

put(vmac, address)[source]¶

Add or update a VMAC-to-address mapping.

Return type:

Parameters:

vmac (bytes)
address (BIP6Address)

get(vmac)[source]¶

Look up an address by VMAC, returning None if not cached or stale.

Return type:: BIP6Address | None
Parameters:: vmac (bytes)

evict_stale()[source]¶

Remove all entries older than the TTL.

Return type:: None

all_entries()[source]¶

Return all current entries (for diagnostics).

Return type:: dict[bytes, VMACEntry]

class bac_py.transport.bip6.BIP6Transport(interface='::', port=47808, multicast_address='ff02::bac0', *, vmac=None, vmac_ttl=300.0)[source]¶

Bases: object

BACnet/IPv6 transport using asyncio UDP.

Provides send/receive for BACnet/IPv6 datagrams wrapped in BVLL6. Uses 3-byte VMACs for network-layer addressing and IPv6 multicast for broadcasts per Annex U.

Parameters:

interface (str)
port (int)
multicast_address (str)
vmac (bytes | None)
vmac_ttl (float)

async start()[source]¶

Bind UDP6 socket, generate VMAC, and join multicast group.

Return type:: None

async stop()[source]¶

Stop BBMD/FD managers, leave multicast group, and close UDP socket.

Return type:: None

on_receive(callback)[source]¶

Parameters:: callback (Callable[[bytes, bytes], None]) – Called with (npdu_bytes, source_vmac) for each received datagram containing an NPDU. source_vmac is the 3-byte VMAC of the sender.
Return type:: None

send_unicast(npdu, mac_address)[source]¶

Send a directed message (Original-Unicast-NPDU).

Parameters:

npdu (bytes) – NPDU bytes to send.
mac_address (bytes) – 3-byte destination VMAC.

Return type:

send_broadcast(npdu)[source]¶

Send a local broadcast (Original-Broadcast-NPDU) to the multicast group.

If registered as a foreign device, uses Distribute-Broadcast-NPDU instead per Annex U. If a BBMD is attached, also forwards to BDT peers and registered foreign devices.

Parameters:: npdu (bytes) – NPDU bytes to broadcast.
Return type:: None

property local_address: BIP6Address¶: The local BACnet/IPv6 address of this transport.

property local_mac: bytes¶: The 3-byte VMAC address of this port.

property max_npdu_length: int¶: Maximum NPDU length for BACnet/IPv6 per Clause 6.

property vmac_cache: VMACCache¶: The VMAC resolution cache (for diagnostics).

property bbmd: BBMD6Manager | None¶: The attached BBMD6 manager, or None if not configured.

async attach_bbmd(bdt_entries=None)[source]¶

Attach an IPv6 BBMD manager to this transport.

Creates and starts a BBMD6Manager integrated with this transport. The BBMD intercepts incoming BVLC6 messages before they reach the normal receive path, and outgoing broadcasts are also forwarded to BDT peers and foreign devices.

Parameters:: bdt_entries (list[BDT6Entry] | None (default: None)) – Optional initial BDT entries.
Return type:: BBMD6Manager
Returns:: The attached BBMD6Manager instance.
Raises:: RuntimeError – If transport not started or BBMD already attached.

property foreign_device: ForeignDevice6Manager | None¶: The attached foreign device manager, or None.

async attach_foreign_device(bbmd_address, ttl)[source]¶

Attach an IPv6 foreign device manager to this transport.

Creates and starts a ForeignDevice6Manager that will register with the specified BBMD and periodically re-register.

Parameters:

bbmd_address (BIP6Address) – Address of the BBMD to register with.
ttl (int) – Time-to-Live for the registration in seconds.

Return type:

ForeignDevice6Manager

Returns:

The attached ForeignDevice6Manager instance.

Raises:

RuntimeError – If transport not started or FD already attached.

BACnet/IPv6 BBMD¶

BACnet/IPv6 Broadcast Management Device (BBMD) per Annex U.

Provides BBMD6Manager for BDT/FDT management, broadcast forwarding between BACnet/IPv6 subnets, and foreign device registration handling. IPv6 BBMDs use multicast instead of directed broadcast and all BVLL6 messages include a 3-byte source VMAC.

class bac_py.transport.bbmd6.BDT6Entry(address)[source]¶

Bases: object

Broadcast Distribution Table entry for BACnet/IPv6 per Annex U.

IPv6 BBMDs do not use broadcast masks (IPv6 uses multicast). Each entry contains only the B/IPv6 address of a peer BBMD.

Parameters:: address (BIP6Address)

address: BIP6Address¶

encode()[source]¶

Encode to 18-byte wire format.

Return type:: bytes

classmethod decode(data)[source]¶

Decode from 18-byte wire format.

Return type:: BDT6Entry
Parameters:: data (bytes | memoryview)

class bac_py.transport.bbmd6.FDT6Entry(address, vmac, ttl, expiry)[source]¶

Bases: object

Foreign Device Table entry for BACnet/IPv6 per Annex U.

Tracks a registered foreign device with its TTL and the absolute time at which the entry expires.

Parameters:

address (BIP6Address)
vmac (bytes)
ttl (int)
expiry (float)

address: BIP6Address¶

vmac: bytes¶

ttl: int¶

expiry: float¶

property remaining: int¶

Seconds remaining before this entry expires.

Capped at 65535 per wire encoding (2-octet field).

class bac_py.transport.bbmd6.BBMD6Manager(local_address, local_vmac, send_callback, local_broadcast_callback=None, multicast_send_callback=None, max_fdt_entries=128, accept_fd_registrations=True, fdt_cleanup_interval=10.0)[source]¶

Bases: object

BACnet/IPv6 Broadcast Management Device per Annex U.

Manages BDT (Broadcast Distribution Table) and FDT (Foreign Device Table), handles broadcast forwarding between BACnet/IPv6 subnets, and processes foreign device registration requests.

Key differences from IPv4 BBMD: - No broadcast mask – all BDT peer forwarding is unicast - All BVLL6 messages include source_vmac parameter - Forwarded-NPDU includes 18-byte originating_address - Uses multicast for local re-broadcast of Forwarded-NPDUs - FDT keyed on BIP6Address (18 bytes)

Parameters:

local_address (BIP6Address)
local_vmac (bytes)
send_callback (Callable[[bytes, BIP6Address], None])
local_broadcast_callback (Callable[[bytes, bytes], None] | None)
multicast_send_callback (Callable[[bytes], None] | None)
max_fdt_entries (int)
accept_fd_registrations (bool)
fdt_cleanup_interval (float)

property bdt: list[BDT6Entry]¶: Current Broadcast Distribution Table.

property fdt: dict[BIP6Address, FDT6Entry]¶: Current Foreign Device Table.

property accept_fd_registrations: bool¶: Whether foreign device registrations are accepted.

set_bdt(entries)[source]¶

Set the Broadcast Distribution Table.

Parameters:: entries (list[BDT6Entry]) – New BDT entries. Should include this BBMD’s own entry.
Return type:: None

async start()[source]¶

Start the FDT cleanup background task.

Return type:: None

async stop()[source]¶

Stop the FDT cleanup background task.

Return type:: None

handle_bvlc(function, data, source, source_vmac=None)[source]¶

Process a BVLC6 message directed at the BBMD.

Parameters:

function (Bvlc6Function) – BVLC6 function code.
data (bytes) – Payload after BVLL6 header fields.
source (BIP6Address) – UDP source address.
source_vmac (bytes | None (default: None)) – 3-byte VMAC of the sender.

Return type:

Returns:

True if the message was fully consumed by the BBMD and should not be delivered locally. False if the NPDU should also be processed by the normal receive path.

read_bdt()[source]¶

Return a copy of the current BDT.

Return type:: list[BDT6Entry]

read_fdt()[source]¶

Return a copy of the current FDT.

Return type:: dict[BIP6Address, FDT6Entry]

BACnet/IPv6 Foreign Device¶

Foreign device registration manager for BACnet/IPv6 per Annex U.

Provides ForeignDevice6Manager for registering with a remote IPv6 BBMD, periodic re-registration, and broadcast distribution via the BBMD.

class bac_py.transport.foreign_device6.ForeignDevice6Manager(bbmd_address, ttl, send_callback, local_vmac, local_address=None)[source]¶

Bases: object

Manages registration with a remote IPv6 BBMD per Annex U.

A BACnet/IPv6 device on a different subnet registers with a BBMD as a “foreign device” to receive broadcast traffic. This manager handles:

Initial registration with the BBMD
Periodic re-registration at TTL/2 intervals to prevent expiry
Tracking registration state (registered, failed)
Sending broadcasts via Distribute-Broadcast-NPDU

Usage:

fd_mgr = ForeignDevice6Manager(
    bbmd_address=BIP6Address("fd00::1", 47808),
    ttl=60,
    send_callback=transport_send,
    local_vmac=b"\x01\x02\x03",
)
await fd_mgr.start()
# ...later...
await fd_mgr.stop()

Parameters:

bbmd_address (BIP6Address)
ttl (int)
send_callback (Callable[[bytes, BIP6Address], None])
local_vmac (bytes)
local_address (BIP6Address | None)

property bbmd_address: BIP6Address¶: The BBMD address this device is registered with.

property ttl: int¶: The registration TTL in seconds.

property is_registered: bool¶: Whether the device is currently registered with the BBMD.

property last_result: Bvlc6ResultCode | None¶: The last BVLC6-Result code received from the BBMD.

async start()[source]¶

Start the registration loop.

Sends an initial registration and begins periodic re-registration at TTL/2 intervals.

Return type:: None

async stop()[source]¶

Stop the registration loop and deregister from the BBMD.

If the device is currently registered and a local address was provided, sends a Delete-Foreign-Device-Table-Entry to the BBMD so it can immediately remove the FDT entry.

Return type:: None

handle_bvlc_result(data)[source]¶

Process a BVLC6-Result received from the BBMD.

Parameters:: data (bytes) – 2-octet result code payload.
Return type:: None

send_distribute_broadcast(npdu)[source]¶

Send a broadcast via the BBMD using Distribute-Broadcast-NPDU.

This is used by foreign devices instead of multicast broadcast. The BBMD will distribute the NPDU to all BDT peers and other registered foreign devices.

Parameters:: npdu (bytes) – NPDU bytes to broadcast.
Raises:: RuntimeError – If not registered with a BBMD.
Return type:: None

BACnet Ethernet¶

BACnet Ethernet (ISO 8802-3) transport per ASHRAE 135-2020 Clause 7.

Provides raw 802.3 frame transport with IEEE 802.2 LLC headers for BACnet communication over Ethernet data links. Requires raw socket access (CAP_NET_RAW on Linux or /dev/bpf* on macOS).

Frame format (Clause 7):

+-----------+----------+--------+------+------+---------+------+
| Dst MAC   | Src MAC  | Length | DSAP | SSAP | Control | NPDU |
| (6 bytes) | (6 bytes)| (2)    | 0x82 | 0x82 | 0x03    | ...  |
+-----------+----------+--------+------+------+---------+------+

The 802.2 LLC header uses DSAP=0x82, SSAP=0x82, Control=0x03 (Unnumbered Information) as specified in Clause 7.

class bac_py.transport.ethernet.EthernetTransport(interface, *, mac_address=None)[source]¶

Bases: object

BACnet Ethernet (ISO 8802-3) transport per Clause 7.

Provides raw Ethernet frame I/O with 802.2 LLC headers for BACnet data-link communication. Satisfies the TransportPort protocol.

Platform support:

Linux: Uses AF_PACKET / SOCK_RAW sockets (requires CAP_NET_RAW).
macOS: Uses BPF devices (/dev/bpf*), requires root or appropriate permissions.
Windows: Not supported (raises NotImplementedError). Use Npcap or WinPcap for raw Ethernet access on Windows.

Parameters:

interface (str)
mac_address (bytes | None)

async start()[source]¶

Bind the raw socket and begin listening for BACnet frames.

Raises:

NotImplementedError – On unsupported platforms (Windows).
OSError – If socket creation fails.

Return type:

async stop()[source]¶

Stop listening and close the raw socket.

Return type:: None

on_receive(callback)[source]¶

Parameters:: callback (Callable[[bytes, bytes], None]) – Called with (npdu_bytes, source_mac) for each received BACnet Ethernet frame.
Return type:: None

send_unicast(npdu, mac_address)[source]¶

Send an NPDU to a specific station.

Parameters:

npdu (bytes) – NPDU bytes to send.
mac_address (bytes) – 6-byte destination Ethernet MAC address.

Raises:

ValueError – If npdu exceeds MAX_NPDU_LENGTH.

Return type:

send_broadcast(npdu)[source]¶

Send an NPDU as a local broadcast.

Parameters:: npdu (bytes) – NPDU bytes to broadcast.
Raises:: ValueError – If npdu exceeds MAX_NPDU_LENGTH.
Return type:: None

property local_mac: bytes¶: The 6-byte IEEE MAC address of this port.

property max_npdu_length: int¶: Maximum NPDU length for BACnet Ethernet: 1497 bytes per Clause 6.

Transport Port¶

Transport port abstraction for the network layer.

Defines the TransportPort protocol that all data-link transports (BACnet/IP, MS/TP, etc.) must satisfy so the network router can operate over heterogeneous data links without coupling to a specific technology.

class bac_py.transport.port.TransportPort(*args, **kwargs)[source]¶

Bases: Protocol

Abstract interface for a data-link transport port.

Each transport port represents a single attachment to one BACnet network. The network layer (and network router) interact with ports exclusively through this interface, using raw MAC-address bytes so that the same forwarding logic works regardless of the underlying data-link technology.

MAC encoding conventions (by data-link type):

BACnet/IP: 6 bytes (4-byte IPv4 + 2-byte port, big-endian)
BACnet/Ethernet: 6 bytes (IEEE 802 MAC address)
BACnet/IPv6: 3 bytes (VMAC virtual address)
MS/TP: 1 byte (station address 0-254)

async start()[source]¶

Bind the underlying transport and begin listening.

Return type:: None

async stop()[source]¶

Release resources and stop listening.

Return type:: None

on_receive(callback)[source]¶

Parameters:: callback (Callable[[bytes, bytes], None]) – Called with (npdu_bytes, source_mac) for each received datagram. source_mac is the raw MAC address of the sender in the encoding native to this data-link.
Return type:: None

send_unicast(npdu, mac_address)[source]¶

Send an NPDU to a specific station.

Parameters:

npdu (bytes) – Encoded NPDU bytes.
mac_address (bytes) – Destination MAC in this port’s native encoding.

Return type:

send_broadcast(npdu)[source]¶

Send an NPDU as a local broadcast.

Parameters:: npdu (bytes) – Encoded NPDU bytes.
Return type:: None

property local_mac: bytes¶: The MAC address of this port in its native encoding.

property max_npdu_length: int¶: Maximum NPDU length supported by this data-link per Clause 6.

BACnet Secure Connect¶

BACnet Secure Connect (BACnet/SC) transport per ASHRAE 135-2020 Annex AB.

Provides SCTransport which implements the TransportPort protocol, wiring together the hub connector, optional hub function, and optional direct-connection node switch.

class bac_py.transport.sc.SCTransport(config=None)[source]¶

Bases: object

BACnet/SC transport implementing the TransportPort protocol.

Wraps the hub connector (client to primary/failover hub), optional hub function (if this node is a hub), and optional node switch (for direct peer-to-peer connections).

Parameters:: config (SCTransportConfig | None)

property local_mac: bytes¶: 6-byte VMAC address as raw bytes.

property max_npdu_length: int¶: Maximum NPDU length for SC transport.

property hub_connector: SCHubConnector¶: The hub connector instance.

property hub_function: SCHubFunction | None¶: The hub function instance (if this node is a hub).

property node_switch: SCNodeSwitch | None¶: The node switch instance (if direct connections enabled).

on_receive(callback)[source]¶

Parameters:: callback (Callable[[bytes, bytes], None]) – Called with (npdu_bytes, source_mac) for each received NPDU. source_mac is the 6-byte VMAC of the sender.
Return type:: None

async start()[source]¶

Start the SC transport: hub function, hub connector, node switch.

Return type:: None

async stop()[source]¶

Stop the SC transport and release all resources.

Return type:: None

send_unicast(npdu, mac_address)[source]¶

Send an NPDU to a specific VMAC.

Tries direct connection first (if available), then hub. Uses a per-destination header cache to avoid SCVMAC/SCMessage creation and encoding overhead on the hot path.

Return type:

Parameters:

npdu (bytes)
mac_address (bytes)

send_broadcast(npdu)[source]¶

Send an NPDU as a broadcast via the hub.

Uses a pre-computed broadcast header to skip encoding overhead.

Return type:: None
Parameters:: npdu (bytes)

class bac_py.transport.sc.SCTransportConfig(primary_hub_uri='', failover_hub_uri=None, hub_function_config=None, node_switch_config=None, tls_config=<factory>, connection_config=<factory>, vmac=None, device_uuid=None, max_bvlc_length=1600, max_npdu_length=1497, min_reconnect_time=10.0, max_reconnect_time=600.0, network_number=None, connect_timeout=15.0)[source]¶

Bases: object

Configuration for a BACnet/SC transport.

Parameters:

primary_hub_uri (str)
failover_hub_uri (str | None)
hub_function_config (SCHubConfig | None)
node_switch_config (SCNodeSwitchConfig | None)
tls_config (SCTLSConfig)
connection_config (SCConnectionConfig)
vmac (SCVMAC | None)
device_uuid (DeviceUUID | None)
max_bvlc_length (int)
max_npdu_length (int)
min_reconnect_time (float)
max_reconnect_time (float)
network_number (int | None)
connect_timeout (float)

primary_hub_uri: str = ''¶: WebSocket URI of the primary hub (e.g. wss://hub.example.com:4443).

failover_hub_uri: str | None = None¶: WebSocket URI of the failover hub, if configured.

hub_function_config: SCHubConfig | None = None¶: If this node IS a hub, provide hub function configuration.

node_switch_config: SCNodeSwitchConfig | None = None¶: Optional direct peer-to-peer connection configuration.

tls_config: SCTLSConfig¶: TLS configuration for hub connections and direct connections.

connection_config: SCConnectionConfig¶: Timeouts and tuning for individual SC connections.

vmac: SCVMAC | None = None¶: Local VMAC address (auto-generated if None).

device_uuid: DeviceUUID | None = None¶: Device UUID (auto-generated if None).

max_bvlc_length: int = 1600¶: Maximum BVLC-SC message length.

max_npdu_length: int = 1497¶: Maximum NPDU length.

min_reconnect_time: float = 10.0¶: Minimum reconnect delay in seconds (AB.6.1: fixed range 10..30s).

max_reconnect_time: float = 600.0¶: Maximum reconnect delay in seconds.

network_number: int | None = None¶: Local network number for the SC network. Used by the network layer for NPDU routing. None if unknown.

connect_timeout: float = 15.0¶: Timeout in seconds for initial hub connection during start().

SC BVLC Codec¶

BVLC-SC message encoding and decoding per Annex AB.2.

Wire format (minimum 4 bytes):

Function(1) | Control(1) | MessageID(2) | [OrigVMAC(6)] | [DestVMAC(6)]
| [DestOptions(var)] | [DataOptions(var)] | [Payload(var)]

All multi-octet numeric values are big-endian (most significant octet first).

class bac_py.transport.sc.bvlc.SCHeaderOption(type, must_understand, data=b'')[source]¶

Bases: object

A single BVLC-SC header option (AB.2.3).

Header options appear in the Destination Options and Data Options fields of a BVLC-SC message. Each option has a type, a must-understand flag, and optional data.

Parameters:

type (int)
must_understand (bool)
data (bytes)

type: int¶

must_understand: bool¶

data: bytes¶

encode(*, more)[source]¶

Encode this header option to wire bytes.

Parameters:: more (bool) – True if more options follow in the list.
Return type:: bytes

static decode_list(data)[source]¶

Decode a list of chained header options.

Return type:: tuple[tuple[SCHeaderOption, ...], int]
Returns:: Tuple of (options, bytes_consumed).
Parameters:: data (memoryview)

class bac_py.transport.sc.bvlc.SCMessage(function, message_id, originating=None, destination=None, dest_options=(), data_options=(), payload=b'')[source]¶

Bases: object

A decoded BVLC-SC message (AB.2.1).

This is the generic envelope for all 13 BVLC-SC message types. The payload field contains the raw payload bytes; use the typed payload dataclasses (e.g. ConnectRequestPayload) to decode specific message payloads.

Parameters:

function (BvlcSCFunction)
message_id (int)
originating (SCVMAC | None)
destination (SCVMAC | None)
dest_options (tuple[SCHeaderOption, ...])
data_options (tuple[SCHeaderOption, ...])
payload (bytes)

function: BvlcSCFunction¶

message_id: int¶

originating: SCVMAC | None¶

destination: SCVMAC | None¶

dest_options: tuple[SCHeaderOption, ...]¶

data_options: tuple[SCHeaderOption, ...]¶

payload: bytes¶

encode()[source]¶

Encode this message to wire bytes.

Return type:: bytes

static decode(data, *, skip_payload=False)[source]¶

Decode a BVLC-SC message from wire bytes.

Parameters:

skip_payload (bool (default: False)) – If True, set payload to b"" instead of copying the remaining bytes. Used by the hub function which forwards raw bytes and never inspects the payload.
data (bytes | memoryview)

Raises:

ValueError – If the message is malformed or truncated.

Return type:

SCMessage

bac_py.transport.sc.bvlc.encode_encapsulated_npdu(originating, destination, payload)[source]¶

Fast-path encode for Encapsulated-NPDU (AB.2.12).

Avoids creating an SCMessage object and the generic encode() method overhead. Used by SCTransport.send_unicast() and SCTransport.send_broadcast() on the hot path.

Return type:

Parameters:

originating (SCVMAC)
destination (SCVMAC | None)
payload (bytes)

bac_py.transport.sc.bvlc.ConnectRequestPayload¶: alias of _ConnectPayload

bac_py.transport.sc.bvlc.ConnectAcceptPayload¶: alias of _ConnectPayload

class bac_py.transport.sc.bvlc.BvlcResultPayload(for_function, result_code, error_header_marker=0, error_class=0, error_code=0, error_details='')[source]¶

Bases: object

Payload for BVLC-Result messages (AB.2.4).

For ACK: only for_function and result_code are meaningful. For NAK: error_header_marker, error_class, error_code, and optionally error_details describe the error.

Parameters:

for_function (BvlcSCFunction)
result_code (SCResultCode)
error_header_marker (int)
error_class (int)
error_code (int)
error_details (str)

for_function: BvlcSCFunction¶

result_code: SCResultCode¶

error_header_marker: int¶

error_class: int¶

error_code: int¶

error_details: str¶

encode()[source]¶

Encode BVLC-Result payload.

Return type:: bytes

static decode(data)[source]¶

Decode BVLC-Result payload.

Return type:: BvlcResultPayload
Parameters:: data (bytes | memoryview)

class bac_py.transport.sc.bvlc.AdvertisementPayload(hub_connection_status, accept_direct_connections, max_bvlc_length, max_npdu_length)[source]¶

Bases: object

Payload for Advertisement messages (AB.2.8).

Parameters:

hub_connection_status (SCHubConnectionStatus)
accept_direct_connections (bool)
max_bvlc_length (int)
max_npdu_length (int)

hub_connection_status: SCHubConnectionStatus¶

accept_direct_connections: bool¶

max_bvlc_length: int¶

max_npdu_length: int¶

encode()[source]¶

Encode to 6 bytes.

Return type:: bytes

static decode(data)[source]¶

Decode from payload bytes.

Return type:: AdvertisementPayload
Parameters:: data (bytes | memoryview)

class bac_py.transport.sc.bvlc.AddressResolutionAckPayload(websocket_uris)[source]¶

Bases: object

Payload for Address-Resolution-ACK messages (AB.2.7).

WebSocket URIs are space-separated in the wire format.

Parameters:: websocket_uris (tuple[str, ...])

websocket_uris: tuple[str, ...]¶

encode()[source]¶

Encode URI list to UTF-8 payload.

Return type:: bytes

static decode(data)[source]¶

Decode URI list from payload bytes (max 16 URIs).

Return type:: AddressResolutionAckPayload
Parameters:: data (bytes | memoryview)

class bac_py.transport.sc.bvlc.ProprietaryMessagePayload(vendor_id, proprietary_function, proprietary_data=b'')[source]¶

Bases: object

Payload for Proprietary-Message (AB.2.16).

Parameters:

vendor_id (int)
proprietary_function (int)
proprietary_data (bytes)

vendor_id: int¶

proprietary_function: int¶

proprietary_data: bytes¶

encode()[source]¶

Encode proprietary payload.

Return type:: bytes

static decode(data)[source]¶

Decode proprietary payload.

Return type:: ProprietaryMessagePayload
Parameters:: data (bytes | memoryview)

bac_py.transport.sc.bvlc.build_secure_path_option()[source]¶

Build a Secure Path data option (AB.2.3.1).

The Secure Path header option has Must-Understand=1 and no data.

Return type:: SCHeaderOption

SC VMAC Addressing¶

VMAC addressing and Device UUID for BACnet/SC (AB.1.5).

class bac_py.transport.sc.vmac.SCVMAC(address)[source]¶

Bases: object

6-byte virtual MAC address for BACnet/SC nodes (AB.1.5.2).

VMAC addresses use the EUI-48 format. The broadcast VMAC is FF:FF:FF:FF:FF:FF. The all-zeros address 00:00:00:00:00:00 is reserved to indicate an unknown or uninitialized VMAC.

Parameters:: address (bytes)

address: bytes¶

classmethod random()[source]¶

Generate a random locally-administered unicast VMAC.

Sets the locally-administered bit (bit 1 of first octet) and clears the multicast bit (bit 0 of first octet) per IEEE 802.

Return type:: SCVMAC

classmethod broadcast()[source]¶

Return the local broadcast VMAC (FF:FF:FF:FF:FF:FF).

Return type:: SCVMAC

classmethod from_hex(hex_str)[source]¶

Parse a VMAC from hex string (with or without colons/hyphens).

Accepts formats: "AABBCCDDEEFF", "AA:BB:CC:DD:EE:FF", "AA-BB-CC-DD-EE-FF".

Return type:: SCVMAC
Parameters:: hex_str (str)

property is_broadcast: bool¶: Return True if this is the broadcast VMAC.

property is_uninitialized: bool¶: Return True if this is the all-zeros (uninitialized) VMAC.

class bac_py.transport.sc.vmac.DeviceUUID(value)[source]¶

Bases: object

16-byte device UUID for BACnet/SC (AB.1.5.3).

Every BACnet/SC device has a UUID (RFC 4122) that persists across restarts and is independent of VMAC or device instance.

Parameters:: value (bytes)

value: bytes¶

classmethod generate()[source]¶

Generate a new random UUID (version 4).

Return type:: DeviceUUID

classmethod from_hex(hex_str)[source]¶

Parse UUID from hex string (with or without hyphens).

Accepts "550e8400e29b41d4a716446655440000" or standard "550e8400-e29b-41d4-a716-446655440000" format.

Return type:: DeviceUUID
Parameters:: hex_str (str)

SC Connection¶

BACnet/SC connection state machine (AB.6.2).

Implements both the initiating peer (Figure AB-11) and accepting peer (Figure AB-12) state machines for BACnet/SC connections.

class bac_py.transport.sc.connection.SCConnectionState(*values)[source]¶

Bases: IntEnum

Connection state machine states.

IDLE = 0¶

AWAITING_WEBSOCKET = 1¶

AWAITING_ACCEPT = 2¶

AWAITING_REQUEST = 3¶

CONNECTED = 4¶

DISCONNECTING = 5¶

class bac_py.transport.sc.connection.SCConnectionRole(*values)[source]¶

Bases: IntEnum

Whether this side initiated or accepted the connection.

INITIATING = 0¶

ACCEPTING = 1¶

class bac_py.transport.sc.connection.SCConnectionConfig(connect_wait_timeout=10.0, disconnect_wait_timeout=5.0, heartbeat_timeout=300.0)[source]¶

Bases: object

Timeouts and tuning for an SC connection.

Parameters:

connect_wait_timeout (float)
disconnect_wait_timeout (float)
heartbeat_timeout (float)

connect_wait_timeout: float = 10.0¶

disconnect_wait_timeout: float = 5.0¶

heartbeat_timeout: float = 300.0¶

class bac_py.transport.sc.connection.SCConnection(local_vmac, local_uuid, config=None, max_bvlc_length=1600, max_npdu_length=1497, *, hub_mode=False)[source]¶

Bases: object

BACnet/SC connection state machine (AB.6.2).

Manages the lifecycle of a single WebSocket connection to a hub or direct peer, including handshake, heartbeat, and graceful disconnect.

Parameters:

local_vmac (SCVMAC)
local_uuid (DeviceUUID)
config (SCConnectionConfig | None)
max_bvlc_length (int)
max_npdu_length (int)
hub_mode (bool)

peer_vmac: SCVMAC | None¶

peer_uuid: DeviceUUID | None¶

peer_max_bvlc: int¶

peer_max_npdu: int¶

on_connected: Callable[[], None] | None¶

on_disconnected: Callable[[], None] | None¶

on_message: Callable[[SCMessage, bytes | None], Awaitable[None] | None] | None¶

on_vmac_collision: Callable[[], None] | None¶

property state: SCConnectionState¶: Current connection state.

property role: SCConnectionRole | None¶: Connection role (initiating or accepting).

property local_vmac: SCVMAC¶: Local VMAC address.

async initiate(ws)[source]¶

Run the initiating peer state machine on an established WebSocket.

Transitions: IDLE → AWAITING_ACCEPT → CONNECTED (or IDLE on failure).

Return type:: None
Parameters:: ws (SCWebSocket)

async accept(ws, vmac_checker=None)[source]¶

Run the accepting peer state machine on an established WebSocket.

Parameters:

ws (SCWebSocket) – The WebSocket connection (already upgraded).
vmac_checker (Callable[[SCVMAC, DeviceUUID], bool] | None (default: None)) – Optional callback (vmac, uuid) -> ok. Returns False if the VMAC collides with an existing connection.

Return type:

Transitions: IDLE → AWAITING_REQUEST → CONNECTED (or IDLE on failure).

async send_message(msg)[source]¶

Send a BVLC-SC message on this connection.

Return type:: None
Parameters:: msg (SCMessage)

async send_raw(data)[source]¶

Send pre-encoded BVLC-SC bytes, skipping encode().

Used by the hub to forward messages without re-encoding.

Return type:: None
Parameters:: data (bytes)

write_raw_no_drain(data)[source]¶

Buffer pre-encoded bytes without draining.

Returns True if data was buffered. Call drain() afterwards. Used by hub broadcast to batch writes before draining concurrently.

Return type:: bool
Parameters:: data (bytes)

async drain()[source]¶

Drain the write buffer. Pair with write_raw_no_drain().

Return type:: None

async disconnect()[source]¶

Initiate graceful disconnect.

Return type:: None

SC Hub Function¶

BACnet/SC Hub Function (AB.5.3).

The Hub Function accepts BACnet/SC connections as hub connections and routes unicast/broadcast BVLC-SC messages between connected nodes.

class bac_py.transport.sc.hub_function.SCHubConfig(bind_address='0.0.0.0', bind_port=4443, tls_config=<factory>, connection_config=<factory>, max_connections=1000, max_bvlc_length=6000, max_npdu_length=1497)[source]¶

Bases: object

Configuration for an SC Hub Function.

Parameters:

bind_address (str)
bind_port (int)
tls_config (SCTLSConfig)
connection_config (SCConnectionConfig)
max_connections (int)
max_bvlc_length (int)
max_npdu_length (int)

bind_address: str = '0.0.0.0'¶

bind_port: int = 4443¶

tls_config: SCTLSConfig¶

connection_config: SCConnectionConfig¶

max_connections: int = 1000¶

max_bvlc_length: int = 6000¶

max_npdu_length: int = 1497¶

class bac_py.transport.sc.hub_function.SCHubFunction(hub_vmac, hub_uuid, config=None)[source]¶

Bases: object

BACnet/SC Hub Function (AB.5.3).

WebSocket server that accepts hub connections from SC nodes. Routes unicast messages to destination VMAC, replicates broadcasts to all connected nodes except the source.

Parameters:

hub_vmac (SCVMAC)
hub_uuid (DeviceUUID)
config (SCHubConfig | None)

property connections: dict[SCVMAC, SCConnection]¶: Active connections indexed by peer VMAC.

property connection_count: int¶: Number of active hub connections.

async start()[source]¶

Start the hub function WebSocket server.

Return type:: None

async stop()[source]¶

Stop the hub function, force-close all connections.

Return type:: None

SC Hub Connector¶

BACnet/SC Hub Connector (AB.5.2).

Maintains a persistent connection to a primary hub with automatic reconnection and failover to a secondary hub when configured.

class bac_py.transport.sc.hub_connector.SCHubConnectorConfig(primary_hub_uri='', failover_hub_uri=None, tls_config=<factory>, connection_config=<factory>, min_reconnect_time=10.0, max_reconnect_time=600.0, max_bvlc_length=1600, max_npdu_length=1497)[source]¶

Bases: object

Configuration for an SC Hub Connector.

Parameters:

primary_hub_uri (str)
failover_hub_uri (str | None)
tls_config (SCTLSConfig)
connection_config (SCConnectionConfig)
min_reconnect_time (float)
max_reconnect_time (float)
max_bvlc_length (int)
max_npdu_length (int)

primary_hub_uri: str = ''¶

failover_hub_uri: str | None = None¶

tls_config: SCTLSConfig¶

connection_config: SCConnectionConfig¶

min_reconnect_time: float = 10.0¶

max_reconnect_time: float = 600.0¶

max_bvlc_length: int = 1600¶

max_npdu_length: int = 1497¶

class bac_py.transport.sc.hub_connector.SCHubConnector(local_vmac, local_uuid, config=None)[source]¶

Bases: object

BACnet/SC Hub Connector (AB.5.2).

Maintains a persistent connection to the primary hub with automatic reconnection and failover to a secondary hub when configured.

Parameters:

local_vmac (SCVMAC)
local_uuid (DeviceUUID)
config (SCHubConnectorConfig | None)

on_message: Callable[[SCMessage, bytes | None], Awaitable[None] | None] | None¶

on_status_change: Callable[[SCHubConnectionStatus], None] | None¶

property is_connected: bool¶: Whether the connector has an active hub connection.

property connection_status: SCHubConnectionStatus¶: Current hub connection status.

property local_vmac: SCVMAC¶: Local VMAC address.

async start()[source]¶

Start the hub connector, begin connection attempts.

Return type:: None

async stop()[source]¶

Disconnect from hub and stop reconnection loop.

Return type:: None

async send(msg)[source]¶

Send a message to the hub.

Raises:: ConnectionError – If not connected.
Return type:: None
Parameters:: msg (SCMessage)

async send_raw(data)[source]¶

Send pre-encoded bytes to the hub.

Raises:: ConnectionError – If not connected.
Return type:: None
Parameters:: data (bytes)

async wait_connected(timeout=None)[source]¶

Wait until the connector is connected to a hub.

Return type:: bool
Returns:: True if connected, False if timeout expired.
Parameters:: timeout (float | None)

SC Node Switch¶

BACnet/SC Node Switch (AB.4).

Manages direct peer-to-peer connections between SC nodes, bypassing the hub for unicast traffic. Listens for inbound direct connections and initiates outbound connections via address resolution through the hub.

class bac_py.transport.sc.node_switch.SCNodeSwitchConfig(enable=False, bind_address='0.0.0.0', bind_port=0, tls_config=<factory>, connection_config=<factory>, address_resolution_timeout=5.0, max_connections=100, max_bvlc_length=1600, max_npdu_length=1497)[source]¶

Bases: object

Configuration for an SC Node Switch.

Parameters:

enable (bool)
bind_address (str)
bind_port (int)
tls_config (SCTLSConfig)
connection_config (SCConnectionConfig)
address_resolution_timeout (float)
max_connections (int)
max_bvlc_length (int)
max_npdu_length (int)

enable: bool = False¶

bind_address: str = '0.0.0.0'¶

bind_port: int = 0¶

tls_config: SCTLSConfig¶

connection_config: SCConnectionConfig¶

address_resolution_timeout: float = 5.0¶

max_connections: int = 100¶

max_bvlc_length: int = 1600¶

max_npdu_length: int = 1497¶

class bac_py.transport.sc.node_switch.SCNodeSwitch(local_vmac, local_uuid, config=None)[source]¶

Bases: object

BACnet/SC Node Switch (AB.4).

Manages direct connections between SC nodes. Listens for inbound direct connections and initiates outbound connections via address resolution through the hub.

Parameters:

local_vmac (SCVMAC)
local_uuid (DeviceUUID)
config (SCNodeSwitchConfig | None)

on_message: Callable[[SCMessage, bytes | None], Awaitable[None] | None] | None¶

property connections: dict[SCVMAC, SCConnection]¶: Active direct connections indexed by peer VMAC.

property connection_count: int¶: Number of active direct connections.

property local_vmac: SCVMAC¶: Local VMAC address.

async start()[source]¶

Start the node switch, listening for inbound direct connections.

Return type:: None

async stop()[source]¶

Stop the node switch and close all direct connections.

Return type:: None

has_direct(dest)[source]¶

Check if a direct connection exists to the given VMAC.

Return type:: bool
Parameters:: dest (SCVMAC)

async send_direct(dest, msg)[source]¶

Send via direct connection if available.

Return type:

Returns:

True if sent successfully, False if no direct connection.

Parameters:

dest (SCVMAC)
msg (SCMessage)

async resolve_address(dest, hub_send)[source]¶

Request WebSocket URIs for a peer via Address-Resolution through the hub.

Parameters:

dest (SCVMAC) – Target VMAC to resolve.
hub_send (Callable[[SCMessage], Awaitable[None]]) – Callback to send a message through the hub connection.

Return type:

list[str]

Returns:

List of WebSocket URIs for the target, empty if not resolved.

handle_address_resolution_ack(msg)[source]¶

Handle an Address-Resolution-ACK received from the hub.

Resolves the pending future for the originating VMAC.

Return type:: None
Parameters:: msg (SCMessage)

async establish_direct(dest, uris)[source]¶

Try connecting to a peer using resolved URIs.

Return type:

Returns:

True if connected successfully.

Parameters:

dest (SCVMAC)
uris (list[str])

SC WebSocket¶

Sans-I/O WebSocket wrapper for BACnet/SC (AB.7).

Uses the websockets library’s sans-I/O protocol objects together with asyncio TCP/TLS streams. Each SCWebSocket instance owns one WebSocket connection backed by a (StreamReader, StreamWriter) pair.

class bac_py.transport.sc.websocket.SCWebSocket(reader, writer, protocol, *, max_frame_size=0)[source]¶

Bases: object

Async WebSocket connection using the websockets sans-I/O protocol.

Each instance represents one open WebSocket connection and is backed by asyncio StreamReader/StreamWriter objects. All framing is handled by the websockets sans-I/O ClientProtocol or ServerProtocol.

Parameters:

reader (StreamReader)
writer (StreamWriter)
protocol (ClientProtocol | ServerProtocol)
max_frame_size (int)

async classmethod connect(uri, ssl_ctx, subprotocol, *, handshake_timeout=10.0, max_size=None)[source]¶

Initiate a WebSocket client connection.

Parameters:

uri (str) – WebSocket URI (wss://host:port/path).
ssl_ctx (SSLContext | None) – TLS context, or None for plaintext ws://.
subprotocol (str) – WebSocket subprotocol to negotiate.
handshake_timeout (float (default: 10.0)) – Maximum seconds for the WebSocket handshake.
max_size (int | None (default: None)) – Maximum WebSocket message size. Passed to the protocol layer so oversized frames are rejected early.

Return type:

SCWebSocket

async classmethod accept(reader, writer, subprotocol, *, handshake_timeout=10.0, max_size=None)[source]¶

Accept an inbound WebSocket connection on existing streams.

Parameters:

reader (StreamReader) – asyncio StreamReader from accepted connection.
writer (StreamWriter) – asyncio StreamWriter from accepted connection.
subprotocol (str) – WebSocket subprotocol to accept.
handshake_timeout (float (default: 10.0)) – Maximum seconds to wait for the handshake.
max_size (int | None (default: None)) – Maximum WebSocket message size. Passed to the protocol layer so oversized frames are rejected early.

Return type:

SCWebSocket

async send(data)[source]¶

Send a binary WebSocket frame.

Return type:: None
Parameters:: data (bytes)

async send_raw(data)¶

Send a binary WebSocket frame.

Return type:: None
Parameters:: data (bytes)

write_no_drain(data)[source]¶

Buffer a binary WebSocket frame without draining.

Returns True if data was written to the transport buffer. Call drain() afterwards to flush. Used by hub broadcast to batch writes across connections before draining concurrently.

Return type:: bool
Parameters:: data (bytes)

async drain()[source]¶

Drain the write buffer. Pair with write_no_drain().

Return type:: None

async recv()[source]¶

Receive the next binary WebSocket message.

Raises:

ConnectionClosedOK – On graceful close.
ConnectionClosedError – On abnormal close.

Return type:

async close(code=1000, reason='')[source]¶

Initiate graceful WebSocket close.

Return type: