transmit flowchart: send message to switch lights on (successful on 1cm distance)

transmit/message_generator_block.py

@@ -0,0 +1,26 @@
+import json
+import pathlib
+
+import gnuradio.gr
+import numpy
+
+
+class MessageGeneratorBlock(gnuradio.gr.basic_block):
+    def __init__(self):
+        gnuradio.gr.basic_block.__init__(
+            self, name="Message Generator", in_sig=None, out_sig=[numpy.uint8]
+        )
+        # __file__ is not available
+        self._messages = json.loads(
+            pathlib.Path().resolve().parent.joinpath("messages.json").read_text()
+        )
+
+    def general_work(self, input_items, output_items) -> int:
+        # pylint: disable=unused-argument
+        message = self._messages["ON0"] + [0] * 128
+        if len(output_items[0]) < len(message):
+            print("output buffer too short", len(output_items[0]))
+            output_items[0][:] = 0
+            return len(output_items[0])
+        output_items[0][: len(message)] = message
+        return len(message)

transmit/transmit.grc

@@ -44,47 +44,53 @@ blocks:
     coordinate: [670, 313]
     rotation: 0
     state: enabled
-- name: analog_sig_source_x_0
-  id: analog_sig_source_x
+- name: blocks_throttle_0
+  id: blocks_throttle
   parameters:
     affinity: ''
     alias: ''
-    amp: '1'
     comment: ''
-    freq: '10'
+    ignoretag: 'True'
     maxoutbuf: '0'
     minoutbuf: '0'
-    offset: '0'
-    phase: '0'
-    samp_rate: sample_rate_hz
+    samples_per_second: sample_rate_hz
     type: byte
-    waveform: analog.GR_SQR_WAVE
+    vlen: '1'
   states:
     bus_sink: false
     bus_source: false
     bus_structure: null
-    coordinate: [70, 187]
+    coordinate: [649, 177]
     rotation: 0
-    state: true
-- name: blocks_throttle_0
-  id: blocks_throttle
+    state: bypassed
+- name: message_generator_block
+  id: epy_block
   parameters:
+    _source_code: "import json\nimport pathlib\n\nimport gnuradio.gr\nimport numpy\n\
+      \n\nclass MessageGeneratorBlock(gnuradio.gr.basic_block):\n    def __init__(self):\n\
+      \        gnuradio.gr.basic_block.__init__(\n            self, name=\"Message\
+      \ Generator\", in_sig=None, out_sig=[numpy.uint8]\n        )\n        # __file__\
+      \ is not available\n        self._messages = json.loads(\n            pathlib.Path().resolve().parent.joinpath(\"\
+      messages.json\").read_text()\n        )\n\n    def general_work(self, input_items,\
+      \ output_items) -> int:\n        # pylint: disable=unused-argument\n       \
+      \ message = self._messages[\"ON0\"] + [0] * 128\n        if len(output_items[0])\
+      \ < len(message):\n            print(\"output buffer too short\", len(output_items[0]))\n\
+      \            output_items[0][:] = 0\n            return len(output_items[0])\n\
+      \        output_items[0][: len(message)] = message\n        return len(message)\n"
     affinity: ''
     alias: ''
     comment: ''
-    ignoretag: 'True'
     maxoutbuf: '0'
     minoutbuf: '0'
-    samples_per_second: sample_rate_hz
-    type: byte
-    vlen: '1'
   states:
+    _io_cache: ('Message Generator', 'MessageGeneratorBlock', [], [], [('0', 'byte',
+      1)], '', [])
     bus_sink: false
     bus_source: false
     bus_structure: null
-    coordinate: [649, 177]
+    coordinate: [286, 318]
     rotation: 0
-    state: bypassed
+    state: true
 - name: qtgui_time_raster_sink_x_0
   id: qtgui_time_raster_sink_x
   parameters:
@@ -199,9 +205,9 @@ blocks:
     state: enabled
 
 connections:
-- [analog_sig_source_x_0, '0', blocks_throttle_0, '0']
-- [analog_sig_source_x_0, '0', serial_sink_block, '0']
 - [blocks_throttle_0, '0', qtgui_time_raster_sink_x_0, '0']
+- [message_generator_block, '0', blocks_throttle_0, '0']
+- [message_generator_block, '0', serial_sink_block, '0']
 
 metadata:
   file_format: 1