Engines

COV Manager

COV (Change of Value) subscription manager per ASHRAE 135-2016 Clause 13.1.

class bac_py.app.cov.COVSubscription(subscriber, process_id, monitored_object, confirmed, lifetime, created_at=<factory>, expiry_handle=None, last_present_value=None, last_status_flags=None)[source]

Bases: object

Tracks a single COV subscription.

Parameters:
subscriber: BACnetAddress

BACnet address of the subscribing device.

process_id: int

Subscriber-assigned process identifier.

monitored_object: ObjectIdentifier

Object identifier being monitored.

confirmed: bool

True for confirmed notifications, False for unconfirmed.

lifetime: float | None

Subscription duration in seconds, or None for indefinite.

created_at: float

Monotonic timestamp when the subscription was created.

expiry_handle: TimerHandle | None = None

Timer handle for subscription expiry, if any.

last_present_value: Any = None

Last notified Present_Value, used for COV comparison.

last_status_flags: Any = None

Last notified Status_Flags, used for change detection.

class bac_py.app.cov.PropertySubscription(subscriber, process_id, monitored_object, monitored_property, property_array_index, confirmed, lifetime, cov_increment, created_at=<factory>, last_value=None, expiry_handle=None)[source]

Bases: object

Tracks a single property-level COV subscription (Clause 13.15/13.16).

Parameters:
subscriber: BACnetAddress

BACnet address of the subscribing device.

process_id: int

Subscriber-assigned process identifier.

monitored_object: ObjectIdentifier

Object identifier being monitored.

monitored_property: int

Property identifier being monitored.

property_array_index: int | None

Optional array index within the monitored property.

confirmed: bool

True for confirmed notifications, False for unconfirmed.

lifetime: float | None

Subscription duration in seconds, or None for indefinite.

cov_increment: float | None

Subscription-specific COV increment override, or None.

created_at: float

Monotonic timestamp when the subscription was created.

last_value: Any = None

Last notified value for the monitored property.

expiry_handle: TimerHandle | None = None

Timer handle for subscription expiry, if any.

class bac_py.app.cov.COVManager(app, *, max_subscriptions=1000, max_property_subscriptions=1000)[source]

Bases: object

Manages COV subscriptions and notification dispatch.

Per Clause 13.1, COV notifications are sent when: - Analog objects: |new - last| >= COV_INCREMENT (any change if no increment set) - Binary/multistate objects: any change in Present_Value - Any object: change in Status_Flags

Parameters:
subscribe(subscriber, request, object_db)[source]

Add or update a COV subscription.

Parameters:
Raises:

BACnetError – If the monitored object does not exist.

Return type:

None

unsubscribe(subscriber, process_id, monitored_object)[source]

Remove a subscription (cancellation).

Silently ignores if no matching subscription exists.

Parameters:

  • subscriber (BACnetAddress) – Address of the subscribing device.

  • process_id (int) – Subscriber-assigned process identifier.

  • monitored_object (ObjectIdentifier) – Object identifier being unsubscribed.

Return type:

None

check_and_notify(obj, changed_property)[source]

Check all subscriptions for this object and send notifications if needed.

Called after a property write. Compares current values against last-reported values using COV increment logic per Clause 13.1.

Parameters:
Return type:

None

get_active_subscriptions(object_id=None)[source]

Return active subscriptions, optionally filtered by object.

Return type:

list[COVSubscription]

Parameters:

object_id (ObjectIdentifier | None)

shutdown()[source]

Cancel all subscription timers.

Return type:

None

remove_object_subscriptions(object_id)[source]

Remove all subscriptions for a deleted object.

Called when an object is removed from the database to clean up any outstanding COV subscriptions per Clause 13.1.

Return type:

None

Parameters:

object_id (ObjectIdentifier)

subscribe_property(subscriber, request, object_db)[source]

Add or update a property-level COV subscription (Clause 13.15).

Parameters:
Raises:

BACnetError – If the monitored object does not exist.

Return type:

None

subscribe_property_multiple(subscriber, request, object_db)[source]

Add or update multiple property-level COV subscriptions (Clause 13.16).

Parameters:
Raises:

BACnetError – If any monitored object does not exist.

Return type:

None

unsubscribe_property(subscriber, process_id, obj_id, property_id, array_index=None)[source]

Remove a property-level subscription (cancellation).

Silently ignores if no matching subscription exists.

