Added svlog support to omniscan450

Author: BradG13531

examples/omniscan450Example.py

@@ -32,7 +32,7 @@
 
 # Signal handler to stop pinging on the Omniscan450
 def signal_handler(sig, frame):
-    print("Stopping pinging on Omniscan450...")
+    print("\nStopping pinging on Omniscan450...")
     myOmniscan450.control_os_ping_params(enable=0)
     if myOmniscan450.iodev:
         try:
@@ -50,14 +50,21 @@ def signal_handler(sig, frame):
 new_log = False
 log_path = ""
 replay_path = None
+default_dir = Path("logs/omniscan").resolve()
 if args.log is not None:
     if args.log is True:
         # Logging to default directory
-        myOmniscan450 = Omniscan450(logging=True)
-        print(f"Logging to new file: {myOmniscan450.current_log}")
+        default_dir.mkdir(parents=True, exist_ok=True)
+        myOmniscan450 = Omniscan450(logging=True, log_directory=default_dir)
+        print(f"Logging to new file in: {default_dir}")
         new_log = True
     elif isinstance(args.log, str):
-        log_path = Path(args.log).expanduser().resolve()
+        log_path = Path(args.log).expanduser()
+
+        if log_path.suffix == ".svlog" and log_path.parent == Path("."):
+            log_path = default_dir / log_path.name
+
+        log_path = log_path.resolve()
 
         if log_path.suffix == ".svlog":
             if log_path.exists() and log_path.is_file():
@@ -66,7 +73,6 @@ def signal_handler(sig, frame):
                 myOmniscan450 = Omniscan450(logging=False)
                 replay_path = log_path
                 print(f"Replaying from: {replay_path}")
-            # TODO add default path logic
             else:
                 raise FileNotFoundError(f"Log file not found: {log_path}")
 
@@ -78,8 +84,6 @@ def signal_handler(sig, frame):
 
         else:
             raise ValueError(f"Invalid log argument: {args.log}")
-    else:
-        log_filename = None
 
 if args.device is not None:
     myOmniscan450.connect_serial(args.device, args.baudrate)
@@ -95,8 +99,8 @@ def signal_handler(sig, frame):
         print("Failed to initialize Omniscan450!")
         exit(1)
 
-    data1 = myOmniscan450.get_device_information()
-    print("Device type: %s" % data1["device_type"])
+    data1 = myOmniscan450.readDeviceInformation()
+    print("Device type: %s" % data1.device_type)
 
 print("------------------------------------")
 print("Starting Omniscan450..")
@@ -109,43 +113,21 @@ def signal_handler(sig, frame):
 if args.log is not None and not new_log:
     with open(log_path, 'rb') as f:
         while True:
-            start_bytes = f.read(2)
-            if len(start_bytes) < 2:
-                break   # EOF
-
-            if start_bytes != b'BR':
-                print("Invalid start bytes. Skipping...")
-                continue
-
-            header = f.read(6)
-            if len(header) < 6:
-                print("Incomplete header. Ending replay.")
-                break
-
-            payload_length = int.from_bytes(header[0:2], 'little')
-            total_len = 2 + 2 + 2 + 2 + payload_length + 2  # BR + num payload bytes + packet id + reserved + payload + checksum 
-
-            remaining = payload_length + 2
-            rest = f.read(remaining)
+            data = Omniscan450.read_packet(f)
 
-            if len(rest) < remaining:
-                print("Incomplete payload or checksum. Ending replay.")
-            
-            msg_bytes = start_bytes + header + rest
-            data = PingMessage(msg_data=msg_bytes)
+            if data == None:
+                break # EOF or bad packet
 
-            # print(data)
+            if data.message_id == definitions.OMNISCAN450_OS_MONO_PROFILE:
+                # print(data)
 
-            # Printing the same results as if directly connected to the Omniscan
-            scaled_result = Omniscan450.scale_power(data)
-            # for i in range(len(scaled_result)):
-                # print(f"{i+1}: Raw: {data.pwr_results[i]}\tScaled: {scaled_result[i]}dB")
-            print(f"Min power: {data.min_pwr_db} dB")
-            print(f"Max power: {data.max_pwr_db} dB")
-            print(f"Average power: {sum(scaled_result) / len(scaled_result)}")
-
-        raw_bytes = f.read()
-        data = PingMessage(msg_data=raw_bytes)
+                # Printing the same results as if directly connected to the Omniscan
+                scaled_result = Omniscan450.scale_power(data)
+                # for i in range(len(scaled_result)):
+                    # print(f"{i+1}: Raw: {data.pwr_results[i]}\tScaled: {scaled_result[i]}dB")
+                # print(f"Min power: {data.min_pwr_db} dB")
+                # print(f"Max power: {data.max_pwr_db} dB")
+                print(f"Average power: {sum(scaled_result) / len(scaled_result)}")
 
 # Connected to physical omniscan
 else:
