Method run

examples/NI DAQmx/daqbridge.py:271–305

Main acquisition loop.

(self)

        self._loop_times = deque(maxlen=1000)

    def run(self) -> None:
        """Main acquisition loop."""
        self.running = True
        signal.signal(signal.SIGINT, self._signal_handler)
        signal.signal(signal.SIGTERM, self._signal_handler)

        print("=" * 50)
        print("DAQBridge - NI DAQmx")
        print("=" * 50)
        print(f"Device: {DEVICE_NAME}")
        print(f"Sample Rate: {SAMPLE_RATE} Hz | Samples/Read: {SAMPLES_PER_READ}")
        print(f"UDP: {UDP_HOST}:{UDP_PORT}")
        print(f"Channels: {self.num_channels}")
        print("-" * 50)

        try:
            self.daq.setup()
            print("Press Ctrl+C to stop\n")

            while self.running:
                loop_start = time.perf_counter()

                raw_data = self.daq.read()
                processed = process_readings(raw_data, CHANNEL_CONFIG)
                self.publisher.publish(processed)

                self._loop_times.append(time.perf_counter() - loop_start)
                if ENABLE_STATS:
                    self._print_stats()

        except Exception as e:
            print(f"\nError: {e}")
            raise
        finally:
            self._cleanup()

    def _signal_handler(self, signum, frame) -> None:
        print("\nShutting down...")