Merge pull request #1387 from anarkiwi/hang

Fix refactoring

gamutrf/grpduzmq.py

@@ -2,6 +2,7 @@
 # -*- coding: utf-8 -*-
 import logging
 import sys
+import time
 import pmt
 import zmq
 import zstandard
@@ -40,6 +41,8 @@ def __init__(
         self.message_port_register_in(pmt.intern("json"))
         self.set_msg_handler(pmt.intern("json"), self.receive_pdu)
         self.context = zstandard.ZstdCompressor()
+        self.last_log = None
+        self.item_counter = 0
 
     def stop(self):
         self.zmq_pub.close()
@@ -52,3 +55,8 @@ def receive_pdu(self, pdu):
             self.zmq_pub.send(data, flags=zmq.NOBLOCK)
         except zmq.ZMQError as e:
             logging.error(str(e))
+        now = time.time()
+        self.item_counter += 1
+        if self.last_log is None or now - self.last_log > 10:
+            logging.info("sent %u FFT updates", self.item_counter)
+            self.last_log = now

gamutrf/grscan.py

@@ -347,7 +347,7 @@ def __init__(
             if self.iq_inference_block:
                 iq_inference_blocks = [self.iq_inference_block]
                 if iq_inference_squelch_db is not None:
-                    self.iq_inference_block = (
+                    iq_inference_blocks = (
                         self.wrap_batch(
                             [
                                 analog.pwr_squelch_cc(

gamutrf/scan.py

@@ -546,6 +546,10 @@ def check_options(options):
     if iq_inference and not options.pretune:
         return "I/Q inference requires pretune"
 
+    dc_block = options.dc_block_len or options.correct_iq
+    if dc_block and not options.pretune:
+        return "DC blocking requires pretune"
+
     return ""
 
 

gamutrflib/gamutrflib/zmqbucket.py

@@ -84,14 +84,27 @@ def fft_proxy(
     compress_context = zstandard.ZstdCompressor()
     decompress_context = zstandard.ZstdDecompressor()
     shutdown = False
+    last_log_time = None
+    last_data_time = None
     while not shutdown:
         with open(tmp_buff_file, "wb") as zbf:
             with compress_context.stream_writer(zbf) as bf:
                 while not shutdown:
                     shutdown = live_file is not None and not live_file.exists()
+                    now = time.time()
                     try:
                         sock_txt = socket.recv(flags=zmq.NOBLOCK)
                     except zmq.error.Again:
+                        if last_log_time is None or now - last_log_time > 10:
+                            if last_data_time is None:
+                                logging.warning("no data yet from %s", zmq_addr)
+                            else:
+                                logging.warning(
+                                    "no data from %s for %u seconds",
+                                    zmq_addr,
+                                    now - last_data_time,
+                                )
+                            last_log_time = now
                         time.sleep(poll_timeout)
                         continue
                     # gamutrf might send compressed message