Server Mode¶
bac-py can act as a full BACnet server, exposing local objects, responding to client requests, and running engines for scheduling, trend logging, events, and auditing. This guide covers everything from basic setup to advanced customization.
Basic Server Setup¶
Every bac-py server follows the same four-step pattern regardless of transport:
Create a
DeviceConfig(transport selection happens here)Start a
BACnetApplicationPopulate the
ObjectDatabasewith aDeviceObjectand application objectsRegister
DefaultServerHandlersfor APDU dispatch
The transport is determined entirely by DeviceConfig — the object
creation, handler registration, and lifecycle management are identical
across BACnet/IP, IPv6, SC, and Ethernet.
BACnet/IP server (default)¶
The default transport. Uses UDP on port 47808 with broadcast discovery:
import asyncio
from bac_py import BACnetApplication, DefaultServerHandlers, DeviceConfig, DeviceObject
from bac_py.objects.analog import AnalogInputObject
from bac_py.types.enums import EngineeringUnits
async def serve():
config = DeviceConfig(
instance_number=100,
name="My-Device",
vendor_name="ACME",
vendor_id=999,
interface="0.0.0.0", # bind address (all interfaces)
port=0xBAC0, # UDP 47808
broadcast_address="255.255.255.255", # or subnet-directed, e.g. "192.168.1.255"
)
async with BACnetApplication(config) as app:
device = DeviceObject(
instance_number=100,
object_name="My-Device",
vendor_name="ACME",
vendor_identifier=999,
)
app.object_db.add(device)
app.object_db.add(AnalogInputObject(
instance_number=1,
object_name="Temperature",
units=EngineeringUnits.DEGREES_CELSIUS,
present_value=22.5,
))
handlers = DefaultServerHandlers(app, app.object_db, device)
handlers.register()
# Server now responds to Who-Is, ReadProperty,
# ReadPropertyMultiple, WriteProperty, COV subscriptions,
# and other standard services.
await app.run()
asyncio.run(serve())
BACnet/IPv6 server¶
BACnet/IPv6 (Annex U) uses UDP over IPv6 with multicast discovery and 3-byte
VMAC addressing. Set ipv6=True on DeviceConfig — everything else
is identical to BACnet/IP:
async def serve_ipv6():
config = DeviceConfig(
instance_number=100,
name="My-IPv6-Device",
vendor_name="ACME",
vendor_id=999,
ipv6=True,
# interface="::" is the default when ipv6=True (all interfaces)
# port=0xBAC0 is the default (47808)
# multicast_address="ff02::bac0" is the default (link-local)
# vmac=None auto-generates a 3-byte VMAC
)
async with BACnetApplication(config) as app:
device = DeviceObject(
instance_number=100,
object_name="My-IPv6-Device",
vendor_name="ACME",
vendor_identifier=999,
)
app.object_db.add(device)
# ... add objects ...
handlers = DefaultServerHandlers(app, app.object_db, device)
handlers.register()
await app.run()
asyncio.run(serve_ipv6())
Use multicast_address="ff05::bac0" for site-local scope (reaches beyond
the link). Provide an explicit vmac (3 bytes) if you need a stable
address across restarts.
See examples/ipv6_server.py for a complete example.
BACnet/SC server¶
BACnet Secure Connect (Annex AB) uses TLS-secured WebSockets in a
hub-and-spoke topology. An SC server typically runs the hub function so
that SC nodes can connect to it. Pass an SCTransportConfig with a
hub_function_config to DeviceConfig:
from bac_py.transport.sc import SCTransportConfig
from bac_py.transport.sc.hub_function import SCHubConfig
from bac_py.transport.sc.tls import SCTLSConfig
async def serve_sc():
tls = SCTLSConfig(
ca_certificates_path="/path/to/ca.pem",
certificate_path="/path/to/hub.pem",
private_key_path="/path/to/hub.key",
)
config = DeviceConfig(
instance_number=100,
name="My-SC-Hub",
vendor_name="ACME",
vendor_id=999,
sc_config=SCTransportConfig(
hub_function_config=SCHubConfig(
bind_address="0.0.0.0",
bind_port=8443,
tls_config=tls,
),
tls_config=tls,
),
)
async with BACnetApplication(config) as app:
device = DeviceObject(
instance_number=100,
object_name="My-SC-Hub",
vendor_name="ACME",
vendor_identifier=999,
)
app.object_db.add(device)
# ... add objects ...
handlers = DefaultServerHandlers(app, app.object_db, device)
handlers.register()
await app.run()
asyncio.run(serve_sc())
Key SCTransportConfig fields for server mode:
hub_function_config— required for a server that accepts node connections. Omit this (and setprimary_hub_uri) for a node that connects to an existing hub.tls_config— TLS 1.3 certificate configuration. UseSCTLSConfig(allow_plaintext=True)only for testing.primary_hub_uri— leave empty ("") when this node is the hub. Set it to connect to another hub as a client simultaneously (hub chaining).
For testing without TLS certificates:
config = DeviceConfig(
instance_number=100,
name="SC-Test-Hub",
sc_config=SCTransportConfig(
hub_function_config=SCHubConfig(
bind_address="0.0.0.0",
bind_port=4443,
tls_config=SCTLSConfig(allow_plaintext=True),
),
tls_config=SCTLSConfig(allow_plaintext=True),
),
)
See examples/sc_server.py for a complete example and
BACnet Secure Connect for TLS certificate generation, failover, and VMAC
addressing.
BACnet Ethernet server¶
BACnet Ethernet (Clause 7) sends BACnet packets directly over IEEE 802.3
frames with 802.2 LLC headers, bypassing IP entirely. Set
ethernet_interface on DeviceConfig:
async def serve_ethernet():
config = DeviceConfig(
instance_number=100,
name="My-Ethernet-Device",
vendor_name="ACME",
vendor_id=999,
ethernet_interface="eth0",
# ethernet_mac=b"\x02\x42\xAC\x11\x00\x02", # required on macOS
)
async with BACnetApplication(config) as app:
device = DeviceObject(
instance_number=100,
object_name="My-Ethernet-Device",
vendor_name="ACME",
vendor_identifier=999,
)
app.object_db.add(device)
# ... add objects ...
handlers = DefaultServerHandlers(app, app.object_db, device)
handlers.register()
await app.run()
asyncio.run(serve_ethernet())
Platform requirements:
Linux — raw sockets require
CAP_NET_RAWor root. The MAC address is auto-detected from the interface.macOS — requires a BPF device (
/dev/bpf*), typically root. You must provide an explicitethernet_macbecause BPF does not expose the interface MAC address directly.
See examples/ethernet_server.py for a complete example.
Transport-independent wiring¶
Because transport selection is isolated in DeviceConfig, the rest of
your server code — object creation, handler registration, engines, custom
handlers — is identical across all four transports. You can switch transports
by changing only the DeviceConfig, which makes it straightforward to
support multiple deployment environments from a single codebase.
DefaultServerHandlers.register() installs handlers for all standard
BACnet server services. The server will respond to Who-Is with I-Am,
ReadProperty and ReadPropertyMultiple with values from the object database,
and WriteProperty/WritePropertyMultiple to update writable objects. See
Object Model for the full list of supported object types.
DeviceConfig Options¶
DeviceConfig controls device identity,
network parameters, and security:
from bac_py import DeviceConfig
from bac_py.app.application import RouterConfig, RouterPortConfig
config = DeviceConfig(
instance_number=100, # BACnet device instance (0-4194302)
name="My-Device", # Device name
vendor_name="ACME", # Vendor string
vendor_id=999, # ASHRAE vendor ID
model_name="Controller-1", # Model name
firmware_revision="2.0.0", # Firmware version (default: bac-py version)
application_software_version="1.0.0", # Software version (default: bac-py version)
interface="0.0.0.0", # IP address to bind
port=0xBAC0, # UDP port (47808)
max_apdu_length=1476, # Max APDU size (bytes)
max_segments=None, # Max segments (None = unlimited)
apdu_timeout=6000, # Request timeout (ms)
apdu_segment_timeout=2000, # Segment timeout (ms)
apdu_retries=3, # Retry count
broadcast_address="255.255.255.255", # Directed broadcast address
password="secret123", # Optional password for DCC/ReinitializeDevice
router_config=None, # Multi-network router (see below)
ipv6=False, # Use BACnet/IPv6 (Annex U) transport
sc_config=None, # BACnet/SC transport config (Annex AB)
ethernet_interface=None, # Ethernet interface, e.g. "eth0" (Clause 7)
ethernet_mac=None, # Explicit 6-byte MAC (auto-detected on Linux)
)
Transport selection – ipv6, sc_config, and ethernet_interface
are mutually exclusive. When none are set, the default BACnet/IP (UDP)
transport is used. See Transport Setup for transport configuration
details and examples for each transport type.
The password field (1–20 characters) is used by the
DeviceCommunicationControl and ReinitializeDevice handlers. When set, incoming
requests must include a matching password or the server responds with a
PASSWORD_FAILURE error. The comparison uses hmac.compare_digest() for
constant-time security.
The broadcast_address defaults to "255.255.255.255" (global broadcast).
Override it for subnet-directed broadcasts in Docker or segmented networks
(e.g. "192.168.1.255").
See Multi-Network Routing for router_config details.
Object Database¶
The ObjectDatabase is the runtime registry for
all BACnet objects hosted by the server.
Adding and removing objects¶
from bac_py.objects.analog import AnalogInputObject, AnalogOutputObject
from bac_py.objects.binary import BinaryInputObject
from bac_py.types.enums import EngineeringUnits, ObjectType
from bac_py.types.primitives import ObjectIdentifier
# Add objects
ai = AnalogInputObject(
instance_number=1,
object_name="Zone-Temp",
units=EngineeringUnits.DEGREES_CELSIUS,
present_value=22.5,
)
app.object_db.add(ai)
# Object names must be unique (Clause 12.1.5)
# Duplicate names or IDs raise BACnetError
# Remove an object
app.object_db.remove(ObjectIdentifier(ObjectType.ANALOG_INPUT, 1))
# Note: Device objects cannot be removed
Querying objects¶
# Look up by identifier
obj = app.object_db.get(ObjectIdentifier(ObjectType.ANALOG_INPUT, 1))
# All objects of a type
all_ai = app.object_db.get_objects_of_type(ObjectType.ANALOG_INPUT)
# Full object list (auto-computed)
obj_list = app.object_db.object_list
# Iterate all objects
for obj in app.object_db:
print(obj.object_identifier, obj.read_property(PropertyIdentifier.OBJECT_NAME))
# Count and membership
count = len(app.object_db)
exists = ObjectIdentifier(ObjectType.ANALOG_INPUT, 1) in app.object_db
The Object_List property on the Device object is automatically computed
from the database contents. Adding or removing objects increments the
Database_Revision property.
Change callbacks¶
Register callbacks to be notified when a property value is written:
from bac_py.types.enums import PropertyIdentifier
def on_temp_change(prop_id, old_value, new_value):
print(f"Temperature changed: {old_value} -> {new_value}")
app.object_db.register_change_callback(
ObjectIdentifier(ObjectType.ANALOG_INPUT, 1),
PropertyIdentifier.PRESENT_VALUE,
on_temp_change,
)
# Later, to stop receiving callbacks:
app.object_db.unregister_change_callback(
ObjectIdentifier(ObjectType.ANALOG_INPUT, 1),
PropertyIdentifier.PRESENT_VALUE,
on_temp_change,
)
Change callbacks power both COV-based trend logging and the event engine’s intrinsic reporting.
Supported Object Types¶
bac-py includes 62 object types covering the full BACnet standard:
Sensing: AnalogInput, BinaryInput, MultiStateInput
Control (commandable): AnalogOutput, BinaryOutput, MultiStateOutput
Values: AnalogValue, BinaryValue, MultiStateValue
Extended values: IntegerValue, PositiveIntegerValue, LargeAnalogValue, CharacterStringValue, OctetStringValue, BitStringValue, DateValue, TimeValue, DateTimeValue, DatePatternValue, TimePatternValue, DateTimePatternValue
Infrastructure: Device, File, NetworkPort, Channel
Scheduling: Schedule, Calendar
Trending: TrendLog, TrendLogMultiple
Events: EventEnrollment, NotificationClass, EventLog, AlertEnrollment, NotificationForwarder
Safety: LifeSafetyPoint, LifeSafetyZone
Auditing: AuditReporter, AuditLog
Access control: AccessDoor, AccessPoint, AccessZone, AccessUser, AccessRights, AccessCredential, CredentialDataInput
Advanced control: Command, Timer, Staging, LoadControl, Loop, PulseConverter, Accumulator
Lighting: LightingOutput, BinaryLightingOutput
Transportation: ElevatorGroup, Lift, Escalator
Other: Program, Averaging, Group, GlobalGroup, StructuredView
All objects are created as frozen dataclasses with validated property definitions and read/write access control.
Commandable Objects and Priority Arrays¶
BACnet commandable objects support a 16-level command priority array (Clause 19.2). When multiple sources write to the same object, the highest priority (lowest number) wins.
Always-commandable objects¶
AnalogOutput, BinaryOutput, and MultiStateOutput are always commandable:
from bac_py.objects.analog import AnalogOutputObject
from bac_py.types.enums import EngineeringUnits
ao = AnalogOutputObject(
instance_number=1,
object_name="Damper-Position",
units=EngineeringUnits.PERCENT,
relinquish_default=0.0,
)
app.object_db.add(ao)
# Write at priority 8 (manual operator)
ao.write_property(PropertyIdentifier.PRESENT_VALUE, 75.0, priority=8)
# The present value is now 75.0 (priority 8 is the highest active slot)
# Write at priority 1 (life safety -- overrides priority 8)
ao.write_property(PropertyIdentifier.PRESENT_VALUE, 100.0, priority=1)
# Present value is now 100.0
# Relinquish priority 1
ao.write_property(PropertyIdentifier.PRESENT_VALUE, None, priority=1)
# Present value falls back to 75.0 (priority 8)
# Relinquish priority 8
ao.write_property(PropertyIdentifier.PRESENT_VALUE, None, priority=8)
# All slots empty -- present value falls back to relinquish_default (0.0)
Optionally-commandable objects¶
IntegerValue and PositiveIntegerValue support an optional commandable
flag:
from bac_py.objects.value_types import IntegerValueObject
iv = IntegerValueObject(
instance_number=1,
object_name="Setpoint-Mode",
commandable=True,
relinquish_default=0,
)
When commandable=True, the object gets a full 16-level priority array,
Relinquish_Default, Current_Command_Priority, and value source tracking.
When commandable=False (the default), writes go directly to
Present_Value without priority handling.
COV Subscriptions (Server Side)¶
The server’s COV manager (COVManager) handles
incoming SubscribeCOV, SubscribeCOVProperty, and SubscribeCOVPropertyMultiple
requests automatically when DefaultServerHandlers is registered.
How it works¶
A remote client sends a SubscribeCOV request for an object
The server registers the subscription with a lifetime timer
An initial notification is sent immediately (Clause 13.1.2)
On each property write, the server checks if the value change exceeds the COV threshold and sends notifications to all matching subscribers
Notification thresholds:
Analog objects: Notify when
|new - last| >= COV_Increment, or on any change if no increment is setBinary/multistate objects: Notify on any change in
Present_ValueAll objects: Notify on any change in
Status_Flags
Subscription types:
Object-level (SubscribeCOV): Monitors
Present_ValueandStatus_FlagsProperty-level (SubscribeCOVProperty): Monitors a specific property with optional per-subscription COV increment override
Property-multiple (SubscribeCOVPropertyMultiple): Multiple property-level subscriptions in a single request
Lifetime management:
Subscriptions with a
lifetime(in seconds) expire automaticallySubscriptions without a lifetime persist indefinitely
Clients can cancel subscriptions or re-subscribe to refresh the lifetime
The COV manager is created automatically during application startup and
shut down during application stop. No additional configuration is needed
beyond registering DefaultServerHandlers.
Inspecting active subscriptions¶
# List all active subscriptions
subs = app.cov_manager.get_active_subscriptions()
# Filter by object
subs = app.cov_manager.get_active_subscriptions(
ObjectIdentifier(ObjectType.ANALOG_INPUT, 1)
)
for sub in subs:
print(f"Subscriber: {sub.subscriber}, Object: {sub.monitored_object}, "
f"Confirmed: {sub.confirmed}, Lifetime: {sub.lifetime}")
Custom Service Handlers¶
The ServiceRegistry dispatches incoming
requests to registered handler functions. DefaultServerHandlers registers
handlers for all standard services, but you can replace or extend them.
Handler signature¶
Confirmed service handlers receive the raw request bytes and return
response bytes (for ComplexACK) or None (for SimpleACK):
async def my_handler(
service_choice: int,
request_data: bytes,
source: BACnetAddress,
) -> bytes | None:
# Decode request_data, process, return response or None
...
Unconfirmed service handlers process the request without returning a response:
async def my_unconfirmed_handler(
service_choice: int,
request_data: bytes,
source: BACnetAddress,
) -> None:
# Decode and process, no response needed
...
Registering custom handlers¶
from bac_py.types.enums import ConfirmedServiceChoice, UnconfirmedServiceChoice
# Override a standard handler
app.service_registry.register_confirmed(
ConfirmedServiceChoice.CONFIRMED_PRIVATE_TRANSFER,
my_private_transfer_handler,
)
# Register an unconfirmed handler
app.service_registry.register_unconfirmed(
UnconfirmedServiceChoice.UNCONFIRMED_PRIVATE_TRANSFER,
my_unconfirmed_handler,
)
Custom validation example¶
Add a write-access filter that restricts writes to a whitelist of source addresses:
from bac_py.services.errors import BACnetError
from bac_py.services.property_access import WritePropertyRequest
from bac_py.types.enums import ErrorClass, ErrorCode
ALLOWED_WRITERS = {"192.168.1.10", "192.168.1.20"}
async def restricted_write_handler(service_choice, data, source):
if str(source) not in ALLOWED_WRITERS:
raise BACnetError(ErrorClass.SECURITY, ErrorCode.WRITE_ACCESS_DENIED)
# Fall through to default handling
return await default_handlers.handle_write_property(service_choice, data, source)
app.service_registry.register_confirmed(
ConfirmedServiceChoice.WRITE_PROPERTY,
restricted_write_handler,
)
Error responses¶
Handlers signal errors by raising exceptions:
from bac_py.services.errors import (
BACnetError, # -> Error-PDU (error_class, error_code)
BACnetRejectError, # -> Reject-PDU (reason)
BACnetAbortError, # -> Abort-PDU (reason)
)
# Property not found
raise BACnetError(ErrorClass.PROPERTY, ErrorCode.UNKNOWN_PROPERTY)
# Malformed request
raise BACnetRejectError(RejectReason.MISSING_REQUIRED_PARAMETER)
If no handler is registered for a confirmed service, the application
automatically sends a Reject-PDU with UNRECOGNIZED_SERVICE. Unregistered
unconfirmed services are silently ignored per Clause 5.4.2.
Event Engine¶
The EventEngine evaluates all 18 standard
BACnet event algorithms and generates event notifications routed through
NotificationClass recipient lists.
Starting the event engine¶
from bac_py.app.event_engine import EventEngine
from bac_py.objects.analog import AnalogInputObject
from bac_py.objects.notification import NotificationClassObject
from bac_py.types.enums import EngineeringUnits, EventType, NotifyType
async def serve_with_events():
config = DeviceConfig(instance_number=100, name="My-Device",
vendor_name="ACME", vendor_id=999)
async with BACnetApplication(config) as app:
# ... add device object and register handlers ...
# Notification class for routing alarm notifications
app.object_db.add(NotificationClassObject(
instance_number=1,
object_name="Critical-Alarms",
notification_class=1,
priority=[3, 3, 3], # [to_offnormal, to_fault, to_normal]
))
# Analog input with intrinsic out-of-range reporting
app.object_db.add(AnalogInputObject(
instance_number=1,
object_name="Zone-Temp",
units=EngineeringUnits.DEGREES_CELSIUS,
present_value=22.5,
high_limit=30.0,
low_limit=15.0,
deadband=1.0,
notification_class=1,
event_enable=[True, True, True],
notify_type=NotifyType.ALARM,
))
engine = EventEngine(app, scan_interval=1.0)
await engine.start()
try:
await app.run()
finally:
await engine.stop()
Supported algorithms¶
The engine evaluates these event types automatically based on object configuration:
CHANGE_OF_BITSTRING
CHANGE_OF_STATE
CHANGE_OF_VALUE
COMMAND_FAILURE
FLOATING_LIMIT
OUT_OF_RANGE
CHANGE_OF_LIFE_SAFETY
EXTENDED
BUFFER_READY
UNSIGNED_RANGE
ACCESS_EVENT
DOUBLE_OUT_OF_RANGE
SIGNED_OUT_OF_RANGE
UNSIGNED_OUT_OF_RANGE
CHANGE_OF_CHARACTERSTRING
CHANGE_OF_STATUS_FLAGS
CHANGE_OF_RELIABILITY
CHANGE_OF_DISCRETE_VALUE
Intrinsic reporting¶
Objects that define INTRINSIC_EVENT_ALGORITHM (AnalogInput, BinaryInput,
AnalogValue, etc.) automatically participate in event evaluation when their
Event_Enable property has at least one transition enabled and a valid
Notification_Class is assigned.
Algorithmic reporting¶
For custom event detection, create an
EventEnrollmentObject that
references the monitored object and specifies the algorithm parameters:
from bac_py.objects.event_enrollment import EventEnrollmentObject
from bac_py.types.enums import EventType, EventState
app.object_db.add(EventEnrollmentObject(
instance_number=1,
object_name="Temp-Out-Of-Range",
event_type=EventType.OUT_OF_RANGE,
notify_type=NotifyType.ALARM,
notification_class=1,
event_enable=[True, True, True],
))
The engine routes notifications through
NotificationClassObject recipient lists
with day/time filtering and per-recipient confirmed/unconfirmed delivery.
See Event Notifications for client-side event handling.
Audit Logging (Server Side)¶
The AuditManager instruments server handlers to
automatically record write, create, and delete operations as audit log entries
(new in ASHRAE 135-2020).
Setting up audit logging¶
from bac_py.app.audit import AuditManager
from bac_py.objects.audit_reporter import AuditReporterObject
from bac_py.objects.audit_log import AuditLogObject
# Create an audit reporter that monitors all objects
app.object_db.add(AuditReporterObject(
instance_number=1,
object_name="System-Auditor",
))
# Create an audit log buffer
app.object_db.add(AuditLogObject(
instance_number=1,
object_name="System-Audit-Log",
buffer_size=1000,
))
# The AuditManager is created in BACnetApplication and
# is invoked automatically by DefaultServerHandlers
# on write_property, create_object, and delete_object.
Key AuditReporter properties:
monitored_objects: List of ObjectIdentifiers to audit (empty = all)audit_level: Filtering level (NONE, DEFAULT, AUDIT_CONFIG)auditable_operations: BitString filter for operation types
Key AuditLog properties:
buffer_size: Maximum records before circular overwrite (default 100)stop_when_full: If True, stop logging when buffer is fulllog_enable: Enable/disable loggingrecord_count: Current records in buffertotal_record_count: Monotonically increasing sequence number
See Audit Logging for client-side audit queries.
Error Handling¶
Server handlers use a consistent error hierarchy:
from bac_py.services.errors import (
BACnetError, # Error-PDU (error_class, error_code)
BACnetRejectError, # Reject-PDU (reason)
BACnetAbortError, # Abort-PDU (reason)
)
Common server-side error responses:
Situation |
Exception |
Clause |
|---|---|---|
Unknown object |
|
12.1 |
Unknown property |
|
12.1 |
Write to read-only property |
|
15.9 |
Wrong password |
|
16.1 |
DCC disabled |
|
16.1 |
Missing parameter |
|
5.4 |
Value out of range |
|
15.9 |
Password validation¶
When DeviceConfig.password is set, the DeviceCommunicationControl and
ReinitializeDevice handlers validate the password using constant-time
comparison (hmac.compare_digest()). Requests with a missing or incorrect
password receive a PASSWORD_FAILURE error.
DeviceCommunicationControl states¶
The DCC handler supports three states:
ENABLE: Normal operation (default)
DISABLE: Only DeviceCommunicationControl and ReinitializeDevice requests are processed; all other confirmed services are rejected
DISABLE_INITIATION: The server suppresses outbound unsolicited messages but still responds to incoming requests
An optional time_duration (minutes) automatically re-enables the device
after the specified period.
Application Lifecycle¶
BACnetApplication manages the full lifecycle
of transport, network layer, TSM, COV manager, and engines.
Async context manager (recommended)¶
async with BACnetApplication(config) as app:
# app.start() called automatically
handlers = DefaultServerHandlers(app, app.object_db, device)
handlers.register()
await app.run()
# app.stop() called automatically on exit
Manual lifecycle¶
app = BACnetApplication(config)
await app.start() # Bind socket, start transport/network
try:
handlers = DefaultServerHandlers(app, app.object_db, device)
handlers.register()
await app.run()
finally:
await app.stop() # Idempotent, safe to call multiple times
Combined client and server¶
A single application can act as both client and server simultaneously:
from bac_py.app.client import BACnetClient
async with BACnetApplication(config) as app:
device = DeviceObject(instance_number=100, object_name="My-Device",
vendor_name="ACME", vendor_identifier=999)
app.object_db.add(device)
# Server side
handlers = DefaultServerHandlers(app, app.object_db, device)
handlers.register()
# Client side
bc = BACnetClient(app)
value = await bc.read("192.168.1.200", "ai,1", "pv")
await app.run()
Auto-computed properties¶
When DefaultServerHandlers.register() is called, it automatically
computes and sets:
Protocol_Services_Supported: BitString reflecting all registered service handlersProtocol_Object_Types_Supported: BitString for all object types supported by the library
Scheduling¶
Create a Schedule object with weekly time-value pairs and run the
ScheduleEngine to evaluate it:
import asyncio
from bac_py.app.application import BACnetApplication, DeviceConfig
from bac_py.app.schedule_engine import ScheduleEngine
from bac_py.objects.schedule import ScheduleObject
from bac_py.types.constructed import BACnetTimeValue
from bac_py.types.primitives import BACnetTime
async def serve_with_schedule():
config = DeviceConfig(instance_number=100, name="My-Device",
vendor_name="ACME", vendor_id=999)
async with BACnetApplication(config) as app:
# ... add device and other objects ...
# Occupied/unoccupied schedule (Mon-Fri 8am-6pm = 1, else = 0)
weekday_entries = [
BACnetTimeValue(time=BACnetTime(8, 0, 0, 0), value=1),
BACnetTimeValue(time=BACnetTime(18, 0, 0, 0), value=0),
]
app.object_db.add(ScheduleObject(
instance_number=1,
object_name="Occupancy-Schedule",
weekly_schedule=[
weekday_entries, # Monday
weekday_entries, # Tuesday
weekday_entries, # Wednesday
weekday_entries, # Thursday
weekday_entries, # Friday
[], # Saturday
[], # Sunday
],
schedule_default=0,
))
# Start the schedule engine
engine = ScheduleEngine(app, scan_interval=10.0)
await engine.start()
try:
await app.run()
finally:
await engine.stop()
asyncio.run(serve_with_schedule())
Trend Logging¶
Create a TrendLog object that records AnalogInput present-value readings
using the TrendLogEngine:
import asyncio
from bac_py.app.application import BACnetApplication, DeviceConfig
from bac_py.app.trendlog_engine import TrendLogEngine
from bac_py.objects.trendlog import TrendLogObject
from bac_py.types.enums import LoggingType, ObjectType, PropertyIdentifier
from bac_py.types.primitives import ObjectIdentifier
async def serve_with_trendlog():
config = DeviceConfig(instance_number=100, name="My-Device",
vendor_name="ACME", vendor_id=999)
async with BACnetApplication(config) as app:
# ... add device and AnalogInput objects ...
# Log ai,1 present-value every 60 seconds (polled mode)
app.object_db.add(TrendLogObject(
instance_number=1,
object_name="Zone-Temp-Log",
log_device_object_property=ObjectIdentifier(
ObjectType.ANALOG_INPUT, 1),
logging_type=LoggingType.POLLED,
log_interval=60, # seconds
buffer_size=1000,
))
engine = TrendLogEngine(app, scan_interval=1.0)
await engine.start()
try:
await app.run()
finally:
await engine.stop()
asyncio.run(serve_with_trendlog())
COV-based trend logging¶
For change-of-value recording (Clause 12.25.13), set logging_type to
LoggingType.COV. The engine registers a change callback on the monitored
local object and records a log entry whenever the value is written:
# Log ai,1 present-value on every change (COV mode)
app.object_db.add(TrendLogObject(
instance_number=2,
object_name="Zone-Temp-COV-Log",
log_device_object_property=ObjectIdentifier(
ObjectType.ANALOG_INPUT, 1),
logging_type=LoggingType.COV,
buffer_size=1000,
))
COV-mode trend logs do not poll. They only record when the monitored property is actually written, which can be more efficient for slowly changing values.
Registered Services¶
DefaultServerHandlers.register() installs handlers for the following
services:
Confirmed services:
ReadProperty, WriteProperty
ReadPropertyMultiple, WritePropertyMultiple
ReadRange
SubscribeCOV, SubscribeCOVProperty, SubscribeCOVPropertyMultiple
ConfirmedCOVNotificationMultiple
DeviceCommunicationControl, ReinitializeDevice
AtomicReadFile, AtomicWriteFile
CreateObject, DeleteObject
AddListElement, RemoveListElement
AcknowledgeAlarm, ConfirmedEventNotification
GetAlarmSummary, GetEnrollmentSummary, GetEventInformation
ConfirmedTextMessage
VT-Open, VT-Close, VT-Data
AuditLogQuery, ConfirmedAuditNotification
ConfirmedPrivateTransfer
Unconfirmed services:
Who-Is, Who-Has
TimeSynchronization, UTCTimeSynchronization
UnconfirmedCOVNotificationMultiple
UnconfirmedEventNotification
UnconfirmedTextMessage
WriteGroup
Who-Am-I, You-Are
UnconfirmedAuditNotification