Question regarding speeding up data collection

[SaiChaitanya1320]

Dear @ChristianTremblay,
Thank you for all your work on BAC0. I was working on collecting BACnet data from devices but speeding things up using asyncio or threading. However, the issue is that the controller disconnects and reconnects every iteration (I understand it kills the process due to sleep). When I define the controller outside such as in the code below, the points are not updated and there is a noResponse warning from the controller.
My ultimate aim is to collect data that is changing from multiple BACnet devices in a faster way so that I can collect all data regularly (say every minute). This is also assuming I want to collect all the properties of the BACnet devices. I was just wondering if there was any possible method to refresh or update the points every iteration. If not, is there any other way?
I also understand that the BACnet data is meant to be polled to collect data. I was wondering if polling and retrieving the history was a better way than this? Would it be able to collect many points (In thousands) quickly (within a minute)? And ultimately, can that be integrated into using asyncio? I really appreciate it thank you!

Edit: I have also tried adding point.update_bacnet_properties to try and retrieve the new data but while the new data updates, every iteration there are some warnings for no response from controller and takes much longer.

Edit: Tried it on an actual device with scheduler! Seems to work. Thank you so much!

import asyncio
import time
from datetime import datetime
import BAC0
import bacpypes

bacpypes.settings.settings['route_aware'] = True
network_ip = "192.168.79.6"
split_string = network_ip.rsplit(".", 1)
main_ip = split_string[0]

# Connect to BACnet (move this outside the loop)
bacnet = BAC0.connect(network_ip)

# Perform whois and collect devices
bacnet.whois(global_broadcast=True)
time.sleep(2)  # Reduced sleep time

print("Discovered devices:", bacnet.devices)

# Define the maximum number of concurrent queries
CONCURRENT_QUERIES = 5
semaphore = asyncio.Semaphore(CONCURRENT_QUERIES)

# Create a dictionary to store controllers for each device
controllers = {}

# Function to initialize controllers for each device
def initialize_controllers():
    for device in bacnet.devices:
        device_name = "{}_{}_{}".format(device[0], device[3], device[2])
        controllers[device_name] = BAC0.device(device[2], device[3], bacnet)

# Initialize controllers once
initialize_controllers()

# Function to handle device queries asynchronously with semaphore
async def query_device(device):
    async with semaphore:
        device_name = "{}_{}_{}".format(device[0], device[3], device[2])
        mycontroller = controllers[device_name]
        points = mycontroller.points

        device_info = {
            "DeviceName": device[0],
            "DeviceIP": device[2],
            "DeviceID": device[3],
            "TotalObjectFound": len(points)
        }

        print("==================================")
        print("Timestamp: {}".format(datetime.now()))
        print(device_info)
        print("==================================")

        for point in points:
            properties = point.bacnet_properties
            for k, v in properties.items():
                print("{}: {}".format(k, v))
            print("------------------------------")

        # Introduce a delay between queries to manage network load
        await asyncio.sleep(1)

# Asynchronously query all devices
async def collect_data():
    while True:
        start_time = datetime.now()
        start_timestamp = time.time()
        print("Start Time: {}".format(start_time))

        tasks = [query_device(device) for device in bacnet.devices]
        await asyncio.gather(*tasks)

        end_time = datetime.now()
        end_timestamp = time.time()
        print("End Time: {}".format(end_time))
        print("Total Time Taken: {:.2f} seconds".format(end_timestamp - start_timestamp))

        # Print the devices
        print("All discovered devices:")
        print(bacnet.devices)

        # Wait for 10 seconds before the next iteration
        await asyncio.sleep(10)

# Run the asynchronous main function
asyncio.run(collect_data())

[bbartling]

what version of Python are you using? Python 3.12 and above is all asyncio based including BAC0 but older versions its all sync type code.

[SaiChaitanya1320]

I am using Python 3.12

[bbartling]

