BAC0.core.devices package

Subpackages

• BAC0.core.devices.local package
  • Submodules
  • BAC0.core.devices.local.decorator module
    • add_feature()
    • bacnet_properties()
    • create()
    • make_commandable()
    • make_outOfService()
  • BAC0.core.devices.local.object module
    • CharacterStringValueObject
    • DateTimeValueObject
    • DateValueObject
    • TrendLogObject
  • Module contents
• BAC0.core.devices.mixins package
  • Submodules
  • BAC0.core.devices.mixins.read_mixin module
    • DiscoveryUtilsMixin
      • DiscoveryUtilsMixin.read_objects_list()
      • DiscoveryUtilsMixin.rp_discovered_values()
    • RPMObjectsProcessing
    • RPObjectsProcessing
    • ReadProperty
      • ReadProperty.poll()
      • ReadProperty.read_multiple()
      • ReadProperty.read_single()
    • ReadPropertyMultiple
      • ReadPropertyMultiple.poll()
      • ReadPropertyMultiple.read_multiple()
      • ReadPropertyMultiple.read_single()
    • ReadUtilsMixin
    • TrendLogCreationException
    • batch_requests()
    • create_trendlogs()
    • retrieve_type()
    • to_float_if_possible()
  • Module contents

Submodules

BAC0.core.devices.Device module

Device.py - describe a BACnet Device

class BAC0.core.devices.Device.Device(address: str | None = None, device_id: int | None = None, network: Any | None = None, *, poll: int = 10, from_backup: str | None = None, segmentation_supported: bool = True, object_list: List | None = None, auto_save: bool = False, save_resampling: str = '1s', clear_history_on_save: bool = False, history_size: int | None = None, reconnect_on_failure: bool = True)

Bases: SQLMixin

This class represents a BACnet device. It provides methods to read, write, simulate, and release communication with the device on the network.

Parameters: address (str, optional): The address of the device (e.g., ‘2:5’). Defaults to None. device_id (int, optional): The BACnet device ID (boid). Defaults to None. network (BAC0.scripts.ReadWriteScript.ReadWriteScript, optional): Defined by BAC0.connect(). Defaults to None. poll (int, optional): If greater than 0, the device will poll every point each x seconds. Defaults to None. from_backup (str, optional): SQLite backup file. Defaults to None. segmentation_supported (bool, optional): When set to False, BAC0 will not use read property multiple to poll the device. Defaults to None. object_list (list, optional): User can provide a custom object list for the creation of the device. The object list must be built using the same pattern returned by bacpypes when polling the objectList property. Defaults to None. auto_save (bool or int, optional): If False or 0, auto_save is disabled. To activate, pass an integer representing the number of polls before auto_save is called. Will write the histories to SQLite db locally. Defaults to None. clear_history_on_save (bool, optional): If set to True, will clear device history. Defaults to None.

property analog_units: Dict[str, str]

property binary_states: Dict[str, str]

clear_histories() → None

clear_notes()

Clear notes object

connect() → None

Connect the device to the network

df(list_of_points: List[str], force_read: bool = True)

Build a pandas DataFrame from a list of points. DataFrames are used to present and analyze data.

Parameters:

list_of_points – a list of point names as str

Returns:

pd.DataFrame

disconnect(save_on_disconnect=True, unregister=True) → None

do(func: Any) → None

find_overrides(force: bool = False) → None

find_overrides_progress() → float

find_point(objectType: str, objectAddress: float) → Point

Find point based on type and address

initialize_device_from_db() → None

property initialized

log(note, *, level: str | int = 10)

Add a log entry…no note

log_subtitle(subtitle, args=None, width=35)

log_title(title, args=None, width=35)

logname = 'BAC0.core.devices.Device | Device'

property multi_states: Dict[str, str]

async new_state(newstate: Any) → None

Changes the state of the device.

This method forms the basis of the state machine mechanism and is used to transition between device states. It also calls the state initialization function.

Parameters:

newstate (Any) – The new state to transition to.

Returns:

None

note(note, *, level=20, log=True)

Add note to the object. By default, the note will also be logged :param note: (str) The note itself :param level: (logging.level) :param log: (boolean) Enable or disable logging of note

