A Prometheus exporter for BACnet device properties using the BAC0 library.

bac0-exporter.py

#!/usr/bin/python3 -u

import argparse
argument_parser = argparse.ArgumentParser('Run the HVAC BAC0 server')
argument_parser.add_argument('--local-address', help='local IP address to bind to', required=True)
argument_parser.add_argument('--local-port', help='UDP port to bind to')
argument_parser.add_argument('--bbmd-address', help='IP:port of the BACnet BBMD to register against, if any')
argument_parser.add_argument('--bbmd-ttl', help='TTL of the BACnet foreign device registration, in seconds', type=int)
argument_parser.add_argument('--bacnet-network', help='BACnet network to discover devices on; can be specified multiple times', type=int, action='append', required=True)
argument_parser.add_argument('--prometheus-port', help='Serve Prometheus metrics on this port', type=int)
#argument_parser.add_argument('--trend', help='Name of a point to trend in the form "DeviceName/PropertyName"; can be specified multiple times', action='append')
args = argument_parser.parse_args()

import BAC0
import select
import sys

#import bacpypes.consolelogging
#import logging
# Log low-level BACnet messages
#for loggerName in logging.Logger.manager.loggerDict.keys():
#       bacpypes.consolelogging.ConsoleLogHandler(loggerName)

# Log BAC0 debug messages
#BAC0.log_level(log_file='critical', stderr='debug', stdout='critical')
# Log error information for HTTP 500s
#bacpypes.consolelogging.ConsoleLogHandler('tornado.application')
# Debug the plot renderer
#bacpypes.consolelogging.ConsoleLogHandler('BAC0_Root.BAC0.web.BokehRenderer.DynamicPlotHandler')

flatten = lambda l: [item for sublist in l for item in sublist]

# We use the lite version because there is no point in spinning up the HTTP server - the web app (Flask/Bokeh) is a giant mess that has localhost restrictions hardcoded and is virtually impossible to proxy in a reasonable way.
bacnet = BAC0.lite(ip=args.local_address, port=args.local_port, bbmdAddress=args.bbmd_address, bbmdTTL=args.bbmd_ttl)

# We provide a history_size otherwise BAC0 will keep history forever, leading to unbounded memory usage and performance degradation.
devices = [BAC0.device(device[0], device[1], bacnet, history_size=10, poll=30) for device in bacnet.discover(networks=args.bacnet_network)]

if not devices:
        raise Exception('Could not find any devices')

#trend_points = filter(lambda point: '%s/%s' % (point.properties.device.properties.name, point.properties.name) in args.trend, flatten([device.points for device in devices]))
#for trend_point in trend_points:
#       print('Trending "%s" on device "%s"' % (trend_point.properties.name, trend_point.properties.device.properties.name), file=sys.stderr)
#       bacnet.add_trend(trend_point)

if args.prometheus_port is not None:
        import prometheus_client

        class PrometheusCollector:
                def collect(self):
                        metrics = {}
                        for point in flatten([device.points for device in devices]):
                                def add_to_metrics(unit, value, documentation):
                                        # Sometimes BAC0 sets a property value to ''. No idea why.
                                        try:
                                                value = float(value)
                                        except ValueError:
                                                return
                                        if unit is None:
                                                unit = 'noUnit'
                                        if unit not in metrics:
                                                metrics[unit] = prometheus_client.metrics_core.GaugeMetricFamily('bacnet_' + unit, documentation, labels=['device', 'name'], unit=unit)
                                        metrics[unit].add_metric([point.properties.device.properties.name, point.properties.name], value)
                                if isinstance(point, BAC0.core.devices.Points.NumericPoint):
                                        add_to_metrics(point.properties.units_state, point.lastValue, 'Present Value of BACnet objects with a %s unit' % point.properties.units_state)
                                if isinstance(point, BAC0.core.devices.Points.BooleanPoint):
                                        add_to_metrics('binary', 1 if point.lastValue == 'active' else 0, 'Present Value of BACnet binary objects')
                                if isinstance(point, BAC0.core.devices.Points.EnumPoint):
                                        add_to_metrics('multistate', str(point.lastValue).split(':', 1)[0], 'Present Value of BACnet multi-state objects')

                        return metrics.values()

        prometheus_client.REGISTRY.register(PrometheusCollector())

        prometheus_client.start_http_server(args.prometheus_port)

print("BAC0 exporter ready", file=sys.stderr)

while True:
        select.select([], [], [])

The basic idea is to set up Prometheus to scrape and store metrics from the bac0 exporter, and then set up Grafana and point it to the Prometheus instance. You can set up Prometheus first, use its basic query & graphing features (it has a rudimentary web interface) to verify it works, and then set up Grafana to get nice user-facing graphs and dashboards. Prometheus and Grafana are well-documented, but they are not trivial pieces of software; they require some time investment to set up and to understand how they work and what they offer.

Does this exporter work on Windows?
Please let me know if it works correctly.

OS: Win11 Pro
Python: 3.10.5
BAC0:21.12.3

but
OSError:[WinError 10022] comes back

.

@b169d127 I use it on Linux. I have no idea if it can work on Windows. Looking at the specific error you have, I get the impression Windows doesn't like select() calls with no FDs. You might be able to get past the problem by changing the select.select([], [], []) call on the last line with anything else that sleeps forever, for example:

while True:
  time.sleep(86400)