BACnet refactor

framework/python/src/api/api.py

@@ -460,13 +460,21 @@ async def get_reports(self):
 
   async def delete_report(self, response: Response, report_name: str):
 
-    device_with_report = self._session.get_report(report_name)
-    if device_with_report.device is None or device_with_report.report is None:
+    mac = report_name.split("_")[0]
+    device = self._session.get_device_by_mac_addr(mac)
+
+    # If the device not found
+    if device is None:
+      LOGGER.info("Device not found, returning 404")
+      response.status_code = 404
+      return self._generate_msg(False, "Device not found")
+
+    report = device.get_report_by_folder_name(report_name)
+    LOGGER.debug(f"Looking for report with name {report_name}")
+    if not report:
       LOGGER.info("Report could not be found, returning 404")
       response.status_code = 404
-      return self._generate_msg(False, "Report not found from list")
-    device = device_with_report.device
-    report = device_with_report.report
+      return self._generate_msg(False, "Report not found")
 
     if self._testrun.delete_report(device, report):
       return self._generate_msg(True, "Deleted report")
@@ -672,13 +680,22 @@ async def edit_device(self, request: Request, response: Response):
 
   async def get_report(self, response: Response, report_name):
     """Serve report pdf file for a given report name"""
-    device_with_report = self._session.get_report(report_name)
-    if device_with_report.device is None or device_with_report.report is None:
+    mac = report_name.split("_")[0]
+    device = self._session.get_device_by_mac_addr(mac)
+
+    # If the device not found
+    if device is None:
+      LOGGER.info("Device not found, returning 404")
+      response.status_code = 404
+      return self._generate_msg(False, "Device not found")
+
+    report = device.get_report_by_folder_name(report_name)
+    LOGGER.debug(f"Looking for report with name {report_name}")
+    if not report:
       LOGGER.info("Report could not be found, returning 404")
       response.status_code = 404
-      return self._generate_msg(False, "Report not found from list")
-    device = device_with_report.device
-    report = device_with_report.report
+      return self._generate_msg(False, "Report could not be found")
+
 
     # Regenerate the pdf if the device profile has been updated
     test_orc = self._get_testrun().get_test_orc()