property notes

Retrieve notes list as a Pandas Series

property percent: Dict[str, str]

property points_name: List[str]

When iterating a device, iterate points of it.

release_all_overrides(force: bool = False) → None

property simulated_points: Iterator[Point]

iterate over simulated points

Returns:

points if simulated (out_of_service == True)

Return type:

BAC0.core.devices.Points.Point

property temperatures: Dict[str, str]

to_excel() → None

Using xlwings, make a dataframe of all histories and save it

update_history_size(size: int | None = None) → None

class BAC0.core.devices.Device.DeviceConnected(address: str | None = None, device_id: int | None = None, network: Any | None = None, *, poll: int = 10, from_backup: str | None = None, segmentation_supported: bool = True, object_list: List | None = None, auto_save: bool = False, save_resampling: str = '1s', clear_history_on_save: bool = False, history_size: int | None = None, reconnect_on_failure: bool = True)

Bases: Device

Find a device on the BACnet network. Set its state to ‘connected’. Once connected, all subsequent commands use this BACnet connection.

property analog_units

Shortcut to retrieve all analog points units [Used by Bokeh trending feature]

property bacnet_properties: Dict

property binary_states

async connect(*args, db=None, **kwargs)

A connected device can be switched to ‘database mode’ where the device will not use the BACnet network but instead obtain its contents from a previously stored database.

df(list_of_points, force_read=True)

When connected, calling DF should force a reading on the network.

property multi_states

property percent

async ping()

property points_name

When iterating a device, iterate points of it.

property pollable_points_name

async read_property(prop) → Any

property temperatures

property trendlogs

property trendlogs_names

async update_bacnet_properties()

Retrieve bacnet properties for this device

async update_description(value)

async write_property(prop, value, priority=None)

class BAC0.core.devices.Device.DeviceDisconnected(address: str | None = None, device_id: int | None = None, network: Any | None = None, *, poll: int = 10, from_backup: str | None = None, segmentation_supported: bool = True, object_list: List | None = None, auto_save: bool = False, save_resampling: str = '1s', clear_history_on_save: bool = False, history_size: int | None = None, reconnect_on_failure: bool = True)

Bases: Device

[Device state] Initial state of a device. Disconnected from BACnet.

property analog_units

property binary_states

async connect(*args, network=None, db=None, **kwargs)

Attempt to connect to device. If unable, attempt to connect to a controller database (so the user can use previously saved data).

df(list_of_points, force_read=True)

Build a pandas DataFrame from a list of points. DataFrames are used to present and analyze data.

Parameters:

list_of_points – a list of point names as str

Returns:

pd.DataFrame

property multi_states

property percent

property points_name

When iterating a device, iterate points of it.

poll(command='start', *, delay=10)

read_multiple(points_list, *, points_per_request=25, discover_request=(None, 6))

property simulated_points

iterate over simulated points

Returns:

points if simulated (out_of_service == True)

Return type:

BAC0.core.devices.Points.Point

property temperatures

to_excel()

Using xlwings, make a dataframe of all histories and save it

class BAC0.core.devices.Device.DeviceFromDB(address: str | None = None, device_id: int | None = None, network: Any | None = None, *, poll: int = 10, from_backup: str | None = None, segmentation_supported: bool = True, object_list: List | None = None, auto_save: bool = False, save_resampling: str = '1s', clear_history_on_save: bool = False, history_size: int | None = None, reconnect_on_failure: bool = True)

Bases: DeviceConnected

[Device state] Where requests for a point’s present value returns the last valid value from the point’s history.

async connect(*args, network=None, from_backup=None, **kwargs)

In DBState, a device can be reconnected to BACnet using:

device.connect(network=bacnet) (bacnet = BAC0.connect())

async initialize_device_from_db()

poll(command='start', *, delay=10)

read_multiple(points_list, *, points_per_request=25, discover_request=(None, 6))

property simulated_points

iterate over simulated points

Returns:

points if simulated (out_of_service == True)

Return type:

BAC0.core.devices.Points.Point

to_excel()

Using xlwings, make a dataframe of all histories and save it

class BAC0.core.devices.Device.DeviceLoad(filename=None)

