BAC0.core.devices package

Submodules

BAC0.core.devices.Device module

Device.py - describe a BACnet Device

class BAC0.core.devices.Device.Device(address=None, device_id=None, network=None, *, poll=10, from_backup=None, segmentation_supported=True, object_list=None, auto_save=False, save_resampling='1s', clear_history_on_save=False, history_size=None, reconnect_on_failure=True)[source]

Bases: BAC0.db.sql.SQLMixin

Represent a BACnet device. Once defined, it allows use of read, write, sim, release functions to communicate with the device on the network.

Parameters:
  • address ((str)) – address of the device (ex. ‘2:5’)
  • device_id (int) – bacnet device ID (boid)
  • network (BAC0.scripts.ReadWriteScript.ReadWriteScript) – defined by BAC0.connect()
  • poll – (int) if > 0, will poll every points each x seconds.
From_backup:

sqlite backup file

Segmentation_supported:
 

(boolean) When segmentation is not supported, BAC0 will not use read property multiple to poll the device.

Object_list:

(list) Use can provide a custom object_list to use for the the creation of the device. the object list must be built using the same pattern returned by bacpypes when polling the objectList property example

 my_obj_list = [('file', 1),
('analogInput', 2),
('analogInput', 3),
('analogInput', 5),
('analogInput', 4),
('analogInput', 0),
('analogInput', 1)]
Auto_save:

(False or int) 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.

Clear_history_on_save:
 

(boolean) Will clear device history

analog_units
binary_states
clear_histories()[source]
clear_notes()

Clear notes object

connect()[source]

Connect the device to the network

df(list_of_points, force_read=True)[source]

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()[source]
do(func)[source]
find_overrides(force=False)[source]
find_overrides_progress() → float[source]
find_point(objectType, objectAddress)[source]

Find point based on type and address

initialize_device_from_db()[source]
log(note, *, level=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'
multi_states
new_state(newstate)[source]

Base of the state machine mechanism. Used to make transitions between device states. Take care to call the state init function.

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

notes

Retrieve notes list as a Pandas Series

percent
points_name

When iterating a device, iterate points of it.

release_all_overrides(force=False)[source]
simulated_points

iterate over simulated points

Returns:points if simulated (out_of_service == True)
Return type:BAC0.core.devices.Points.Point
temperatures
to_excel()[source]

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

update_history_size(size=None)[source]
class BAC0.core.devices.Device.DeviceConnected(address=None, device_id=None, network=None, *, poll=10, from_backup=None, segmentation_supported=True, object_list=None, auto_save=False, save_resampling='1s', clear_history_on_save=False, history_size=None, reconnect_on_failure=True)[source]

Bases: BAC0.core.devices.Device.Device

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

analog_units

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

bacnet_properties
binary_states
connect(*, db=None)[source]

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)[source]

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

disconnect(save_on_disconnect=True, unregister=True)[source]
multi_states
percent
ping()[source]
points_name

When iterating a device, iterate points of it.

pollable_points_name
read_property(prop)[source]
temperatures
trendlogs
trendlogs_names
update_bacnet_properties()[source]

Retrieve bacnet properties for this device

update_description(value)[source]
write_property(prop, value, priority=None)[source]
class BAC0.core.devices.Device.DeviceDisconnected(address=None, device_id=None, network=None, *, poll=10, from_backup=None, segmentation_supported=True, object_list=None, auto_save=False, save_resampling='1s', clear_history_on_save=False, history_size=None, reconnect_on_failure=True)[source]

Bases: BAC0.core.devices.Device.Device

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

analog_units
binary_states
connect(*, db=None, network=None)[source]

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)[source]

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
multi_states
percent
points_name

When iterating a device, iterate points of it.

poll(command='start', *, delay=10)[source]
read_multiple(points_list, *, points_per_request=25, discover_request=(None, 6))[source]
simulated_points

iterate over simulated points

Returns:points if simulated (out_of_service == True)
Return type:BAC0.core.devices.Points.Point
temperatures
to_excel()[source]

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

