Changed logging feature on omniscan450.py to log .svlog files

Author: BradG13531

examples/omniscan450Example.py

@@ -12,6 +12,7 @@
 import signal
 import sys
 import os
+from pathlib import Path
 from datetime import datetime
 
 ##Parse Command line options
@@ -23,7 +24,7 @@
 parser.add_argument('--udp', action="store", required=False, type=str, help="Omniscan IP:Port. E.g: 192.168.2.92:51200")
 parser.add_argument('--tcp', action="store", required=False, type=str, help="Omniscan IP:Port. E.g: 192.168.2.92:51200")
 parser.add_argument('--range', action="store", required=False, type=str, help="Set range. E.g: 5000 or 0:5000")
-parser.add_argument('--log', action="store", nargs='?', const=True, type=str, help="Log filename. Will log if it doesn't exist, or replay if it does. Blank creates new log.")
+parser.add_argument('--log', action="store", nargs='?', const=True, type=str, help="Log filename and/or directory path. Will create new log if blank or directory is specified. Will replay if file is specified and exists.")
 args = parser.parse_args()
 if args.device is None and args.udp is None and args.tcp is None and args.log is None:
     parser.print_help()
@@ -42,37 +43,52 @@ def signal_handler(sig, frame):
 
 signal.signal(signal.SIGINT, signal_handler)
 
-# Make a new Omniscan450
-myOmniscan450 = Omniscan450()
+# Check for log argument and make new Omniscan450
+# If no .svlog is specified, create one using default directory
+# If directory specified, .svlog be created in specified directory
+# If a .svlog is specified, existing log will be opened
+new_log = False
+log_path = ""
+replay_path = None
+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}")
+        new_log = True
+    elif isinstance(args.log, str):
+        log_path = Path(args.log).expanduser().resolve()
+
+        if log_path.suffix == ".svlog":
+            if log_path.exists() and log_path.is_file():
+                # File exists, replaying
+                new_log = False
+                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}")
+
+        elif log_path.is_dir() or log_path.suffix == "":
+            # Path is directory, logging to that directory
+            myOmniscan450 = Omniscan450(logging=True, log_directory=log_path)
+            print(f"Logging to new file: {myOmniscan450.current_log}")
+            new_log = True
+        
+        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)
 elif args.udp is not None:
     (host, port) = args.udp.split(':')
     myOmniscan450.connect_udp(host, int(port))
 elif args.tcp is not None:
     (host, port) = args.tcp.split(':')
-    myOmniscan450.connect_tcp(host, int(port)) 
-
-# Check for log argument
-# If no log is specified, create one using date and time
-# If a log is specified, existing log will be opened
-new_log = False
-if args.log is not None:
-    if args.log is True:
-        log_filename = datetime.now().strftime("omniscan-%Y-%m-%d_%H-%M-%S.txt")
-    elif isinstance(args.log, str):
-        log_filename = args.log
-    else:
-        log_filename = None
-    
-    if log_filename:
-        log_path = os.path.join("logs/omniscan", log_filename)
-        if os.path.exists(log_path):
-            print(f"Replaying from existing log file: {log_filename}")
-            new_log = False
-        else:
-            print(f"Logging to new file: {log_filename}")
-            new_log = True
+    myOmniscan450.connect_tcp(host, int(port))    
 
 if args.log is None or new_log:
     if myOmniscan450.initialize() is False:
@@ -91,24 +107,46 @@ def signal_handler(sig, frame):
 
 # Running Omniscan from existing log file
 if args.log is not None and not new_log:
-    os.makedirs("logs/omniscan", exist_ok=True)
-    log_path = os.path.join("logs/omniscan", log_filename)
     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)
+
+            if len(rest) < remaining:
+                print("Incomplete payload or checksum. Ending replay.")
+            
+            msg_bytes = start_bytes + header + rest
+            data = PingMessage(msg_data=msg_bytes)
+
+            # 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)
 
-    if data:
-        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")
-    else:
-        print("Failed to get report")
-
 # Connected to physical omniscan
 else:
     if args.range is not None:
@@ -148,15 +186,18 @@ def signal_handler(sig, frame):
     # )
 
     # View power results
-    data = myOmniscan450.wait_message([definitions.OMNISCAN450_OS_MONO_PROFILE])
-    if data:
-        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")
+    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")
 
     # Disable pinging and close socket
     myOmniscan450.control_os_ping_params(enable=0)

generate/templates/omniscan450.py.in

@@ -15,8 +15,21 @@ import time
 import math
 import struct
 import socket
+from datetime import datetime
+from pathlib import Path
+
+MAX_LOG_SIZE_MB = 500
 
 class Omniscan450(PingDevice):
+    def __init__(self, logging = False, log_directory = None):
+        super().__init__()
+        self.logging = logging
+        self.log_directory = log_directory
+        self.bytes_written = None
+        self.current_log = None
+        if logging:
+            self.new_log(log_directory)
+
     def initialize(self):
         if (self.readDeviceInformation() is None):
             return False
@@ -114,6 +127,39 @@ class Omniscan450(PingDevice):
         final_power_results = tuple(scaled_power_results)
         return final_power_results
 
+    # 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"
+        else:
+            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}")
+
+        self.current_log = log_path
+        self.bytes_written = 0
+
+    # Write data to .svlog file
+    def write_data(self, msg):
+        if not self.logging:
+            return
+        
+        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)
+
     # Override wait_message to format power results before returning
     def wait_message(self, message_ids, timeout=0.5):
         tstart = time.time()
@@ -126,6 +172,8 @@ class Omniscan450(PingDevice):
                     msg.pwr_results = power_results
 
                 if msg.message_id in message_ids:
+                    if self.logging:
+                        self.write_data(msg)
                     return msg
             time.sleep(0.005)
         return None