Bases: DeviceFromDB

class BAC0.core.devices.Device.DeviceProperties

Bases: object

property asdict: Dict

class BAC0.core.devices.Device.RPDeviceConnected(address: str | None = None, device_id: int | None = None, network: Any | None = None, *, poll: int = 10, from_backup: str | None = None, segmentation_supported: bool = True, object_list: List | None = None, auto_save: bool = False, save_resampling: str = '1s', clear_history_on_save: bool = False, history_size: int | None = None, reconnect_on_failure: bool = True)

Bases: DeviceConnected, ReadProperty

[Device state] If device is connected but doesn’t support ReadPropertyMultiple

BAC0 will not poll such points automatically (since it would cause excessive network traffic). Instead manual polling must be used as needed via the poll() function.

class BAC0.core.devices.Device.RPMDeviceConnected(address: str | None = None, device_id: int | None = None, network: Any | None = None, *, poll: int = 10, from_backup: str | None = None, segmentation_supported: bool = True, object_list: List | None = None, auto_save: bool = False, save_resampling: str = '1s', clear_history_on_save: bool = False, history_size: int | None = None, reconnect_on_failure: bool = True)

Bases: DeviceConnected, ReadPropertyMultiple

[Device state] If device is connected and supports ReadPropertyMultiple

async BAC0.core.devices.Device.device(*args: Any, **kwargs: Any) → Device

BAC0.core.devices.Points module

Points.py - Definition of points so operations on Read results are more convenient.

class BAC0.core.devices.Points.BooleanPoint(device=None, pointType=None, pointAddress=None, pointName=None, description=None, presentValue=None, units_state=None, history_size=None)

Bases: Point

Representation of a Boolean value

property boolValue

returns : (boolean) Value

property units

Boolean points don’t have units

property value

Read the value from BACnet network

class BAC0.core.devices.Points.BooleanPointOffline(device=None, pointType=None, pointAddress=None, pointName=None, description=None, presentValue=None, units_state=None, history_size=None)

Bases: BooleanPoint

property history

returns : (pd.Series) containing timestamp and value of all readings

release(value, *, prop='presentValue', priority='')

Clears the Out_Of_Service property [to False] - so the controller regains control of the point.

sim(value, *, prop='presentValue', priority='', force=False)

Simulate a value. Sets the Out_Of_Service property- to disconnect the point from the controller’s control. Then writes to the Present_Value. The point name is added to the list of simulated points (self.simPoints)

Parameters:

value – (float) value to simulate

property value

Read the value from BACnet network

write(value, *, prop='presentValue', priority='')

Write to present value of a point

Parameters:

• value – (float) numeric value

• prop – (str) property to write. Default = presentValue

• priority – (int) priority to which write.

class BAC0.core.devices.Points.COVPointSubscription(point: Point | None = None, lifetime: int = 900, confirmed: bool = False, callback=None)

Bases: object

COVPointSubscription is a class that handles Change of Value (COV) subscriptions for BACnet points. It allows subscribing to COV notifications for a specific point, and handles the asynchronous communication with the BACnet device to receive these notifications.

Attributes:

address (Address): The BACnet address of the device. cov_fini (asyncio.Event): An event to signal the end of the COV subscription. task (asyncio.Task): The asyncio task that runs the COV subscription. obj_identifier (ObjectIdentifier): The BACnet object identifier for the point. _app (BACnetApplication): The BACnet application instance. process_identifier (int): The process identifier for the COV subscription. point (Point): The BAC0.point for which the COV subscription is created. lifetime (int): The lifetime of the COV subscription in seconds. confirmed (bool): Whether the COV notifications should be confirmed. callback (Optional[Union[Callable[[str, Any], None], Awaitable[None]]]): The callback function to be called when a COV notification is received.

Methods:

__init__(self, point: Point, lifetime: int = 900, confirmed: bool = False, callback: Optional[Union[Callable[[str, Any], None], Awaitable[None]]] = None):

Initializes the COVPointSubscription instance.

run(self):

Asynchronously runs the COV subscription, listening for COV notifications and calling the callback function if provided.

stop(self):

Stops the COV subscription by setting the cov_fini event.