class BAC0.core.devices.Device.DeviceFromDB(address=None, device_id=None, network=None, *, poll=10, from_backup=None, segmentation_supported=True, object_list=None, auto_save=False, save_resampling='1s', clear_history_on_save=False, history_size=None, reconnect_on_failure=True)[source]

Bases: BAC0.core.devices.Device.DeviceConnected

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

connect(*, network=None, from_backup=None)[source]
In DBState, a device can be reconnected to BACnet using:
device.connect(network=bacnet) (bacnet = BAC0.connect())
initialize_device_from_db()[source]
poll(command='start', *, delay=10)[source]
read_multiple(points_list, *, points_per_request=25, discover_request=(None, 6))[source]
simulated_points

iterate over simulated points

Returns:points if simulated (out_of_service == True)
Return type:BAC0.core.devices.Points.Point
to_excel()[source]

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

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

Bases: BAC0.core.devices.Device.DeviceFromDB

class BAC0.core.devices.Device.DeviceProperties[source]

Bases: object

This serves as a container for device properties

asdict
class BAC0.core.devices.Device.RPDeviceConnected(address=None, device_id=None, network=None, *, poll=10, from_backup=None, segmentation_supported=True, object_list=None, auto_save=False, save_resampling='1s', clear_history_on_save=False, history_size=None, reconnect_on_failure=True)[source]

Bases: BAC0.core.devices.Device.DeviceConnected, BAC0.core.devices.mixins.read_mixin.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=None, device_id=None, network=None, *, poll=10, from_backup=None, segmentation_supported=True, object_list=None, auto_save=False, save_resampling='1s', clear_history_on_save=False, history_size=None, reconnect_on_failure=True)[source]

Bases: BAC0.core.devices.Device.DeviceConnected, BAC0.core.devices.mixins.read_mixin.ReadPropertyMultiple

[Device state] If device is connected and supports ReadPropertyMultiple

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)[source]

Bases: BAC0.core.devices.Points.Point

Representation of a Boolean value

boolValue

returns : (boolean) Value

units

Boolean points don’t have units

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)[source]

Bases: BAC0.core.devices.Points.BooleanPoint

history

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

release(value, *, prop='presentValue', priority='')[source]

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

sim(value, *, prop='presentValue', priority='')[source]

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
value

Read the value from BACnet network

write(value, *, prop='presentValue', priority='')[source]

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.DateTimePoint(device=None, pointType=None, pointAddress=None, pointName=None, description=None, units_state=None, presentValue=None, history_size=None)[source]

Bases: BAC0.core.devices.Points.Point

Representation of DatetimeValue value

units

Characterstring value do not have units or state text

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)[source]

Bases: BAC0.core.devices.Points.Point

Representation of an Enumerated (multiState) value

enumValue

returns: (str) Enum state value

get_state(v)[source]
units

Enums have ‘state text’ instead of units.

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)[source]

Bases: BAC0.core.devices.Points.EnumPoint

enumValue

returns: (str) Enum state value

history

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

release(value, *, prop='presentValue', priority='')[source]

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

sim(value, *, prop='presentValue', priority='')[source]

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
value

Take last known value as the value

write(value, *, prop='presentValue', priority='')[source]

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)[source]

Bases: BAC0.core.devices.Points.Point

Representation of a Numeric value

units

Should return units

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)[source]

Bases: BAC0.core.devices.Points.NumericPoint

history

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

release(value, *, prop='presentValue', priority='')[source]

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

sim(value, *, prop='presentValue', priority='')[source]

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
units

Should return units

value

Take last known value as the value

write(value, *, prop='presentValue', priority='')[source]

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[source]

Bases: Exception

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

Bases: BAC0.core.devices.Points.Point

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

new_state(newstate)[source]
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=[])[source]

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.

auto()[source]
bacnet_properties
cancel_cov(callback=None)[source]
chart(remove=False)[source]

Add point to the bacnet trending list

clear_history()[source]
clear_notes()

Clear notes object

default(value)[source]
history

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

is_overridden
lastTimestamp

returns: last timestamp read

lastValue

returns: last value read