@@ -186,18 +168,22 @@ def signal_handler(sig, frame):
     # )
 
     # View power results
-    while True:
-        data = myOmniscan450.wait_message([definitions.OMNISCAN450_OS_MONO_PROFILE])
-        if data:
-            print("Message Recieved")
-            # scaled_result = Omniscan450.scale_power(data)
-            # for i in range(len(scaled_result)):
-            #     print(f"{i+1}: Raw: {data.pwr_results[i]}\tScaled: {scaled_result[i]}dB")
-            # print(f"Min power: {data.min_pwr_db} dB")
-            # print(f"Max power: {data.max_pwr_db} dB")
-            
-        else:
-            print("Failed to get report")
+    if new_log:
+        print("Logging...\nCTRL+C to stop logging")
+    try:
+        while True:
+            data = myOmniscan450.wait_message([definitions.OMNISCAN450_OS_MONO_PROFILE])
+            if data and not new_log:
+                pass
+                scaled_result = Omniscan450.scale_power(data)
+                for i in range(len(scaled_result)):
+                    print(f"{i+1}: Raw: {data.pwr_results[i]}\tScaled: {scaled_result[i]}dB")
+                print(f"Min power: {data.min_pwr_db} dB")
+                print(f"Max power: {data.max_pwr_db} dB")
+            elif not data:
+                print("Failed to get report")
+    except KeyboardInterrupt:
+        print("Stopping logging...")
 
     # Disable pinging and close socket
     myOmniscan450.control_os_ping_params(enable=0)
@@ -206,9 +192,3 @@ def signal_handler(sig, frame):
                 myOmniscan450.iodev.close()
             except Exception as e:
                 print(f"Failed to close socket: {e}")
-
-    if new_log:
-        os.makedirs("logs/omniscan", exist_ok=True)
-        log_path = os.path.join("logs/omniscan", log_filename)
-        with open(log_path, 'ab') as f:
-            f.write(data.msg_data)

generate/templates/omniscan450.py.in

@@ -18,6 +18,12 @@ import socket
 from datetime import datetime
 from pathlib import Path
 
+# Imports for svlog header
+import json
+import os
+import sys
+import platform
+
 MAX_LOG_SIZE_MB = 500
 
 class Omniscan450(PingDevice):
@@ -27,12 +33,14 @@ class Omniscan450(PingDevice):
         self.log_directory = log_directory
         self.bytes_written = None
         self.current_log = None