Parameters:

  • subscriber (BACnetAddress) – Address of the subscribing device.

  • process_id (int) – Subscriber-assigned process identifier.

  • obj_id (ObjectIdentifier) – Object identifier being monitored.

  • property_id (int) – Property identifier value being monitored.

  • array_index (int | None (default: None)) – Optional array index within the property.

Return type:

None

check_and_notify_property(obj, changed_property)[source]

Check property-level subscriptions and send notifications if needed.

For all property subscriptions matching this object and the changed property, check if the value changed enough to trigger a notification. For analog types, a subscription-specific cov_increment overrides the object’s COV_INCREMENT. For non-analog types, any change triggers a notification.

Parameters:
Return type:

None

Event Engine

Event state machine, algorithm evaluators, and async engine per ASHRAE 135-2020 Clause 13.

The EventStateMachine implements the state transition logic of Clause 13.2. Each evaluate_* function implements one of the 18 event algorithm evaluators defined in Clause 13.3.

The EventEngine is the async integration layer that periodically evaluates EventEnrollment objects and intrinsic-reporting objects, drives the state machines, and dispatches EventNotificationRequest PDUs on state transitions.

The state machine and evaluators are pure logic – no async, no I/O, no side effects. The EventEngine provides the async scheduling and notification dispatch wrapper.

class bac_py.app.event_engine.EventTransition(from_state, to_state, timestamp)[source]

Bases: object

Result of a state-machine evaluation that triggered a transition.

Parameters:

  • from_state (EventState) – The state before the transition.

  • to_state (EventState) – The state after the transition.

  • timestamp (float) – Monotonic time when the transition fired.

from_state: EventState
to_state: EventState
timestamp: float
class bac_py.app.event_engine.EventStateMachine(event_state=EventState.NORMAL, event_enable=<factory>, acked_transitions=<factory>, time_delay=0.0, time_delay_normal=None, _pending_state=None, _pending_since=None)[source]

Bases: object

Per-enrollment event state machine (Clause 13.2).

Tracks state, timestamps, acknowledgments, and time-delay logic. Call evaluate() each scan cycle with the event algorithm result and the fault algorithm result. Returns an EventTransition when a state change fires, or None when no change occurs.

Parameters:

  • event_state (EventState (default: <EventState.NORMAL: 0>)) – Current event state.

  • event_enable (list[bool] (default: <factory>)) – Three-element list [to_offnormal, to_fault, to_normal].

  • acked_transitions (list[bool] (default: <factory>)) – Three-element list [to_offnormal, to_fault, to_normal] indicating which transitions have been acknowledged.

  • time_delay (float (default: 0.0)) – Seconds the event condition must persist before transitioning to an alarm state.

  • time_delay_normal (float | None (default: None)) – Seconds the normal condition must persist before returning to NORMAL. Defaults to time_delay when None.

  • _pending_state (EventState | None)

  • _pending_since (float | None)

event_state: EventState
event_enable: list[bool]
acked_transitions: list[bool]
time_delay: float
time_delay_normal: float | None
property effective_time_delay_normal: float

Return the effective time-delay-normal value.

evaluate(event_result, fault_result, current_time)[source]

Evaluate one scan cycle and return a transition if one fires.

Parameters:

  • event_result (EventState | None) – Target EventState from the event algorithm, or None if no alarm condition is detected.

  • fault_result (Reliability) – Reliability from the fault algorithm. Any value other than NO_FAULT_DETECTED indicates a fault.

  • current_time (float) – Monotonic clock value (seconds).

Return type:

EventTransition | None

Returns:

An EventTransition if a state change fires, None otherwise.

bac_py.app.event_engine.evaluate_out_of_range(value, high_limit, low_limit, deadband, *, current_state=EventState.NORMAL)[source]

Evaluate OUT_OF_RANGE (Clause 13.3.6).

Parameters:

  • value (float) – Current monitored real value.

  • high_limit (float) – High-limit threshold.

  • low_limit (float) – Low-limit threshold.

  • deadband (float) – Hysteresis value for returning to normal.

  • current_state (EventState (default: <EventState.NORMAL: 0>)) – The current event state (for deadband logic).

Return type:

EventState | None

Returns:

Target EventState or None.

bac_py.app.event_engine.evaluate_double_out_of_range(value, high_limit, low_limit, deadband, *, current_state=EventState.NORMAL)[source]

Evaluate DOUBLE_OUT_OF_RANGE (Clause 13.3.14).