if you are using older Python (3.11 and below) where BAC0/bacpypes is synchronous, using asyncio for managing concurrency may not provide the intended benefits. Asynchronous programming is most effective when the underlying operations (like network I/O) can yield control back to the event loop, allowing other tasks to run concurrently. If the BACnet operations are synchronous and blocking, using asyncio will not be effective in improving concurrency or performance.

Could consider using threading or multiprocessing to achieve concurrency using threading for concurrent queries.

Haven't tested this code below but it maybe worth playing around with....I have also have ran into issues where the global BACnet discovery process doesnt exactly grab every BACnet device on the network especially if its older twisted pair communication BACnet MSTP networks where I ended up just hard coding BACnet device address and discovering from a different script.

Also BACnet is UDP protocol based and my high level of understanding it (I am not a protocol wizard)... it is not like TCP based where there is a handshake and multiple checks between client/server devices that data has transferred properly but I think there is Acknowledged BACnet message services. I think people generally using the words "polling" or "discovered" or "offline/online", the words "connected" and "disconnected" is confusing to me even though BAC0 uses it but UDP by nature is a connectionless protocol.

import time
from datetime import datetime
from concurrent.futures import ThreadPoolExecutor, as_completed
import BAC0

# Connect to BACnet
network_ip = "192.168.79.6"
bacnet = BAC0.connect(network_ip)

# Perform whois and collect devices
bacnet.whois(global_broadcast=True)
time.sleep(2)  # Allow time for responses

print("Discovered devices:", bacnet.devices)

# Create a dictionary to store controllers for each device
controllers = {}

# Function to initialize controllers for each device
def initialize_controllers():
    for device in bacnet.devices:
        device_name = f"{device[0]}_{device[3]}_{device[2]}"
        controllers[device_name] = BAC0.device(device[2], device[3], bacnet)

# Initialize controllers once
initialize_controllers()

# Function to handle device queries
def query_device(device):
    try:
        device_name = f"{device[0]}_{device[3]}_{device[2]}"
        mycontroller = controllers[device_name]

        # Manually refresh points by re-querying properties
        points = mycontroller.points
        for point in points:
            point_properties = point.bacnet_properties
            for prop_name, prop_value in point_properties.items():
                try:
                    _ = prop_value  # Accessing the property to refresh it
                except Exception as prop_error:
                    print(f"Error reading property {prop_name} of point {point}: {prop_error}")

        device_info = {
            "DeviceName": device[0],
            "DeviceIP": device[2],
            "DeviceID": device[3],
            "TotalObjectFound": len(points)
        }

        print("==================================")
        print("Timestamp:", datetime.now())
        print(device_info)
        print("==================================")

        for point in points:
            properties = point.bacnet_properties
            for k, v in properties.items():
                print(f"{k}: {v}")
            print("------------------------------")

    except Exception as e:
        print(f"Error querying device {device_name}: {e}")

# Function to collect data concurrently
def collect_data():
    while True:
        start_time = datetime.now()
        start_timestamp = time.time()
        print("Start Time:", start_time)

        with ThreadPoolExecutor(max_workers=5) as executor:
            futures = [executor.submit(query_device, device) for device in bacnet.devices]
            for future in as_completed(futures):
                try:
                    future.result()
                except Exception as e:
                    print(f"Error in future: {e}")

        end_time = datetime.now()
        end_timestamp = time.time()
        print("End Time:", end_time)
        print("Total Time Taken: {:.2f} seconds".format(end_timestamp - start_timestamp))

        # Print the devices
        print("All discovered devices:")
        print(bacnet.devices)

        # Wait for 10 seconds before the next iteration
        time.sleep(10)

# Run the data collection
collect_data()

[SaiChaitanya1320]

Thank you @bbartling! Appreciate it!

[SaiChaitanya1320]

@ChristianTremblay. @bbartling Apologies, just wanted to ask if I could create as many controllers or is there a limit. So in my code below, I am creating a controller for every device, therefore was just wondering if say there are 300 devices, will it be okay? this code works perfectly but I am just worried in case there are more devices.