log(note, *, level=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)[source]
This allow functions like :
device[‘status’].match(‘command’)

A fan status for example will follow the command…

match_value(value, *, delay=5, use_last_value=False)[source]
This allow functions like :
device[‘point’].match(‘value’)

A sensor will follow a calculation…

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

notes

Retrieve notes list as a Pandas Series

out_of_service()[source]

Sets the Out_Of_Service property [to True].

ovr(value)[source]
poll(command='start', *, delay: int = 10) → None[source]

Poll a point every x seconds (delay=x sec) Stopped by using point.poll(‘stop’) or .poll(0) or .poll(False) or by setting a delay = 0

priority(priority=None)[source]
read_priority_array()[source]

Retrieve priority array of the point

read_property(prop)[source]
release()[source]

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

release_ovr()[source]
sim(value, *, force=False)[source]

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
subscribe_cov(confirmed=True, lifetime=None, callback=None)[source]
tag(tag_id, tag_value, lst=None)[source]

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

units

Should return units

update_bacnet_properties()[source]

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

update_description(value)[source]

This will write to the BACnet point and modify the description of the object

value

Retrieve value of the point

write(value, *, prop='presentValue', priority='')[source]

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[source]

Bases: object

A container for point properties.

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)[source]

Bases: BAC0.core.devices.Points.Point

Representation of CharacterString value

units

Characterstring value do not have units or state text

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)[source]

Bases: BAC0.core.devices.Points.EnumPoint

history

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

release(value, *, prop='presentValue', priority='')[source]

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

sim(value, *, prop='presentValue', priority='')[source]

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
value

Take last known value as the value

write(value, *, prop='presentValue', priority='')[source]

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.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

class BAC0.core.devices.Trends.TrendLog(OID, device=None, read_log_on_creation=True, multiple_request=None)[source]

Bases: BAC0.core.devices.Trends.TrendLogProperties

BAC0 simplification of TrendLog Object

chart(remove=False)[source]

Add point to the bacnet trending list

clear_notes()

Clear notes object

create_dataframe(log_buffer)[source]
history
log(note, *, level=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.Trends | TrendLog'
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

notes

Retrieve notes list as a Pandas Series

static read_logDatum(logDatum)[source]
read_log_buffer()[source]
update_properties()[source]
class BAC0.core.devices.Trends.TrendLogProperties[source]

Bases: object

A container for trend properties

name

BAC0.core.devices.create_objects module

BAC0.core.devices.create_objects.create_AI(oid=1, pv=0, name='AI', units=None)[source]
BAC0.core.devices.create_objects.create_AO(oid=1, pv=0, name='AO', units=None, pv_writable=False)[source]
BAC0.core.devices.create_objects.create_AV(oid=1, pv=0, name='AV', units=None, pv_writable=False)[source]
BAC0.core.devices.create_objects.create_BI(oid=1, pv=0, name='BI', activeText='On', inactiveText='Off')[source]
BAC0.core.devices.create_objects.create_BO(oid=1, pv=0, name='BO', activeText='On', inactiveText='Off', pv_writable=False)[source]
BAC0.core.devices.create_objects.create_BV(oid=1, pv=0, name='BV', activeText='On', inactiveText='Off', pv_writable=False)[source]
BAC0.core.devices.create_objects.create_CharStrValue(oid=1, pv='null', name='String', pv_writable=False)[source]
BAC0.core.devices.create_objects.create_DateTimeValue(oid=1, date=None, time=None, name='DateTime', pv_writable=False)[source]
BAC0.core.devices.create_objects.create_MV(oid=1, pv=0, name='MV', states=['red', 'green', 'blue'], pv_writable=False)[source]
BAC0.core.devices.create_objects.create_object(object_class, oid, objectName, description, presentValue=None, commandable=False)[source]
BAC0.core.devices.create_objects.create_object_list(objects_dict)[source]

d = {name: (name, description, presentValue, units, commandable)}

BAC0.core.devices.create_objects.deprecate_msg()[source]
BAC0.core.devices.create_objects.set_pv(obj=None, value=None, flags=[0, 0, 0, 0])[source]

Module contents