Same logic as OUT_OF_RANGE but for Double precision values.

Return type:

EventState | None

Parameters:
bac_py.app.event_engine.evaluate_signed_out_of_range(value, high_limit, low_limit, deadband, *, current_state=EventState.NORMAL)[source]

Evaluate SIGNED_OUT_OF_RANGE (Clause 13.3.15).

Same logic as OUT_OF_RANGE but for Signed integer values.

Return type:

EventState | None

Parameters:
bac_py.app.event_engine.evaluate_unsigned_out_of_range(value, high_limit, low_limit, deadband, *, current_state=EventState.NORMAL)[source]

Evaluate UNSIGNED_OUT_OF_RANGE (Clause 13.3.16).

Same logic as OUT_OF_RANGE but for Unsigned integer values.

Return type:

EventState | None

Parameters:
bac_py.app.event_engine.evaluate_unsigned_range(value, high_limit, low_limit)[source]

Evaluate UNSIGNED_RANGE (Clause 13.3.11).

Simpler variant with no deadband.

Parameters:

  • value (int) – Current monitored unsigned value.

  • high_limit (int) – High-limit threshold.

  • low_limit (int) – Low-limit threshold.

Return type:

EventState | None

Returns:

Target EventState or None.

bac_py.app.event_engine.evaluate_floating_limit(value, setpoint, high_diff_limit, low_diff_limit, deadband, *, current_state=EventState.NORMAL)[source]

Evaluate FLOATING_LIMIT (Clause 13.3.5).

Limits are relative to setpoint: setpoint + high_diff_limit and setpoint - low_diff_limit.

Parameters:

  • value (float) – Current monitored real value.

  • setpoint (float) – Reference setpoint.

  • high_diff_limit (float) – Positive offset above setpoint for high limit.

  • low_diff_limit (float) – Positive offset below setpoint for low limit.

  • deadband (float) – Hysteresis value.

  • current_state (EventState (default: <EventState.NORMAL: 0>)) – Current event state for deadband logic.

Return type:

EventState | None

Returns:

Target EventState or None.

bac_py.app.event_engine.evaluate_change_of_state(value, alarm_values)[source]

Evaluate CHANGE_OF_STATE (Clause 13.3.2).

Parameters:

  • value (int) – Current enumerated property value (as int).

  • alarm_values (tuple[int, ...]) – Tuple of enumerated values that trigger OFFNORMAL.

Return type:

EventState | None

Returns:

OFFNORMAL if value is in alarm_values, else None.

bac_py.app.event_engine.evaluate_change_of_bitstring(value, bitmask, alarm_values)[source]

Evaluate CHANGE_OF_BITSTRING (Clause 13.3.1).

Applies bitmask to value and checks if the masked result matches any entry in alarm_values.

Parameters:

  • value (tuple[int, ...]) – Current bitstring as tuple of bit values (0/1).

  • bitmask (tuple[int, ...]) – Bitmask to AND with value.

  • alarm_values (tuple[tuple[int, ...], ...]) – Set of masked bitstring values that trigger OFFNORMAL.

Return type:

EventState | None

Returns:

OFFNORMAL if masked value matches any alarm value, else None.

bac_py.app.event_engine.evaluate_change_of_life_safety(tracking_value, mode, alarm_values, life_safety_alarm_values)[source]

Evaluate CHANGE_OF_LIFE_SAFETY (Clause 13.3.8).

Parameters:

  • tracking_value (LifeSafetyState) – Current life-safety state.

  • mode (int) – Current life-safety mode (as int).

  • alarm_values (tuple[int, ...]) – States triggering OFFNORMAL.

  • life_safety_alarm_values (tuple[int, ...]) – States triggering LIFE_SAFETY_ALARM.

Return type:

EventState | None

Returns:

Target state or None.

bac_py.app.event_engine.evaluate_change_of_characterstring(value, alarm_values)[source]

Evaluate CHANGE_OF_CHARACTERSTRING (Clause 13.3.17).

Parameters:

  • value (str) – Current character string value.

  • alarm_values (tuple[str, ...]) – Strings that trigger OFFNORMAL.

Return type:

EventState | None

Returns:

OFFNORMAL if value is in alarm_values, else None.

bac_py.app.event_engine.evaluate_access_event(access_event, access_event_list)[source]

Evaluate ACCESS_EVENT (Clause 13.3.13).

