Merge pull request #1318 from anarkiwi/zmq

Simplify retune_fft JSON output.

docker/Dockerfile.base

@@ -14,7 +14,7 @@ RUN apt-get update && apt-get install -y --no-install-recommends \
   libvulkan-dev \
   python3-numpy
 WORKDIR /root
-RUN git clone https://github.com/iqtlabs/gr-iqtlabs -b 1.0.108
+RUN git clone https://github.com/iqtlabs/gr-iqtlabs -b 1.0.109
 COPY --from=iqtlabs/gamutrf-vkfft:latest /root /root/gr-iqtlabs
 WORKDIR /root/gr-iqtlabs/build
 COPY --from=iqtlabs/gamutrf-sigmf:latest /usr/local /usr/local

gamutrf/grpduzmq.py

@@ -0,0 +1,43 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+import sys
+import pmt
+import zmq
+
+try:
+    from gnuradio import gr  # pytype: disable=import-error
+except ModuleNotFoundError as err:  # pragma: no cover
+    print(
+        "Run from outside a supported environment, please run via Docker (https://github.com/IQTLabs/gamutRF#readme): %s"
+        % err
+    )
+    sys.exit(1)
+
+DELIM = "\n"
+
+
+class pduzmq(gr.basic_block):
+    def __init__(
+        self,
+        zmq_addr,
+    ):
+        gr.basic_block.__init__(
+            self,
+            name="pduzmq",
+            in_sig=None,
+            out_sig=None,
+        )
+        self.zmq_context = zmq.Context()
+        self.zmq_pub = self.zmq_context.socket(zmq.PUB)
+        self.zmq_pub.setsockopt(zmq.SNDHWM, 100)
+        self.zmq_pub.setsockopt(zmq.SNDBUF, 65536)
+        self.zmq_pub.bind(zmq_addr)
+        self.message_port_register_in(pmt.intern("json"))
+        self.set_msg_handler(pmt.intern("json"), self.receive_pdu)
+
+    def stop(self):
+        self.zmq_pub.close()
+
+    def receive_pdu(self, pdu):
+        item = pmt.to_python(pmt.cdr(pdu)).tobytes().decode("utf8").strip()
+        self.zmq_pub.send_string(item + DELIM, flags=zmq.NOBLOCK)

gamutrf/grscan.py

@@ -25,6 +25,7 @@
 
 from gamutrf.grsource import get_source
 from gamutrf.grinferenceoutput import inferenceoutput
+from gamutrf.grpduzmq import pduzmq
 from gamutrf.utils import endianstr
 
 
@@ -252,10 +253,8 @@ def __init__(
         )
         self.fft_blocks.append(retune_fft)
         fft_zmq_addr = f"tcp://{logaddr}:{logport}"
+        self.pduzmq_block = pduzmq(fft_zmq_addr)
         logging.info("serving FFT on %s", fft_zmq_addr)
-        self.fft_blocks.append(
-            (zeromq.pub_sink(1, 1, fft_zmq_addr, 100, False, 65536, ""))
-        )
 
         if fgaas_port:
             fgaas_zmq_addr = f"tcp://{fgaas_addr}:{fgaas_port}"
@@ -390,13 +389,14 @@ def __init__(
 
         if not retune_fft_output_block:
             retune_fft_output_block = blocks.null_sink(gr.sizeof_float * nfft)
-        self.connect((retune_fft, 1), (retune_fft_output_block, 0))
+        self.connect((retune_fft, 0), (retune_fft_output_block, 0))
 
         if pretune:
             self.msg_connect((self.retune_pre_fft, "tune"), (self.sources[0], cmd_port))
             self.msg_connect((self.retune_pre_fft, "tune"), (retune_fft, "cmd"))
         else:
             self.msg_connect((retune_fft, "tune"), (self.sources[0], cmd_port))
+        self.msg_connect((retune_fft, "json"), (self.pduzmq_block, "json"))
         self.connect_blocks(self.sources[0], self.sources[1:])
 
         self.connect_blocks(self.sources[-1], self.fft_blocks)