@@ -721,13 +738,18 @@ async def get_results(
       pass
 
     # Check if device exists
-    device_with_report = self._session.get_report(report_name)
-    if device_with_report.device is None or device_with_report.report is None:
+    mac = report_name.split("_")[0]
+    device = self._session.get_device_by_mac_addr(mac)
+    if device is None:
+      response.status_code = status.HTTP_404_NOT_FOUND
+      return self._generate_msg(False,
+                                "Device not found")
+    report = device.get_report_by_folder_name(report_name)
+    LOGGER.debug(f"Looking for report with name {report_name}")
+    if not report:
       LOGGER.info("Report could not be found, returning 404")
       response.status_code = 404
-      return self._generate_msg(False, "Report not found from list")
-    device = device_with_report.device
-    report = device_with_report.report
+      return self._generate_msg(False, "Report could not be found")
 
     zip_file_path = self._get_testrun().get_test_orc().zip_results(
         device, report, profile)

framework/python/src/common/device.py

@@ -149,9 +149,3 @@ def __setattr__(self, name: str, value: any) -> None:
       # Update the last_updated timestamp
       super().__setattr__('modified_at', datetime.now())
     super().__setattr__(name, value)
-
-
-@dataclass
-class DeviceWithReport():
-  device: Device | None = None
-  report: TestReport | None = None

framework/python/src/common/testreport.py

@@ -269,10 +269,6 @@ def from_json(self, json_file):
 
       self.add_test(test_case)
 
-  def to_json_updated(self, device):
-    self.update_device_info(device)
-    return self.to_json()
-
   # Create a pdf file in memory and return the bytes
   def to_pdf(self):
     # Resolve the data as html first

framework/python/src/core/session.py

@@ -21,7 +21,7 @@
 from common import util, logger, mqtt
 from common.risk_profile import RiskProfile
 from common.statuses import TestrunStatus, TestResult, TestrunResult
-from common.device import Device, DeviceWithReport
+from common.device import Device
 from net_orc.ip_control import IPControl
 
 # Certificate dependencies
@@ -393,16 +393,6 @@ def get_device_by_mac_addr(self, mac_addr_simmplified: str) -> Device | None:
   def get_device_repository(self):
     return self._device_repository
 
-  def get_report(self, folder_name: str) -> DeviceWithReport:
-    device_with_report = DeviceWithReport()
-    for device in self._device_repository:
-      device_reports = device.get_reports()
-      for report in device_reports:
-        if report.get_folder_name() == folder_name:
-          device_with_report.device = device
-          device_with_report.report = report
-    return device_with_report
-
   def add_device(self, device):
     self._device_repository.append(device)
 
@@ -535,10 +525,7 @@ def get_all_reports(self):
     for device in self.get_device_repository():
       device_reports = device.get_reports()
       reports.extend(
-        [
-          device_report.to_json_updated(device)
-          for device_report in device_reports
-        ]
+        [device_report.to_json() for device_report in device_reports]
       )
     return sorted(reports, key=lambda report: report['started'], reverse=True)
 

framework/python/src/test_orc/test_orchestrator.py

@@ -396,19 +396,18 @@ def regenerate_pdf(self, device: Device, report: TestReport) -> str:
     return self._regenerate_report_files(device, report)
 
   def _regenerate_report_files(self, device: Device, report: TestReport) -> str:
-    """Regenerate the report if the device profile has been updated"""
+    '''Regenerate the report if the device profile has been updated'''
     # Report files path
     report_dir = os.path.join(self._root_path, REPORTS_FOLDER)
     report_dir = os.path.join(report_dir, report.get_folder_name())
-    test_folder_name = report.get_folder_name().split("_")[0]
     test_path = os.path.join(
       report_dir,
-      f"test/{test_folder_name}"
+      f'test/{device.mac_addr.replace(":", "")}'
     )
     try:
       # Copy the original report for comparison
       report_copy = copy.deepcopy(report)
-      # Update the report with additional_info field
+      # Update the report with 'additional_info' field
       report.update_device_info(device)
       device.export_config_json()
       # Overwrite report only if additional_info has been changed
@@ -487,13 +486,17 @@ def _update_html_report(self, report: TestReport, html: str):
         if value_div:
           if "Manufacturer" in h4.string:
             value_div.string = manufacturer
+            LOGGER.debug(f"Updated manufacturer to '{value_div.string}'")
           elif "Model" in h4.string:
             value_div.string = model
+            LOGGER.debug(f"Updated model to '{value_div.string}'")
     all_header_info_divs = bs.find_all("div", class_="header-info")
     for header_info_div in all_header_info_divs:
       header_span = header_info_div.find_next_sibling("span")
       if header_span:
         header_span.string = title
+        LOGGER.debug(f"Updated sibling span to '{header_span.string}'")
+
     return str(bs)
 
   def test_in_progress(self):

make/DEBIAN/control

@@ -1,5 +1,5 @@
 Package: Testrun
-Version: 2.3.4-beta.4
+Version: 2.4.0-beta.2
 Architecture: amd64
 Maintainer: Google <ssm-orcas@google.com>
 Homepage: https://github.com/google/testrun

modules/test/protocol/bin/get_bacnet_i-am_packets.sh

@@ -0,0 +1,27 @@
+#!/bin/bash
+
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+#!/bin/bash
+
+CAPTURE_FILE="$1"
+SRC_IP="$2"
+
+TSHARK_FILTER="bacnet && ip.src == $SRC_IP"
+TSHARK_OUTPUT="-T json -e ip.src -e ip.dst -e eth.src -e eth.dst -e bacapp.instance_number -e _ws.col.Info"
+
+response=$(tshark -r "$CAPTURE_FILE" $TSHARK_OUTPUT -Y "$TSHARK_FILTER")
+
+echo "$response"

modules/test/protocol/python/src/protocol_bacnet.py

@@ -14,6 +14,7 @@
 """Module to run all the BACnet related methods for testing"""
 
 import BAC0
+from bacpypes3.pdu import Address
 from dataclasses import dataclass
 import logging
 import json
@@ -64,29 +65,65 @@ def __init__(self,
     self.device_hw_addr = device_hw_addr
     self._bin_dir = bin_dir
 
-  async def discover(self, local_ip):
+  async def discover(self, local_ip, device_ip):
     LOGGER.info('Performing BACnet discovery...')
-    self.bacnet = BAC0.lite(local_ip)
+    self.devices = []
+    self.bacnet = BAC0.connect(local_ip)
     LOGGER.info('Local BACnet object: ' + str(self.bacnet))
     try:
-      await self.bacnet._discover(global_broadcast=True) # pylint: disable=protected-access
+      await self.bacnet._discover(global_broadcast=True, timeout=10) # pylint: disable=protected-access
     except Exception as e: # pylint: disable=W0718
       LOGGER.error(e)
     LOGGER.info('BACnet discovery complete')
     with open(BAC0_LOG, 'r', encoding='utf-8') as f:
       bac0_log = f.read()
     LOGGER.info('BAC0 Log:\n' + bac0_log)
     # Extract discovered devices as a BACnetDevice.
-    self.devices = []
+    LOGGER.info('discoveredDevices: ' + str(self.bacnet.discoveredDevices))
     if self.bacnet.discoveredDevices is not None:
       for device_info in self.bacnet.discoveredDevices.values():
-        self.devices.append(
-          BACnetDevice(
+        device = BACnetDevice(
               device_id=str(device_info['object_instance'][1]),
               ip=str(device_info['address'])
             )
-          )
+        LOGGER.info(f'Discovered BACnet device: {device}')
+        self.devices.append(device)
     LOGGER.info('BACnet devices found: ' + str(len(self.devices)))
+    if not self.devices:
+      try:
+        await self.bacnet._discover(timeout=10) # pylint: disable=protected-access
+      except Exception as e: # pylint: disable=W0718
+        LOGGER.error(e)
+      LOGGER.info('BACnet discovery complete')
+      with open(BAC0_LOG, 'r', encoding='utf-8') as f:
+        bac0_log = f.read()
+      LOGGER.info('BAC0 Log:\n' + bac0_log)
+      LOGGER.info('discoveredDevices: ' + str(self.bacnet.discoveredDevices))
+      if self.bacnet.discoveredDevices is not None:
+        for device_info in self.bacnet.discoveredDevices.values():
+          device = BACnetDevice(
+                device_id=str(device_info['object_instance'][1]),
+                ip=str(device_info['address'])
+              )
+          self.devices.append(device)
+          LOGGER.info(f'Discovered BACnet device: {device}')
+    if not self.devices:
+      res = await self.bacnet.this_application.app.who_is(
+          low_limit=0,
+          high_limit=4194303,
+          address=Address(f'{device_ip}:47808'),
+          timeout=10,
+      )
+      for iam in res:
+        instance = iam.iAmDeviceIdentifier[1]
+        address = str(iam.pduSource)
+        device = BACnetDevice(
+                device_id=str(instance),
+                ip=str(address)
+              )
+        self.devices.append(device)
+    if not self.devices:
+      self.devices = self._discover_from_packets(device_ip)
 
   # Check if the device being tested is in the discovered devices list
   # discover needs to be called before this method is invoked
@@ -194,3 +231,27 @@ def get_bacnet_packets(
     command = f'{bin_file} {args}'
     response = util.run_command(command)
     return json.loads(response[0].strip())
+
+  def _discover_from_packets(self, device_ip: str) -> list[BACnetDevice]:
+    discovered = set()
+    capture_file = os.path.join(self._captures_dir, self._capture_file)
+    LOGGER.info(f'Discovering BACnet devices from packets in {capture_file}...')
+    bin_file = self._bin_dir + '/get_bacnet_i-am_packets.sh'
+    args = f'"{capture_file}" {device_ip}'
+    command = f'{bin_file} {args}'
+    response = util.run_command(command)
+    packets = json.loads(response[0].strip())
+    for packet in packets:
+      info = packet['_source']['layers']['_ws.col.info'][0]
+      if 'i-Am' in info:
+        discovered.add(
+          (
+            packet['_source']['layers']['bacapp.instance_number'][0],
+            packet['_source']['layers']['ip.src'][0]
+           )
+          )
+    LOGGER.info(f'Discovered BACnet devices from packets: {discovered}')
+    return [
+      BACnetDevice(device_id=device_id, ip=ip)
+      for device_id, ip in discovered
+      ]

modules/test/protocol/python/src/protocol_module.py

@@ -53,8 +53,10 @@ def _protocol_valid_bacnet(self):
     local_address = self.get_local_ip(interface_name)
     if local_address:
       local_address += '/24'
-      self._get_bacnet_loop().run_until_complete(
-          self._bacnet.discover(local_address))
+      loop = self._get_bacnet_loop()
+      loop.run_until_complete(
+        self._bacnet.discover(local_address, self._device_ipv4_addr)
+      )
       result = self._bacnet.validate_device()
       if result[0]:
         self._supports_bacnet = True