async run()

stop()

class BAC0.core.devices.Points.DateTimePoint(device=None, pointType=None, pointAddress=None, pointName=None, description=None, units_state=None, presentValue=None, history_size=None)

Bases: Point

Representation of DatetimeValue value

property units

Characterstring value do not have units or state text

property value

Retrieve value of the point

class BAC0.core.devices.Points.EnumPoint(device=None, pointType=None, pointAddress=None, pointName=None, description=None, presentValue=None, units_state=None, history_size=None)

Bases: Point

Representation of an Enumerated (multiState) value

property enumValue

returns: (str) Enum state value

get_state(v)

property units

Enums have ‘state text’ instead of units.

property value

Retrieve value of the point

class BAC0.core.devices.Points.EnumPointOffline(device=None, pointType=None, pointAddress=None, pointName=None, description=None, presentValue=None, units_state=None, history_size=None)

Bases: EnumPoint

property enumValue

returns: (str) Enum state value

property history

returns : (pd.Series) containing timestamp and value of all readings

release(value, *, prop='presentValue', priority='')

Clears the Out_Of_Service property [to False] - so the controller regains control of the point.

sim(value, *, prop='presentValue', priority='', force=False)

Simulate a value. Sets the Out_Of_Service property- to disconnect the point from the controller’s control. Then writes to the Present_Value. The point name is added to the list of simulated points (self.simPoints)

Parameters:

value – (float) value to simulate

property value

Take last known value as the value

write(value, *, prop='presentValue', priority='')

Write to present value of a point

Parameters:

• value – (float) numeric value

• prop – (str) property to write. Default = presentValue

• priority – (int) priority to which write.

class BAC0.core.devices.Points.NumericPoint(device=None, pointType=None, pointAddress=None, pointName=None, description=None, presentValue=None, units_state=None, history_size=None)

Bases: Point

Representation of a Numeric value

property units

Should return units

property value

Retrieve value of the point

class BAC0.core.devices.Points.NumericPointOffline(device=None, pointType=None, pointAddress=None, pointName=None, description=None, presentValue=None, units_state=None, history_size=None)

Bases: NumericPoint

property history

returns : (pd.Series) containing timestamp and value of all readings

release(value, *, prop='presentValue', priority='')

Clears the Out_Of_Service property [to False] - so the controller regains control of the point.

sim(value, *, prop='presentValue', priority='', force=False)

Simulate a value. Sets the Out_Of_Service property- to disconnect the point from the controller’s control. Then writes to the Present_Value. The point name is added to the list of simulated points (self.simPoints)

Parameters:

value – (float) value to simulate

property units

Should return units

property value

Take last known value as the value

write(value, *, prop='presentValue', priority='')

Write to present value of a point

Parameters:

• value – (float) numeric value

• prop – (str) property to write. Default = presentValue

• priority – (int) priority to which write.

exception BAC0.core.devices.Points.OfflineException

Bases: Exception

class BAC0.core.devices.Points.OfflinePoint(device, name)

Bases: Point

When offline (DB state), points needs to behave in a particular way (we can’t read on bacnet…)

new_state(newstate)

class BAC0.core.devices.Points.Point(device=None, pointType=None, pointAddress=None, pointName=None, description=None, presentValue=None, units_state=None, history_size=None, tags=[])

Bases: object

Represents a device BACnet point. Used to NumericPoint, BooleanPoint and EnumPoints.

Each point implements a history feature. Each time the point is read, its value (with timestamp) is added to a history table. Histories capture the changes to point values over time.

async auto()

property bacnet_properties

async cancel_cov()

chart(remove=False)

Add point to the bacnet trending list

clear_history()

clear_notes()

Clear notes object

async default(value)

property history: Dict[datetime, int | float | str]

returns : (pd.Series) containing timestamp and value of all readings

async is_out_of_service()

Check if the Out_Of_Service property is true.

property is_overridden

property lastTimestamp

returns: last timestamp read

property lastValue

returns: last value read

log(note, *, level: str | int = 10)

Add a log entry…no note

log_subtitle(subtitle, args=None, width=35)

log_title(title, args=None, width=35)

