Add BACnet scan to CLI (#242)

* add dump functionality from records ingress handler

* add cli subcommand for scanning bacnet network

* add bacnet cli test

* add load methods and tests

* make bacnet cli test end to end

* point assert to correct path

* add doc strings

* BACnet ingress free socket after discover

* expose ports in docker compose

* run bacnet tests within a docker container on the same virtual network as the virtual bacnet device

* fix indentation

* cleanup tests

* replace assert with warning

* fix test issues

.github/workflows/ci.yml

@@ -54,6 +54,13 @@ jobs:
         run: poetry run pytest tests/unit --cov=./ --cov-report=xml
       - name: integration tests
         run: poetry run pytest tests/integration
+      - name: bacnet tests
+        run: |
+          cd tests/integration/fixtures/bacnet
+          docker compose build device buildingmotif
+          docker compose run -d device
+          docker compose run buildingmotif poetry run pytest -m bacnet
+          docker compose down
       - name: build tests
         run: poetry build
 

buildingmotif/bin/cli.py

@@ -7,6 +7,7 @@
 
 from buildingmotif import BuildingMOTIF
 from buildingmotif.dataclasses import Library
+from buildingmotif.ingresses.bacnet import BACnetNetwork
 
 cli = argparse.ArgumentParser(
     prog="buildingmotif", description="CLI Interface for common BuildingMOTIF tasks"
@@ -163,6 +164,16 @@ def app():
         args.func(args)
 
 
+@subcommand(
+    arg("-o", "--output_file", help="Output file for BACnet scan", required=True),
+    arg("-ip", help="ip address of BACnet network to scan", default=None),
+)
+def scan(args):
+    """Scans a BACnet network and generates a JSON file for later processing"""
+    bacnet_network = BACnetNetwork(args.ip)
+    bacnet_network.dump(Path(args.output_file))
+
+
 # entrypoint is actually defined in pyproject.toml; this is here for convenience/testing
 if __name__ == "__main__":
     app()

buildingmotif/ingresses/bacnet.py

@@ -1,5 +1,6 @@
 # configure logging output
 import logging
+import warnings
 from functools import cached_property
 from typing import Any, Dict, List, Optional, Tuple
 
@@ -42,17 +43,23 @@ def __init__(self, ip: Optional[str] = None):
         # for each discovered Device, create a BAC0.device object
         # This will read the BACnet objects off of the Device.
         # Save the BACnet objects in the objects dictionary
-        assert self.network.discoveredDevices is not None
-        for (address, device_id) in self.network.discoveredDevices:  # type: ignore
-            # set poll to 0 to avoid reading the points regularly
-            dev = BAC0.device(address, device_id, self.network, poll=0)
-            self.devices.append(dev)
-            self.objects[(address, device_id)] = []
+        try:
+            if self.network.discoveredDevices is None:
+                warnings.warn("BACnet ingress could not find any BACnet devices")
+            for (address, device_id) in self.network.discoveredDevices:  # type: ignore
+                # set poll to 0 to avoid reading the points regularly
+                dev = BAC0.device(address, device_id, self.network, poll=0)
+                self.devices.append(dev)
+                self.objects[(address, device_id)] = []
 
-            for bobj in dev.points:
-                obj = bobj.properties.asdict
-                self._clean_object(obj)
-                self.objects[(address, device_id)].append(obj)
+                for bobj in dev.points:
+                    obj = bobj.properties.asdict
+                    self._clean_object(obj)
+                    self.objects[(address, device_id)].append(obj)
+        finally:
+            for dev in self.devices:
+                self.network.unregister_device(dev)
+            self.network.disconnect()
 
     def _clean_object(self, obj: Dict[str, Any]):
         if "name" in obj:
@@ -83,6 +90,7 @@ def records(self) -> List[Record]:
         for (address, device_id), objs in self.objects.items():
             for obj in objs:
                 fields = obj.copy()
+                del fields["device"]
                 fields["device_id"] = device_id
                 records.append(
                     Record(

buildingmotif/ingresses/base.py

@@ -1,5 +1,8 @@
+import json
 from dataclasses import dataclass
 from functools import cached_property
+from os import PathLike
+from pathlib import Path
 from typing import List
 
 from rdflib import Graph, Namespace
@@ -37,6 +40,46 @@ def records(self) -> List[Record]:
         """
         raise NotImplementedError("Must be overridden by subclass")
 
+    def dump(self, path: PathLike):
+        """
+        Takes the contents of the records of this handler and writes them to a JSON file
+
+        :param path: path to write output file to
+        :type path: PathLike
+        """
+        output_string = self.dumps()
+        output_file = Path(path)
+        with output_file.open("w", encoding="utf-8") as f:
+            f.write(output_string)
+
+    def dumps(self) -> str:
+        """
+        Takes the contents of the records of this handler and writes them to a string
+        """
+        records = [
+            {"rtype": record.rtype, "fields": record.fields} for record in self.records
+        ]
+        return json.dumps(records)
+
+    @classmethod
+    def load(cls, path: PathLike):
+        """
+        Takes a file generated by 'dump' and creates a new ingress handler with those records
+        """
+        return cls.loads(Path(path).read_text())
+
+    @classmethod
+    def loads(cls, s: str):
+        """
+        Takes the string output by 'dumps' and creates a new ingress handler with those records
+        """
+        self = cls.__new__(cls)
+        records = []
+        for record in json.loads(s):
+            records.append(Record(record["rtype"], record["fields"]))
+        self.records = records
+        return self
+
 
 class GraphIngressHandler(IngressHandler):
     """Generates a Graph from an underlying metadata source or RecordIngressHandler"""

pytest.ini

@@ -0,0 +1,6 @@
+[pytest]
+addopts =
+    -m "not bacnet"
+markers =
+    integration: marks tests as integration tests (deselect with '-m "not integration"')
+    bacnet: marks tests to be run with a virtual bacnet network (deselected by default)

tests/integration/fixtures/bacnet/docker-compose.yml

@@ -2,11 +2,19 @@ version: "3.4"
 services:
   device:
     build:
+      context: .
       dockerfile: Dockerfile
     networks:
       bacnet:
         ipv4_address: 172.24.0.3
     command: "python3 virtual_bacnet.py"
+  buildingmotif:
+    build:
+      context: ../../../../
+      dockerfile: tests/integration/fixtures/buildingmotif/Dockerfile
+    networks:
+      bacnet:
+        ipv4_address: 172.24.0.2
 networks:
   bacnet:
     ipam:

tests/integration/fixtures/buildingmotif/Dockerfile

@@ -0,0 +1,18 @@
+FROM python:3.8
+
+WORKDIR /home/buildingmotif
+
+RUN pip install poetry==1.4.0 && poetry config virtualenvs.create false
+
+COPY pyproject.toml .
+COPY poetry.lock .
+
+RUN poetry install --no-root
+
+COPY buildingmotif buildingmotif
+COPY tests tests
+COPY notebooks notebooks
+COPY migrations migrations
+COPY docs docs
+
+RUN poetry install

tests/integration/test_bacnet_ingress.py

@@ -1,6 +1,5 @@
 import shlex
 import subprocess
-from pathlib import Path
 
 import pytest
 from rdflib import Namespace
@@ -9,26 +8,37 @@
 from buildingmotif.ingresses import BACnetNetwork, BACnetToBrickIngress
 from buildingmotif.namespaces import BACNET, BRICK, RDF
 
-# path to docker compose file
-docker_compose_path = Path(__file__).parent / Path("fixtures") / Path("bacnet")
-# command to start docker compose
-docker_compose_start = shlex.split("docker compose up -d --build")
-# command to stop docker compose
-docker_compose_stop = shlex.split("docker compose down")
 
+@pytest.mark.bacnet
+def test_bacnet_ingress(bm):
+    BLDG = Namespace("urn:building/")
+    m = Model.create(BLDG, "test building for bacnet scan")
+    bacnet = BACnetNetwork("172.24.0.2/32")
+    tobrick = BACnetToBrickIngress(bm, bacnet)
+    m.add_graph(tobrick.graph(BLDG))
+
+    devices = list(m.graph.subjects(RDF["type"], BACNET["BACnetDevice"]))
+    assert len(devices) == 1, f"Did not find exactly 1 device (found {len(devices)})"
+    assert devices[0] == BLDG["599"]  # type: ignore
 
-@pytest.fixture()
-def bacnet_network():
-    subprocess.run(docker_compose_start, cwd=docker_compose_path)
-    yield
-    subprocess.run(docker_compose_stop, cwd=docker_compose_path)
+    objects = list(m.graph.subjects(RDF["type"], BRICK["Point"]))
+    assert (
+        len(objects) == 4
+    ), f"Did not find exactly 4 points; found {len(objects)} instead"
 
 
-@pytest.mark.integration
-def test_bacnet_ingress(bm, bacnet_network):
+@pytest.mark.bacnet
+def test_bacnet_scan_cli(bm, tmp_path):
     BLDG = Namespace("urn:building/")
     m = Model.create(BLDG, "test building for bacnet scan")
-    bacnet = BACnetNetwork("172.24.0.1/32")
+    d = tmp_path / "scans"
+    d.mkdir()
+    output_file = d / "output.json"
+    subprocess.run(
+        shlex.split(f'buildingmotif scan -o "{str(output_file)}" -ip 172.24.0.2/32')
+    )
+    assert output_file.exists()
+    bacnet = BACnetNetwork.load(output_file)
     tobrick = BACnetToBrickIngress(bm, bacnet)
     m.add_graph(tobrick.graph(BLDG))
 

tests/unit/test_record_ingress_handler.py

@@ -0,0 +1,21 @@
+from pathlib import Path
+
+from buildingmotif.ingresses.base import Record, RecordIngressHandler
+
+
+def test_ingress_dump_load(bm, tmp_path: Path):
+    records = [
+        Record("a", {"a": 1, "b": 2}),
+        Record("b", {"b": 1, "a": 2}),
+    ]
+
+    output_file = tmp_path / "output.json"
+
+    ingress_handler_1 = RecordIngressHandler.__new__(RecordIngressHandler)
+    ingress_handler_1.records = records
+    ingress_handler_1.dump(output_file)
+
+    ingress_handler_2 = RecordIngressHandler.load(output_file)
+    ingress_records = ingress_handler_2.records
+
+    assert ingress_records == records