aboutsummaryrefslogtreecommitdiffstats
path: root/src
diff options
context:
space:
mode:
authorNao Pross <np@0hm.ch>2021-12-02 12:54:32 +0100
committerNao Pross <np@0hm.ch>2021-12-02 12:56:38 +0100
commitf7b154f2485b2d897db449890684928e12b1734e (patch)
tree64cee402dfe47d9dea15d24ed1b462a7c5ed118f /src
parentDelete ZMQ tests (diff)
downloadFading-f7b154f2485b2d897db449890684928e12b1734e.tar.gz
Fading-f7b154f2485b2d897db449890684928e12b1734e.zip
Delete XOR Correlator (replaced with by phase correction)
Diffstat (limited to '')
-rw-r--r--src/gr-fadingui/grc/CMakeLists.txt1
-rw-r--r--src/gr-fadingui/grc/fadingui_xor_frame_sync.block.yml42
-rw-r--r--src/gr-fadingui/python/CMakeLists.txt2
-rw-r--r--src/gr-fadingui/python/__init__.py2
-rw-r--r--src/gr-fadingui/python/qa_xor_frame_sync.py57
-rw-r--r--src/gr-fadingui/python/xor_frame_sync.py152
6 files changed, 1 insertions, 255 deletions
diff --git a/src/gr-fadingui/grc/CMakeLists.txt b/src/gr-fadingui/grc/CMakeLists.txt
index 2394de4..d5f23a6 100644
--- a/src/gr-fadingui/grc/CMakeLists.txt
+++ b/src/gr-fadingui/grc/CMakeLists.txt
@@ -20,7 +20,6 @@
install(FILES
fadingui_datasource.block.yml
fadingui_dearpygui_sink.block.yml
- fadingui_xor_frame_sync.block.yml
fadingui_deframer.block.yml
fadingui_frame_obj.block.yml
fadingui_multipath_fading.block.yml DESTINATION share/gnuradio/grc/blocks
diff --git a/src/gr-fadingui/grc/fadingui_xor_frame_sync.block.yml b/src/gr-fadingui/grc/fadingui_xor_frame_sync.block.yml
deleted file mode 100644
index 1a8640d..0000000
--- a/src/gr-fadingui/grc/fadingui_xor_frame_sync.block.yml
+++ /dev/null
@@ -1,42 +0,0 @@
-id: fadingui_xor_frame_sync
-label: XOR Correlation Synchronizer
-category: '[fadingui]'
-flags: [ python ]
-
-templates:
- imports: import fadingui
- make: fadingui.xor_frame_sync(sync_pattern=${pattern}, buffer_size=${buffer_size})
-
-# Make one 'parameters' list entry for every parameter you want settable from the GUI.
-# Keys include:
-# * id (makes the value accessible as \$keyname, e.g. in the make entry)
-# * label (label shown in the GUI)
-# * dtype (e.g. int, float, complex, byte, short, xxx_vector, ...)
-parameters:
-- id: pattern
- label: Bit pattern
- dtype: raw
-- id: buffer_size
- label: Delay buffer size
- dtype: raw
-
-# Make one 'inputs' list entry per input and one 'outputs' list entry per output.
-# Keys include:
-# * label (an identifier for the GUI)
-# * domain (optional - stream or message. Default is stream)
-# * dtype (e.g. int, float, complex, byte, short, xxx_vector, ...)
-# * vlen (optional - data stream vector length. Default is 1)
-# * optional (optional - set to 1 for optional inputs. Default is 0)
-inputs:
-- label: in
- domain: stream
- dtype: byte
-
-outputs:
-- label: out
- domain: stream
- dtype: byte
-
-# 'file_format' specifies the version of the GRC yml format used in the file
-# and should usually not be changed.
-file_format: 1
diff --git a/src/gr-fadingui/python/CMakeLists.txt b/src/gr-fadingui/python/CMakeLists.txt
index 95bb852..eb0e7cc 100644
--- a/src/gr-fadingui/python/CMakeLists.txt
+++ b/src/gr-fadingui/python/CMakeLists.txt
@@ -35,7 +35,6 @@ GR_PYTHON_INSTALL(
logger.py
datasource.py
dearpygui_sink.py
- xor_frame_sync.py
deframer.py
frame_obj.py
multipath_fading.py DESTINATION ${GR_PYTHON_DIR}/fadingui
@@ -48,5 +47,4 @@ include(GrTest)
set(GR_TEST_TARGET_DEPS gnuradio-fadingui)
set(GR_TEST_PYTHON_DIRS ${CMAKE_BINARY_DIR}/swig)
-GR_ADD_TEST(qa_xor_frame_sync ${PYTHON_EXECUTABLE} ${CMAKE_CURRENT_SOURCE_DIR}/qa_xor_frame_sync.py)
GR_ADD_TEST(qa_multipath_fading ${PYTHON_EXECUTABLE} ${CMAKE_CURRENT_SOURCE_DIR}/qa_multipath_fading.py)
diff --git a/src/gr-fadingui/python/__init__.py b/src/gr-fadingui/python/__init__.py
index 5a7b546..dba999e 100644
--- a/src/gr-fadingui/python/__init__.py
+++ b/src/gr-fadingui/python/__init__.py
@@ -34,7 +34,7 @@ except ImportError:
# import any pure python here
from .datasource import datasource
from .dearpygui_sink import dearpygui_sink
-from .xor_frame_sync import xor_frame_sync
+
from .deframer import deframer
from .frame_obj import frame_obj
from .multipath_fading import multipath_fading
diff --git a/src/gr-fadingui/python/qa_xor_frame_sync.py b/src/gr-fadingui/python/qa_xor_frame_sync.py
deleted file mode 100644
index 9c480a0..0000000
--- a/src/gr-fadingui/python/qa_xor_frame_sync.py
+++ /dev/null
@@ -1,57 +0,0 @@
-#!/usr/bin/env python
-
-from gnuradio import gr, gr_unittest, blocks
-
-from xor_frame_sync import xor_frame_sync
-import numpy as np
-
-
-class test_xor_frame_sync(gr_unittest.TestCase):
-
- def setUp(self):
- self.tb = gr.top_block()
-
- def tearDown(self):
- self.tb = None
-
- def test_001(self):
- """Test a byte aligned delay"""
- pattern = np.array([0xc0, 0xff, 0xee], dtype=np.uint8)
- testdata = np.packbits(np.concatenate([
- np.unpackbits(np.arange(0, 5, dtype=np.uint8)),
- np.random.randint(0, 2, size = 8 * 5),
- np.unpackbits(pattern),
- np.random.randint(0, 2, size = 64)
- ]))
-
- src = blocks.vector_source_b(testdata)
- op = xor_frame_sync(pattern, 2048)
- dst = blocks.vector_sink_b()
-
- self.tb.connect(src, op, dst)
- self.tb.run()
-
- self.assertEqual(op.synchronized, True)
-
- # FIXME: implement feature
- # def test_002(self):
- # """Test a byte unaligned delay"""
- # pattern = np.array([0xbe, 0xef], dtype=np.uint8)
- # testdata = np.packbits(np.concatenate([
- # np.unpackbits(np.arange(0, 10, dtype=np.uint8)),
- # np.random.randint(0, 2, size = (2 + 8 * 5)), np.unpackbits(pattern),
- # np.random.randint(0, 2, size = 64)
- # ]))
-
- # src = blocks.vector_source_b(testdata)
- # op = xor_frame_sync(pattern, 2048)
- # dst = blocks.vector_sink_b()
-
- # self.tb.connect(src, op, dst)
- # self.tb.run()
-
- # self.assertEqual(op.synchronized, True)
-
-
-if __name__ == "__main__":
- gr_unittest.run(test_xor_frame_sync)
diff --git a/src/gr-fadingui/python/xor_frame_sync.py b/src/gr-fadingui/python/xor_frame_sync.py
deleted file mode 100644
index bb5cfb1..0000000
--- a/src/gr-fadingui/python/xor_frame_sync.py
+++ /dev/null
@@ -1,152 +0,0 @@
-#!/usr/bin/env python
-# -*- coding: utf-8 -*-
-#
-# Copyright 2021 Naoki Pross.
-
-
-import numpy as np
-from numpy_ringbuffer import RingBuffer
-
-from gnuradio import gr
-
-from fadingui.logger import get_logger
-log = get_logger("xor_frame_sync")
-
-
-class xor_frame_sync(gr.sync_block):
- """
- Performs a frame synchronization by XOR matching a preamble bit sequence
- """
- def __init__(self, sync_pattern, buffer_size):
- # TODO: buffer size should be in packets
- gr.sync_block.__init__(self,
- name="xor_frame_sync",
- in_sig=[np.byte],
- out_sig=[np.byte])
-
- # binary pattern to match
- self.pattern = sync_pattern
- self.nbytes = len(self.pattern)
-
- self.pattern_bits = np.unpackbits(np.array(self.pattern, dtype=np.uint8))[::-1]
- self.nbits = len(self.pattern_bits)
-
- log.debug(f"Loaded pattern {self.pattern_bits} length={self.nbits}")
- assert(self.nbits % 8 == 0)
-
- # packed buffer to delay the data
- self.delaybuf = RingBuffer(buffer_size, dtype=np.uint8)
- self.delay = 0
-
- log.debug(f"Created delay ring buffer of size {self.delaybuf.maxlen}")
-
- # unpacked buffer to compute correlation values, initially filled with zeros
- self.corrbuf = RingBuffer(self.nbits, dtype=np.uint8)
- self.corrbuf.extend(np.zeros(self.corrbuf.maxlen))
-
- # synchronization state
- self.synchronized = False
-
- def xcorrelation(self, v):
- """
- Compute the binary correlations between the stored pattern and
- correlation buffer, while shifting v into the buffer.
-
- Binary correlation between two bit vectors is just size of the
- vector(s) minus the number of bits that differ.
-
- @return: Number of bits of v that were shifted into the buffer
- when the correlation matched. If no match is found
- the return value is None.
- """
- # this could be much faster with shifts, bitwise or and xor
- # but this should do alright for the moment
- v_bits = np.unpackbits(np.array(v, dtype=np.uint8))
- for bitnr, b in enumerate(v_bits):
- self.corrbuf.appendleft(b)
- if (np.bitwise_xor(self.corrbuf, self.pattern_bits) == 0).all():
- return bitnr
-
- # no cross correlation found
- return None
-
- def work(self, input_items, output_items):
- """
- Process the inputs, that means:
-
- - Check that the buffer is synchronized, i.e. there is the sync
- pattern appears every k bits, where k is the size of the packet.
-
- - If the buffer is not synchronized, compute a binary cross
- correlation to find how much the stream should be delayed.
- """
- # array of samples, growing index = forward in time
- inp = input_items[0]
- inp_len = len(inp)
-
- if not self.synchronized:
- if inp_len > self.delaybuf.maxlen:
- log.error("Input is bigger than delay buffer")
-
- # FIXME: Makes the QA hang for some reason
- raise NotImplemented
-
- # create space for new samples in the delay buffer
- self.delaybuf.extendleft(np.zeros(inp_len))
-
- # Add values and while processing
- for bytenr, value in enumerate(inp):
- # save value in the buffer
- # FIXME: this is wrong, it should be in reverse order
- self.delaybuf.appendleft(value)
-
- # compute the cross correlation
- bitnr = self.xcorrelation(value)
- if bitnr is not None:
- # correlation was found
- delay_bits = (bitnr - 7)
- delay_bytes = 8 * (bytenr -1)
-
- log.debug(f"Synchronized with delay_bytes={delay_bytes} delay_bits={delay_bits}")
-
- # FIXME: add bit delay
- self.delay = delay_bytes
- self.synchronized = True
-
- # Not aligned to bytes
- if delay_bits != 0:
- log.error("Not implemented: byte unaligned delay")
- self.synchronized = False
- self.delay = 0
-
- # FIXME: Makes the QA hang for some reason
- # raise NotImplemented
-
- # stop processing inputs
- break
-
- if not self.synchronized:
- log.warning(f"Processed {inp_len} samples but could not synchronize")
- else:
- self.delaybuf.extendleft(inp)
-
- # return data with delay
- out = output_items[0]
- # FIXME: this is also wrong
- # out[:] = self.delaybuf[:len(out)]
- out[:] = inp[:]
-
-
- inptmp = np.array(inp[:12], dtype=np.uint8)
- inphex = np.array(list(map(hex, inptmp)))
- inpbits = np.array(list(map("{:08b}".format, inptmp)))
-
- log.debug(f"inp={inptmp}")
- log.debug(f"inp={inphex}")
- log.debug(f"inp={inpbits}")
-
- # outtmp = np.array(out[:12], dtype=np.uint8)
- # log.debug(f"out={outtmp}")
-
- return inp_len
-