import sched
import time
from datetime import datetime
import BAC0
import bacpypes
import asyncio

bacpypes.settings.settings['route_aware'] = True
network_ip = "192.168.0.66"
split_string = network_ip.rsplit(".", 1)
main_ip = split_string[0]

# Connect to BACnet (move this outside the loop)
bacnet = BAC0.connect(network_ip)

# Perform whois and collect devices
bacnet.whois(global_broadcast=True)
time.sleep(2)  # Reduced sleep time

print("Discovered devices:", bacnet.devices)

# Define the maximum number of concurrent queries
CONCURRENT_QUERIES = 2
semaphore = asyncio.Semaphore(CONCURRENT_QUERIES)

# Create a dictionary to store controllers for each device
controllers = {}

# Function to initialize controllers for each device
def initialize_controllers():
    for device in bacnet.devices:
        device_name = "{}_{}_{}".format(device[0], device[3], device[2])
        controllers[device_name] = BAC0.device(device[2], device[3], bacnet)

# Initialize controllers once
initialize_controllers()

# Function to handle device queries
async def query_device(device): 
    async with semaphore:
        device_name = "{}_{}_{}".format(device[0], device[3], device[2])
        try:
            mycontroller = controllers[device_name]
            points = mycontroller.points

            device_info = {
                "DeviceName": device[0],
                "DeviceIP": device[2],
                "DeviceID": device[3],
                "TotalObjectFound": len(points)
            }

            print("==================================")
            print("Timestamp: {}".format(datetime.now()))
            print(device_info)
            print("==================================")

            for point in points:
                point.update_bacnet_properties()
                properties = point.bacnet_properties
                for k, v in properties.items():
                    print("{}: {}".format(k, v))
                print("------------------------------")

        except Exception as e:
            print(f"Error querying device {device_name}: {e}")

# Asynchronously query all devices
async def collect_data():
    start_time = datetime.now()
    start_timestamp = time.time()
    print("Start Time: {}".format(start_time))

    tasks = [query_device(device) for device in bacnet.devices]
    await asyncio.gather(*tasks)

    end_time = datetime.now()
    end_timestamp = time.time()
    print("End Time: {}".format(end_time))
    print("Total Time Taken: {:.2f} seconds".format(end_timestamp - start_timestamp))

    # Print the devices
    print("All discovered devices:")
    print(bacnet.devices)

# Function to run collect_data using a scheduler
def run_scheduled(scheduler, interval, action, actionargs=()):
    scheduler.enter(interval, 1, run_scheduled, (scheduler, interval, action, actionargs))
    asyncio.run(action(*actionargs))

# Create a scheduler
scheduler = sched.scheduler(time.time, time.sleep)

# Schedule the first execution
scheduler.enter(0, 1, run_scheduled, (scheduler, 10, collect_data))

# Start the scheduler
scheduler.run()

[ChristianTremblay]

To work with Python 3.12, you need to use the asynchronous version of BAC0 (see async branch)

here is a small example of what you could use to generate the devices discovered

async def initialize_controllers():
    await bacnet._discover()
    lst = []
    for device_instance, details in bacnet.discoveredDevices.items():
        objInst, address, netNumber, vendorId, vendorName = details.values()
        devInst = device_instance.split(',')[1]
        device_name = f"{objInst}_{netNumber}_{address}"
        # add all coroutines to the list
        lst.append(BAC0.device(address, devInst, bacnet))
    await asyncio.gather(*lst)

# Initialize controllers once
asyncio.create_task(initialize_controllers())

then once done

>>> bacnet.registered_devices
[
    FXSNC1 / Connected,
    PCG-4-004_Garage_Poulailler / Connected,
    OEI6000-0B / Connected,
    CVM-2-009 / Connected,
    CGM-7-005 / Connected,
    TEC2647-4-8 / Connected,
    TEC3000-5 / Connected,
    TEC3000-7 / Connected
]
>>>