acquire data from pipe

vna_system/core/acquisition/data_acquisition.py

@@ -1,5 +1,6 @@
 import io
 import json
+import math
 import os
 import struct
 import threading
@@ -192,6 +193,67 @@ class VNADataAcquisition:
     # --------------------------------------------------------------------- #
     # Acquisition loop
     # --------------------------------------------------------------------- #
+    def _radar_pipe_acquisition_loop(self) -> None:
+        """Acquisition loop for reading radar data from named pipe."""
+        logger.info("Starting radar pipe acquisition loop", pipe_path=cfg.RADAR_PIPE_PATH)
+
+        while self._running and not self._stop_event.is_set():
+            try:
+                # Honor pause
+                if self._paused:
+                    time.sleep(0.1)
+                    continue
+
+                # Open named pipe for reading
+                pipe_path = Path(cfg.RADAR_PIPE_PATH)
+
+                # Check if pipe exists
+                if not pipe_path.exists():
+                    logger.warning("Radar pipe not found, waiting...", path=str(pipe_path))
+                    time.sleep(1.0)
+                    continue
+
+                # Open pipe in non-blocking mode to avoid hanging
+                with open(pipe_path, "rb") as pipe:
+                    # Read data from pipe (adjust buffer size as needed)
+                    data = pipe.read(cfg.EXPECTED_POINTS_PER_SWEEP * cfg.RADAR_BYTES_PER_SAMPLE)
+
+                    if not data:
+                        time.sleep(0.01)
+                        continue
+
+                    # Parse radar data
+                    points = self._parse_radar_data(data)
+
+                    if points:
+                        timestamp = time.time()
+                        sweep_number = self._sweep_buffer.add_sweep(points, timestamp=timestamp)
+                        logger.info(
+                            "Radar sweep collected from pipe",
+                            sweep_number=sweep_number,
+                            points=len(points),
+                            timestamp=timestamp
+                        )
+
+                        # # Play sweep notification sound
+                        # self._sound_player.play()
+
+                        # Handle single-sweep mode transitions
+                        if not self._continuous_mode:
+                            if self._single_sweep_requested:
+                                self._single_sweep_requested = False
+                                logger.info("Single radar sweep completed; pausing acquisition")
+                                self.pause()
+                            else:
+                                self.pause()
+
+            except FileNotFoundError:
+                logger.warning("Radar pipe not found, retrying...", pipe_path=cfg.RADAR_PIPE_PATH)
+                time.sleep(1.0)
+            except Exception as exc:  # noqa: BLE001
+                logger.error("Radar pipe acquisition loop error", error=repr(exc))
+                time.sleep(1.0)
+
     def _simulator_acquisition_loop(self) -> None:
         """Simplified acquisition loop for simulator mode."""
         logger.info("Starting simulator acquisition loop")
@@ -236,6 +298,11 @@ class VNADataAcquisition:
 
     def _acquisition_loop(self) -> None:
         """Main acquisition loop executed by the background thread."""
+        # Use radar pipe loop if enabled
+        if cfg.USE_RADAR_PIPE:
+            self._radar_pipe_acquisition_loop()
+            return
+
         # Use simulator loop if simulator is enabled
         if self._simulator is not None:
             self._simulator_acquisition_loop()
@@ -468,6 +535,58 @@ class VNADataAcquisition:
     # --------------------------------------------------------------------- #
     # Parsing & detection
     # --------------------------------------------------------------------- #
+
+    def _parse_radar_data(self, data: bytes) -> list[tuple[float, float]]:
+        """
+        Parse radar data from named pipe format.
+
+        Expected format: 4 bytes per 32-bit word (big-endian)
+        - Word format: 0xF0XXXXXX where:
+          - F0 is the marker byte (bits 31-24)
+          - XXXXXX is the 24-bit data value (bits 23-0)
+
+        Returns list of (real, imag) tuples where:
+        - real: dB value converted from raw data
+        - imag: always 0.0
+        """
+        points: list[tuple[float, float]] = []
+
+        # Process data in 4-byte chunks as 32-bit words (big-endian)
+        num_words = len(data) // cfg.RADAR_BYTES_PER_SAMPLE
+
+        for i in range(num_words):
+                
+            offset = i * cfg.RADAR_BYTES_PER_SAMPLE
+            chunk = data[offset : offset + cfg.RADAR_BYTES_PER_SAMPLE]
+
+            # Unpack as big-endian 32-bit unsigned integer
+            word = struct.unpack("<I", chunk)[0]
+
+            # Extract marker (top 8 bits)
+            marker = (word >> 24) & 0xFF
+
+            # Extract 24-bit data value (lower 24 bits)
+            raw_value = word & 0xFFFFFF
+
+            # Check marker byte - log warning but continue processing
+            if marker != cfg.RADAR_DATA_MARKER:
+                # Only log occasionally to avoid spam
+                if i % 100 == 0:
+                    logger.debug(
+                        "Non-F0 marker detected",
+                        marker=hex(marker),
+                        word=hex(word),
+                        offset=offset
+                    )
+                # Still process the data if it has valid bits
+
+            # Convert to dB if non-zero
+            if raw_value != 0:
+                points.append((float(int(raw_value)), 0.))
+            if i == 0 or i == 100:
+                logger.debug(f"raw_value: {raw_value}, marker: {marker},  word= {word}" )
+        return points
+
     def _parse_measurement_data(self, payload: bytes) -> list[tuple[float, float]]:
         """Parse complex measurement samples (float32 pairs) from a payload."""
         if len(payload) <= cfg.MEAS_HEADER_LEN:

vna_system/core/config.py

@@ -35,10 +35,18 @@ VNA_PID = 0x5740  # STM32 Virtual ComPort
 # -----------------------------------------------------------------------------
 # Simulator mode settings
 # -----------------------------------------------------------------------------
-USE_SIMULATOR = True  # Set to True to use simulator instead of real device
+USE_SIMULATOR = False  # Set to True to use simulator instead of real device
 SIMULATOR_SWEEP_FILE = BASE_DIR / "binary_input" / "sweep_example" / "example.json"
 SIMULATOR_NOISE_LEVEL = 100  # Standard deviation of Gaussian noise to add to real and imaginary parts
 
+# -----------------------------------------------------------------------------
+# Radar pipe mode settings
+# -----------------------------------------------------------------------------
+USE_RADAR_PIPE = True  # Set to True to read radar data from named pipe
+RADAR_PIPE_PATH = "/tmp/radar_data_pipe"  # Path to the named pipe
+RADAR_DATA_MARKER = 0xF0  # First byte marker for radar data packets
+RADAR_BYTES_PER_SAMPLE = 4  # Total bytes per radar sample (1 marker + 3 data)
+
 # -----------------------------------------------------------------------------
 # Sweep detection and parsing constants
 # -----------------------------------------------------------------------------

vna_system/core/laser/laser_controller.py

@@ -483,7 +483,7 @@ class LaserController:
             # Close serial port
             if self.prt is not None:
                 try:
-                    cmd.close_port(self.prt)
+                    # cmd.close_port(self.prt)
                     logger.info("Serial port closed")
                 except Exception as e:
                     logger.warning(f"Error closing serial port: {e}")

vna_system/core/processors/configs/magnitude_config.json

@@ -1,5 +1,5 @@
 {
-  "y_min": -50,
+  "y_min": -65,
   "y_max": 40,
   "autoscale": true,
   "show_magnitude": true,