-        if logging:
-            self.new_log(log_directory)
+        {# if logging:
+            self.new_log(log_directory) #}
 
     def initialize(self):
         if (self.readDeviceInformation() is None):
             return False
+        if self.logging:
+            self.new_log(self.log_directory)
         return True
 
 {% for msg in messages["get"]|sort %}
@@ -127,38 +135,154 @@ class Omniscan450(PingDevice):
         final_power_results = tuple(scaled_power_results)
         return final_power_results
 
+    # Reads a single packet from a file
+    @staticmethod
+    def read_packet(file):
+        sync = file.read(2)
+        if sync != b'BR':
+            return None
+        
+        payload_len_bytes = file.read(2)
+        if len(payload_len_bytes) < 2:
+            return None
+        payload_len = int.from_bytes(payload_len_bytes, 'little')
+
+        msg_id = file.read(2)
+        if len(msg_id) < 2:
+            return None
+
+        remaining = 2 + payload_len + 2
+        rest = file.read(remaining)
+        if len(rest) < remaining:
+            return None
+
+        msg_bytes = sync + payload_len_bytes + msg_id + rest
+        return pingmessage.PingMessage(msg_data=msg_bytes)
+
+    # Builds the packet containing metadata for the beginning of .svlog
+    def build_metadata_packet(self):
+
+        protocol = "tcp" # default fallback
+        if self.iodev:
+            if self.iodev.type == socket.SOCK_STREAM:
+                protocol = "tcp"
+            elif self.iodev.type == socket.SOCK_DGRAM:
+                protocol = "udp"
+        
+        if self.server_address:
+            url = f"{protocol}://{self.server_address[0]}:{self.server_address[1]}"
+        else:
+            url = f"{protocol}://unknown"
+
+        content = {
+            "session_id": 1,
+            "session_uptime": 0.0,
+            "session_devices": [
+                {
+                    "url": url,
+                    "product_id": "os450"
+                }
+            ],
+            "session_platform": None,
+            "session_clients": [],
+            "session_plan_name": None,
+
+            "sonarlink_version": "",
+            "os_hostname": platform.node(),
+            "os_uptime": None,
+            "os_version": platform.version(),
+            "os_platform": platform.system().lower(),
+            "os_release": platform.release(),
+
+            "process_path": sys.executable,
+            "process_version": f"v{platform.python_version()}",
+            "process_uptime": time.process_time(),
+            "process_arch": platform.machine(),
+
+            "timestamp": datetime.utcnow().isoformat() + "Z",
+            "timestamp_timezone_offset": -time.timezone // 60
+        }
+
+        json_bytes = json.dumps(content, indent=2).encode("utf-8")
+
+        m = pingmessage.PingMessage(definitions.OMNISCAN450_JSON_WRAPPER)
+        m.payload = json_bytes
+        m.payload_length = len(json_bytes)
+
+        msg_data = bytearray()
+        msg_data += b"BR"
+        msg_data += m.payload_length.to_bytes(2, "little")
+        msg_data += m.message_id.to_bytes(2, "little")
+        msg_data += m.dst_device_id.to_bytes(1, "little")
+        msg_data += m.src_device_id.to_bytes(1, "little")
+        msg_data += m.payload
+
+        checksum = sum(msg_data) & 0xFFFF
+        msg_data += bytearray(struct.pack(pingmessage.PingMessage.endianess + pingmessage.PingMessage.checksum_format, checksum))
+
+        m.msg_data = msg_data 
+        m.checksum = checksum
+
+        return m
+    
+    # Enable logging 
+    def start_logging(self, new_log = False, log_directory = None):
+        if self.logging:
+            return
+
+        self.logging = True
+
+        if self.current_log is None or new_log:
+            self.new_log(log_directory)
+
+    # Disable logging
+    def stop_logging(self):
+        self.logging = False
+
     # Creates a new log file
     def new_log(self, log_directory=None):
         dt = datetime.now()
         save_name = dt.strftime("%Y-%m-%d-%H-%M")
 
         if log_directory is None:
             project_root = Path.cwd().parent
-            log_directory = project_root / "logs/omniscan"
+            self.log_directory = project_root / "logs/omniscan"
         else:
-            log_directory = Path(log_directory)
+            self.log_directory = Path(log_directory)
 
         log_directory.mkdir(parents=True, exist_ok=True)
 
         log_path = log_directory / f"{save_name}.svlog"
 
         if log_path.exists():
-            raise FileExistsError(f"Log file already exists: {log_path}")
+            log_path.unlink() # delete existing file (program was restarted quickly)
+            {# raise FileExistsError(f"Log file already exists: {log_path}") #}
 
         self.current_log = log_path
         self.bytes_written = 0
 
+        self.write_data(self.build_metadata_packet())
+
     # Write data to .svlog file
     def write_data(self, msg):
-        if not self.logging:
+        if not self.logging or not self.current_log:
             return
 
-        if self.bytes_written > MAX_LOG_SIZE_MB * 1000000:
-            self.new_log(log_directory=self.log_directory)
+        try:
+            if self.bytes_written > MAX_LOG_SIZE_MB * 1000000:
+                self.new_log(log_directory=self.log_directory)
 
-        with open(self.current_log, 'ab') as f:
-            f.write(msg.msg_data)
-            self.bytes_written += len(msg.msg_data)
+            with open(self.current_log, 'ab') as f:
+                f.write(msg.msg_data)
+                self.bytes_written += len(msg.msg_data)
+
+        except (OSError, IOError) as e:
+            print(f"[LOGGING ERROR] Failed to write to log file {self.current_log}: {e}")
+            self.stop_logging()
+
+        except Exception as e:
+            print(f"[LOGGING ERROR] Unexpected error: {e}")
+            self.stop_logging()
 
     # Override wait_message to format power results before returning
     def wait_message(self, message_ids, timeout=0.5):