Parameters:

  • access_event (int) – Current access event value (as int).

  • access_event_list (tuple[int, ...]) – Events that trigger OFFNORMAL.

Return type:

EventState | None

Returns:

OFFNORMAL if access_event is in the list, else None.

bac_py.app.event_engine.evaluate_change_of_value(value, previous_value, cov_increment)[source]

Evaluate CHANGE_OF_VALUE (Clause 13.3.3).

Triggers OFFNORMAL when the absolute change since the last reported value exceeds cov_increment.

Parameters:

  • value (float) – Current monitored value.

  • previous_value (float) – Value at last notification.

  • cov_increment (float) – Minimum change to trigger.

Return type:

EventState | None

Returns:

OFFNORMAL if change exceeds increment, else None.

bac_py.app.event_engine.evaluate_change_of_status_flags(current_flags, previous_flags, selected_flags)[source]

Evaluate CHANGE_OF_STATUS_FLAGS (Clause 13.3.18).

Triggers OFFNORMAL when any selected flag has changed from its previous value.

Parameters:

  • current_flags (tuple[bool, ...]) – Current status flags (in_alarm, fault, overridden, out_of_service).

  • previous_flags (tuple[bool, ...]) – Previous status flags at last notification.

  • selected_flags (tuple[bool, ...]) – Which flags to monitor (True = monitor).

Return type:

EventState | None

Returns:

OFFNORMAL if any selected flag changed, else None.

bac_py.app.event_engine.evaluate_change_of_reliability(reliability)[source]

Evaluate CHANGE_OF_RELIABILITY (Clause 13.3.19).

Parameters:

reliability (Reliability) – Current reliability value.

Return type:

EventState | None

Returns:

OFFNORMAL if reliability is not NO_FAULT_DETECTED, else None.

bac_py.app.event_engine.evaluate_command_failure(feedback_value, command_value)[source]

Evaluate COMMAND_FAILURE (Clause 13.3.4).

Triggers OFFNORMAL when the feedback value does not match the commanded value (the time-delay enforcement is handled by the state machine, not this evaluator).

Parameters:

  • feedback_value (Any) – Current feedback property value.

  • command_value (Any) – Most recent commanded value.

Return type:

EventState | None

Returns:

OFFNORMAL if feedback != command, else None.

bac_py.app.event_engine.evaluate_buffer_ready(current_count, previous_count, notification_threshold)[source]

Evaluate BUFFER_READY (Clause 13.3.10).

Triggers OFFNORMAL when the number of new records since the last notification meets or exceeds the threshold.

Parameters:

  • current_count (int) – Current record count.

  • previous_count (int) – Record count at last notification.

  • notification_threshold (int) – Minimum new records to trigger.

Return type:

EventState | None

Returns:

OFFNORMAL if threshold met, else None.

bac_py.app.event_engine.evaluate_extended(monitored_value, params, *, vendor_callback=None)[source]

Evaluate EXTENDED (Clause 13.3.9).

Vendor-specific algorithm. Delegates to vendor_callback if provided.

Parameters:

  • monitored_value (Any) – Current property value.

  • params (Any) – Vendor-specific parameters.

  • vendor_callback (Callable[[Any, Any], EventState | None] | None (default: None)) – Optional callable implementing vendor logic.

Return type:

EventState | None

Returns:

Target state from callback, or None.

bac_py.app.event_engine.evaluate_change_of_timer(timer_state, alarm_values)[source]

Evaluate CHANGE_OF_TIMER (Clause 13.3.20, new in 2020).

Parameters:

  • timer_state (TimerState) – Current timer state.

  • alarm_values (tuple[int, ...]) – Timer state values (as int) that trigger OFFNORMAL.

Return type:

EventState | None

Returns:

OFFNORMAL if current state is in alarm values, else None.

bac_py.app.event_engine.evaluate_change_of_discrete_value(current_value, previous_value)[source]

Evaluate CHANGE_OF_DISCRETE_VALUE (Clause 13.3.21, new in 2020).

Fires OFFNORMAL when the monitored property’s discrete value changes. Applies to Integer, Unsigned, Large-Analog, and other discrete-valued objects.

Parameters:

  • current_value (Any) – Current monitored property value.

  • previous_value (Any) – Previous monitored property value.

Return type:

EventState | None

Returns:

OFFNORMAL if values differ, else None.

class bac_py.app.event_engine.EventEngine(app, *, scan_interval=1.0)[source]

