Skip to content

Instantly share code, notes, and snippets.

@dechamps
Last active April 19, 2024 02:36
Show Gist options
  • Save dechamps/f1e5b181501f811ec7b679ccfd63a754 to your computer and use it in GitHub Desktop.
Save dechamps/f1e5b181501f811ec7b679ccfd63a754 to your computer and use it in GitHub Desktop.
A Prometheus exporter for BACnet device properties using the BAC0 library.
#!/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([], [], [])
@dechamps
Copy link
Author

@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)

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment