BAC0.tasks package

Submodules

BAC0.tasks.DoOnce module

DoOnce.py - execute a task once

class BAC0.tasks.DoOnce.DoOnce(fnc)[source]

Bases: BAC0.tasks.TaskManager.OneShotTask

Start a polling task which is in fact a recurring read of the point. ex.

device[‘point_name’].poll(delay=60)
clear_notes()

Clear notes object

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.tasks.DoOnce | DoOnce'
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

task()[source]

BAC0.tasks.Match module

Match.py - verify a point’s status matches its commanded value.

Example:
Is a fan commanded to ‘On’ actually ‘running’?
class BAC0.tasks.Match.Match(status=None, command=None, delay=5, name=None)[source]

Bases: BAC0.tasks.TaskManager.Task

Match two properties of a BACnet Object (i.e. a point status with its command).

clear_notes()

Clear notes object

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.tasks.Match | Match'
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

stop()[source]
task()[source]
class BAC0.tasks.Match.Match_Value(value=None, point=None, delay=5, name=None, use_last_value=False)[source]

Bases: BAC0.tasks.TaskManager.Task

Verify a point’s Present_Value equals the given value after a delay of X seconds. Thus giving the BACnet controller (and connected equipment) time to respond to the command.

Match_Value(On, <AI:1>, 5)

i.e. Does Fan value = On after 5 seconds.

clear_notes()

Clear notes object

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.tasks.Match | Match_Value'
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

stop()[source]
task()[source]

BAC0.tasks.Poll module

Poll.py - create a Polling task to repeatedly read a point.

class BAC0.tasks.Poll.DeviceFastPoll(device, delay=1, name='')[source]

Bases: BAC0.tasks.Poll.DevicePoll

Start a fast polling task to repeatedly read a list of points from a device using ReadPropertyMultiple requests. Delay allowed will be 0 to 10 seconds Normal polling will limit the polling speed to 10 second minimum

Warning : Fast polling must be used with care or network flooding may occur

clear_notes()

Clear notes object

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.tasks.Poll | DeviceFastPoll'
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

class BAC0.tasks.Poll.DeviceNormalPoll(device, delay=10, name='')[source]

Bases: BAC0.tasks.Poll.DevicePoll

Start a normal polling task to repeatedly read a list of points from a device using ReadPropertyMultiple requests.

Normal polling will limit the polling speed to 10 second minimum

clear_notes()

Clear notes object

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.tasks.Poll | DeviceNormalPoll'
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

class BAC0.tasks.Poll.DevicePoll(device: Union[RPMDeviceConnected, RPDeviceConnected], delay: int = 10, name: str = '', prefix: str = 'basic_poll')[source]

Bases: BAC0.tasks.TaskManager.Task

Start a polling task to repeatedly read a list of points from a device using ReadPropertyMultiple requests.

clear_notes()

Clear notes object

device
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.tasks.Poll | DevicePoll'
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

task() → None[source]
exception BAC0.tasks.Poll.MultiplePollingFailures[source]

Bases: Exception

class BAC0.tasks.Poll.SimplePoll(point, *, delay: int = 10)[source]

Bases: BAC0.tasks.TaskManager.Task

Start a polling task to repeatedly read a point’s Present_Value. ex.

device[‘point_name’].poll(delay=60)
clear_notes()

Clear notes object

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.tasks.Poll | SimplePoll'
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

task()[source]

BAC0.tasks.RecurringTask module

RecurringTask.py - execute a recurring task

class BAC0.tasks.RecurringTask.RecurringTask(fnc: Union[Tuple[Callable, Any], Callable], delay: int = 60, name: str = 'recurring')[source]

Bases: BAC0.tasks.TaskManager.Task

Start a recurring task (a function passed)

clear_notes()

Clear notes object

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.tasks.RecurringTask | RecurringTask'
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

task() → None[source]

BAC0.tasks.TaskManager module

TaskManager.py - creation of threads used for repetitive tasks.

A key building block for point simulation.

class BAC0.tasks.TaskManager.Manager[source]

Bases: object

classmethod clean_tasklist()[source]
clear_notes()

Clear notes object

enable = False
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.tasks.TaskManager | Manager'
manager = 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

notes

Retrieve notes list as a Pandas Series

classmethod number_of_tasks()[source]
classmethod process()[source]
classmethod schedule_task(task)[source]
classmethod start_service()[source]
classmethod stopAllTasks()[source]
classmethod stop_service()[source]
tasks = []
class BAC0.tasks.TaskManager.OneShotTask(fn=None, args=None, name='Oneshot')[source]

Bases: BAC0.tasks.TaskManager.Task

clear_notes()

Clear notes object

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.tasks.TaskManager | OneShotTask'
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

class BAC0.tasks.TaskManager.Task(fn=None, name=None, delay=0)[source]

Bases: object

clear_notes()

Clear notes object

execute()[source]
high_latency = 60
is_alive()[source]
last_time
latency
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.tasks.TaskManager | Task'
next_time
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

start()[source]
stop()[source]
task()[source]
BAC0.tasks.TaskManager.random() → x in the interval [0, 1).
BAC0.tasks.TaskManager.stopAllTasks()[source]

Module contents