Bases: object

Async event/alarm evaluation engine per Clause 13.

Mirrors the COVManager lifecycle pattern:

  • Constructed with a reference to BACnetApplication.

  • start() launches the periodic evaluation loop.

  • stop() cancels the loop and cleans up.

Each evaluation cycle iterates EventEnrollment objects and intrinsic-reporting objects in the object database, runs fault algorithms (Clause 13.4) then event algorithms (Clause 13.3), feeds results to per-enrollment EventStateMachine instances, and dispatches EventNotificationRequest PDUs on transitions.

Parameters:
async start()[source]

Start the periodic evaluation loop.

Return type:

None

async stop()[source]

Stop the evaluation loop and clean up.

Return type:

None

Audit Manager

Audit logging manager per ASHRAE 135-2020 Clause 19.6.

Intercepts auditable operations and records them into Audit Log objects.

class bac_py.app.audit.AuditManager(object_db)[source]

Bases: object

Manages audit logging per Clause 19.6.

Intercepts auditable operations (writes, creates, deletes, etc.) and records them into Audit Log objects. Checks audit level and auditable operations filters before recording.

Parameters:

object_db (ObjectDatabase)

record_operation(operation, source_device=None, target_object=None, target_property=None, target_array_index=None, target_priority=None, target_value=None, current_value=None, invoke_id=None, result_error=None, source_comment=None, target_comment=None)[source]

Check audit config and record if auditable.

  1. Find Audit Reporter object(s)

  2. Resolve effective audit level

  3. Check auditable_operations bitstring filter

  4. Construct BACnetAuditNotification

  5. Append to local Audit Log buffer

Parameters:

  • operation (AuditOperation) – The audit operation being recorded.

  • source_device (ObjectIdentifier | None (default: None)) – Object identifier of the device that initiated the operation, or None if unknown.

  • target_object (ObjectIdentifier | None (default: None)) – Object identifier of the affected object.

  • target_property (int | None (default: None)) – Property identifier value of the affected property, or None if not property-specific.

  • target_array_index (int | None (default: None)) – Array index within the property, or None if not applicable.

  • target_priority (int | None (default: None)) – Priority level for commandable writes, or None if not applicable.

  • target_value (bytes | None (default: None)) – Encoded bytes of the new value written, or None if not applicable.

  • current_value (bytes | None (default: None)) – Encoded bytes of the value before the operation, or None if not captured.

  • invoke_id (int | None (default: None)) – BACnet invoke ID from the request, or None.

  • result_error (tuple[int, int] | None (default: None)) – Error class and code tuple if the operation failed, or None on success.

  • source_comment (str | None (default: None)) – Free-text comment from the source device.

  • target_comment (str | None (default: None)) – Free-text comment about the target.

Return type:

None

Schedule Engine

Schedule and Calendar evaluation engine per ASHRAE 135-2020 Clause 12.24.

The ScheduleEngine follows the same async lifecycle pattern as EventEngine (start/stop/periodic loop). On each cycle it:

  1. Evaluates all Calendar objects (updating present_value).

  2. Evaluates all Schedule objects following the resolution order in Clause 12.24.4–12.24.9: effective_period → exception_schedule → weekly_schedule → schedule_default.

  3. On value change, writes the new value to each target listed in list_of_object_property_references at priority_for_writing.

class bac_py.app.schedule_engine.ScheduleEngine(app, *, scan_interval=10.0)[source]

Bases: object

Async engine that evaluates Calendar and Schedule objects periodically.

Parameters:
async start()[source]

Start the periodic evaluation loop.

Return type:

None

async stop()[source]

Stop the evaluation loop and clean up.

Return type:

None

TrendLog Engine

Trend Log recording engine per ASHRAE 135-2020 Clause 12.25.

The TrendLogEngine follows the same async lifecycle pattern as EventEngine. It manages polled, triggered, and COV-based recording for all TrendLogObject instances in the object database.

COV-based logging (Clause 12.25.13) uses property-change callbacks on local objects to record values when the monitored property changes.

class bac_py.app.trendlog_engine.TrendLogEngine(app, *, scan_interval=1.0)[source]

Bases: object

Async engine that drives polled, triggered, and COV-based trend log recording.

Parameters:
async start()[source]

Start the periodic recording loop.

Return type:

None

async stop()[source]

Stop the recording loop and clean up COV subscriptions.

Return type:

None