logname = 'BAC0.core.devices.Points | Point'

match(point, *, delay=5)

match_value(value, *, delay=5, use_last_value=False)

note(note, *, level=20, log=True)

Add note to the object. By default, the note will also be logged :param note: (str) The note itself :param level: (logging.level) :param log: (boolean) Enable or disable logging of note

property notes

Retrieve notes list as a Pandas Series

async out_of_service()

Sets the Out_Of_Service property [to True].

async ovr(value)

poll(command='start', *, delay: int = 10) → None

async priority(priority=None)

async read_priority_array()

Retrieve priority array of the point

async read_property(prop)

async release(value, *, prop='presentValue', priority='')

Clears the Out_Of_Service property [to False] - so the controller regains control of the point.

async release_ovr()

async sim(value, *, force=False)

Simulate a value. Sets the Out_Of_Service property- to disconnect the point from the controller’s control. Then writes to the Present_Value. The point name is added to the list of simulated points (self.simPoints)

Parameters:

value – (float) value to simulate

property status

async subscribe_cov(confirmed: bool = False, lifetime: int = 900, callback=None)

Subscribes to the Change of Value (COV) service for this point.

The COV service allows the device to notify the application of changes to the value of a property. This method sets up the subscription and starts an asynchronous task to listen for these notifications.

Args:

confirmed (bool, optional): If True, the device will wait for a confirmation from the application

after sending a COV notification. Defaults to True.

lifetime (int, optional): The lifetime of the subscription in seconds. If None, the subscription

will last indefinitely. Defaults to None.

callback (function, optional): A function to be called when a COV notification is received.

The function should accept two arguments: the sender (the device) and the args (the notification).

Raises:

RuntimeError: If the task is already running, a RuntimeError will be raised.

Returns:

None

tag(tag_id, tag_value, lst=None)

Add tag to point. Those tags can be used to make queries, add information, etc. They will be included if InfluxDB is used.

property units

Should return units

async update_bacnet_properties()

Retrieve bacnet properties for this point To retrieve something general, forcing vendor id 0

update_description(value)

property value

Retrieve value of the point

async write(value: Any, *, prop: str = 'presentValue', priority: int = 16)

Write to present value of a point

Parameters:

• value – (float) numeric value

• prop – (str) property to write. Default = presentValue

• priority – (int) priority to which write.

class BAC0.core.devices.Points.PointProperties

Bases: object

A container for point properties.

property asdict

class BAC0.core.devices.Points.StringPoint(device=None, pointType=None, pointAddress=None, pointName=None, description=None, units_state=None, presentValue=None, history_size=None)

Bases: Point

Representation of CharacterString value

property units

Characterstring value do not have units or state text

property value

Retrieve value of the point

class BAC0.core.devices.Points.StringPointOffline(device=None, pointType=None, pointAddress=None, pointName=None, description=None, presentValue=None, units_state=None, history_size=None)

Bases: EnumPoint

property history

returns : (pd.Series) containing timestamp and value of all readings

release(value, *, prop='presentValue', priority='')

Clears the Out_Of_Service property [to False] - so the controller regains control of the point.

sim(value, *, prop='presentValue', priority='', force=False)

Simulate a value. Sets the Out_Of_Service property- to disconnect the point from the controller’s control. Then writes to the Present_Value. The point name is added to the list of simulated points (self.simPoints)

Parameters:

value – (float) value to simulate

property value

Take last known value as the value

write(value, *, prop='presentValue', priority='')

Write to present value of a point

Parameters:

• value – (float) numeric value

• prop – (str) property to write. Default = presentValue

• priority – (int) priority to which write.

BAC0.core.devices.Points.extract_value_from_primitive_data(value)

BAC0.core.devices.Trends module

class BAC0.core.devices.Trends.HistoryComponent(index, logdatum, status, choice)

Bases: tuple

choice

Alias for field number 3

index

Alias for field number 0

logdatum

Alias for field number 1

status

Alias for field number 2

BAC0.core.devices.Trends.TrendLog(*args, **kwargs) → _TrendLog

class BAC0.core.devices.Trends.TrendLogProperties

Bases: object

A container for trend properties

property name: str | None

Module contents