Events and Alarms¶
Alarm Management¶
Get alarm summaries, query enrollment summaries, check event information, and acknowledge alarms using string-based addressing:
import datetime
from bac_py import Client
from bac_py.types.constructed import BACnetTimeStamp
from bac_py.types.enums import AcknowledgmentFilter, EventState
from bac_py.types.primitives import BACnetTime
async with Client(instance_number=999) as client:
addr = "192.168.1.100"
# Get a summary of all active alarms
alarm_summary = await client.get_alarm_summary(addr)
for entry in alarm_summary.list_of_alarm_summaries:
print(f" {entry.object_identifier}: {entry.alarm_state}")
# Get enrollment summaries (all event-generating objects)
enrollment = await client.get_enrollment_summary(
addr, acknowledgment_filter=AcknowledgmentFilter.ALL,
)
for entry in enrollment.list_of_enrollment_summaries:
print(f" {entry.object_identifier}: {entry.event_type}")
# Get detailed event information (supports pagination)
event_info = await client.get_event_information(addr)
for summary in event_info.list_of_event_summaries:
print(f" {summary.object_identifier}: {summary.event_state}")
# Acknowledge an alarm (string object identifier)
now = datetime.datetime.now(tz=datetime.UTC)
ts = BACnetTimeStamp(choice=0, value=BACnetTime(now.hour, now.minute, now.second, 0))
await client.acknowledge_alarm(
addr,
acknowledging_process_identifier=1,
event_object_identifier="ai,1",
event_state_acknowledged=EventState.OFFNORMAL,
time_stamp=ts,
acknowledgment_source="operator",
time_of_acknowledgment=ts,
)
Text Messaging¶
Send confirmed (reliable) or unconfirmed (fire-and-forget) text messages:
from bac_py import Client
from bac_py.types.enums import MessagePriority
async with Client(instance_number=999) as client:
# Confirmed text message (waits for acknowledgment)
await client.send_text_message("192.168.1.100", "Maintenance at 2pm")
# Urgent confirmed message
await client.send_text_message(
"192.168.1.100", "High temperature alarm!",
message_priority=MessagePriority.URGENT,
)
# Unconfirmed broadcast message
await client.send_text_message(
"192.168.1.255", "System restart in 5 minutes",
confirmed=False,
)
Event Notifications¶
Configure intrinsic event reporting on an AnalogInput with high/low limits and a NotificationClass for routing notifications:
import asyncio
from bac_py.app.application import BACnetApplication, DeviceConfig
from bac_py.app.server import DefaultServerHandlers
from bac_py.objects.analog import AnalogInputObject
from bac_py.objects.device import DeviceObject
from bac_py.objects.notification import NotificationClassObject
from bac_py.types.enums import EngineeringUnits, EventState, 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:
device = DeviceObject(instance_number=100, object_name="My-Device",
vendor_name="ACME", vendor_identifier=999)
app.object_db.add(device)
# NotificationClass routes events to recipients
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
))
# AnalogInput 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,
))
handlers = DefaultServerHandlers(app, app.object_db, device)
handlers.register()
# The EventEngine starts automatically with the application
# and evaluates intrinsic reporting objects each scan cycle.
await app.run()
asyncio.run(serve_with_events())
Audit Logging¶
Audit logging is built into DefaultServerHandlers. When the server’s
object database contains an AuditReporterObject
and an AuditLogObject, write/create/delete
operations are automatically recorded:
from bac_py.objects.audit_log import AuditLogObject
from bac_py.objects.audit_reporter import AuditReporterObject
from bac_py.types.enums import AuditLevel
# Add audit objects to the server's object database
app.object_db.add(AuditReporterObject(
instance_number=1,
object_name="Audit-Reporter",
audit_level=AuditLevel.DEFAULT,
))
app.object_db.add(AuditLogObject(
instance_number=1,
object_name="Audit-Log",
buffer_size=1000,
))
# Now any WriteProperty, CreateObject, or DeleteObject handled by
# DefaultServerHandlers will automatically create audit records.