BAC0.tasks package

Submodules

BAC0.tasks.DoOnce module

DoOnce.py - execute a task once

class BAC0.tasks.DoOnce.DoOnce(fn: Callable, args: str | None = None)

Bases: OneShotTask

Execute a function once, optionally awaiting it if it’s a coroutine.

Example:

device['point_name'].poll(delay=60)

clear_notes()

Clear notes object

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

property notes

Retrieve notes list as a Pandas Series

async task()

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)

Bases: Task

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

clear_notes()

Clear notes object

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

property notes

Retrieve notes list as a Pandas Series

async stop()

async task()

class BAC0.tasks.Match.Match_Value(value=None, point=None, delay=5, name=None, use_last_value=False)

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

property notes

Retrieve notes list as a Pandas Series

async stop()

async task()

BAC0.tasks.Poll module

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

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

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

property notes

Retrieve notes list as a Pandas Series

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

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

property notes

Retrieve notes list as a Pandas Series

class BAC0.tasks.Poll.DeviceOneShotPoll(device, name='')

Bases: 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: 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.tasks.Poll | DeviceOneShotPoll'

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

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

Bases: Task

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

clear_notes()

Clear notes object

property device: RPMDeviceConnected | RPDeviceConnected | None

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

property notes

Retrieve notes list as a Pandas Series

async task() → None

exception BAC0.tasks.Poll.MultiplePollingFailures

Bases: Exception

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

Bases: Task

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

Example:

device['point_name'].poll(delay=60)

clear_notes()

Clear notes object

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

property notes

Retrieve notes list as a Pandas Series

async task()

BAC0.tasks.RecurringTask module

RecurringTask.py - execute a recurring task

class BAC0.tasks.RecurringTask.RecurringTask(fnc: Tuple[Callable, Any] | Callable | Coroutine, delay: int | float = 60, name: str = 'recurring', minimum_delay: int | float = 5)

Bases: Task

Start a recurring task (a function passed)

clear_notes()

Clear notes object

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

property notes

Retrieve notes list as a Pandas Series

async task() → None

BAC0.tasks.TaskManager module

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

A key building block for point simulation.

class BAC0.tasks.TaskManager.OneShotTask(fn=None, args=None, name=None)

Bases: Task

clear_notes()

Clear notes object

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

property notes

Retrieve notes list as a Pandas Series

class BAC0.tasks.TaskManager.Task(fn: Any = None, name: str | None = None, delay: float = 0, minimum_delay: float = 5)

Bases: object

async classmethod clean_tasklist(all: bool = False) → None

Async-clean the task list. If all=True, cancel and await all running tasks, otherwise remove finished tasks only.

clear_notes()

Clear notes object

property done

async execute()

is_alive()

property last_time

property latency

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

property 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

property notes

Retrieve notes list as a Pandas Series

classmethod number_of_tasks()

start()

stop()

async task()

tasks = []

BAC0.tasks.TaskManager.random() → x in the interval [0, 1).

async BAC0.tasks.TaskManager.stopAllTasks()

Module contents