diff options
Diffstat (limited to '')
| -rw-r--r-- | src/gr-fadingui/python/CMakeLists.txt | 2 | ||||
| -rw-r--r-- | src/gr-fadingui/python/datasource.py | 4 | ||||
| -rw-r--r-- | src/gr-fadingui/python/qa_xor_frame_sync.py | 56 | ||||
| -rw-r--r-- | src/gr-fadingui/python/xor_frame_sync.py | 101 |
4 files changed, 127 insertions, 36 deletions
diff --git a/src/gr-fadingui/python/CMakeLists.txt b/src/gr-fadingui/python/CMakeLists.txt index 1318857..6ee73d3 100644 --- a/src/gr-fadingui/python/CMakeLists.txt +++ b/src/gr-fadingui/python/CMakeLists.txt @@ -47,4 +47,4 @@ include(GrTest) set(GR_TEST_TARGET_DEPS gnuradio-fadingui) set(GR_TEST_PYTHON_DIRS ${CMAKE_BINARY_DIR}/swig) -GR_ADD_TEST(qa_dearpygui_sink ${PYTHON_EXECUTABLE} ${CMAKE_CURRENT_SOURCE_DIR}/qa_dearpygui_sink.py) +GR_ADD_TEST(qa_xor_frame_sync ${PYTHON_EXECUTABLE} ${CMAKE_CURRENT_SOURCE_DIR}/qa_xor_frame_sync.py) diff --git a/src/gr-fadingui/python/datasource.py b/src/gr-fadingui/python/datasource.py index 12573a8..38771c5 100644 --- a/src/gr-fadingui/python/datasource.py +++ b/src/gr-fadingui/python/datasource.py @@ -35,7 +35,7 @@ class datasource(gr.basic_block): # a frame has 5 id bits so, there can only be 2 ** 5 chunks per file # see docstring of frame_obj for more details nblocks = int(self.fsize / self.frame.payload_length) - log.debug(f"Loaded {self.fsize} bytes = {nblocks} blocks from {self.fname}") + log.debug(f"Loaded {self.fsize} bytes == {nblocks} blocks from {self.fname}") assert nblocks < 2 ** 5, "Payload size too small or file too big" self.fpos = 0 @@ -73,7 +73,7 @@ class datasource(gr.basic_block): self.outbuffer = frame_bytes[len(out):] log.debug(f"Sent frame nr={self.blocknr}") - log.debug(f"Set bytes {out}") + log.debug(f"Sent bytes {out}") self.fpos += self.frame.payload_length self.blocknr += 1 diff --git a/src/gr-fadingui/python/qa_xor_frame_sync.py b/src/gr-fadingui/python/qa_xor_frame_sync.py new file mode 100644 index 0000000..280c694 --- /dev/null +++ b/src/gr-fadingui/python/qa_xor_frame_sync.py @@ -0,0 +1,56 @@ +#!/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([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 = 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 index 51af35a..e52b493 100644 --- a/src/gr-fadingui/python/xor_frame_sync.py +++ b/src/gr-fadingui/python/xor_frame_sync.py @@ -25,23 +25,24 @@ class xor_frame_sync(gr.sync_block): out_sig=[np.byte]) # binary pattern to match - self.pattern = np.unpackbits(np.array(sync_pattern, dtype=np.uint8))[::-1] - self.nbytes = len(sync_pattern) - self.nbits = len(self.pattern) + self.pattern = sync_pattern + self.nbytes = len(self.pattern) - log.debug(f"Loaded pattern {self.pattern} length={self.nbits}") + 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 - # unpacked buffer to compute correlation values, initially filled with zeros - self.corrbuf = RingBuffer(self.nbits) - self.corrbuf.extend(np.zeros(self.nbits)) + log.debug(f"Created delay ring buffer of size {self.delaybuf.maxlen}") - # buffer to store correlation values - self.xcorrs = RingBuffer(buffer_size) + # 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 @@ -53,11 +54,21 @@ class xor_frame_sync(gr.sync_block): 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. """ - v_arr = np.array(v, dtype=np.uint8) - for b in np.unpackbits(v_arr): + # 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) - yield self.nbits - np.sum(np.logical_xor(self.corrbuf, self.pattern)) + 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): """ @@ -68,35 +79,59 @@ class xor_frame_sync(gr.sync_block): - If the buffer is not synchronized, compute a binary cross correlation to find how much the stream should be delayed. - - Notes: - - - Even though the block input is of type np.byte, inp is an array - of 255 bytes, probably for performance reasons. - TODO: block processing """ inp = input_items[0] out = output_items[0] - if not self.synchronized: - for v in inp: - # compute the cross correlation - xcs = self.xcorrelation(v) + inp_len = len(inp) - # add cross correlations to buffer and save value - self.xcorrs.extend(list(xcs)) - self.delaybuf.appendleft(v) + if not self.synchronized: + if inp_len > self.delaybuf.maxlen: + log.error("Input is bigger than delay buffer") - peak = np.argmax(self.xcorrs) - if self.xcorrs[peak] == self.nbits: - self.delay = peak - self.synchronized = True - log.debug(f"Synchronized with delay={peak}") + # FIXME: Makes the QA hang for some reason + # raise NotImplemented - else: - self.synchronized = False - log.warning(f"Did not find a peak (max={self.xcorrs[peak]}, should be {self.nbits})") + for bytenr, value in enumerate(inp): + # save value in the buffer + self.delaybuf.appendleft(value) + # compute the cross correlation + bitnr = self.xcorrelation(value) + if bitnr is not None: + # correlation was found + delay_bits = 8 * bytenr + bitnr + + # FIXME: add bit delay + self.delay = bytenr + self.synchronized = True + log.debug(f"Synchronized with delay={self.delay} delay_bits={delay_bits}") + + # Not aligned to bytes + if bitnr != 7: + log.error("Not implemented: byte unaligned delay") + self.synchronized = False + self.delay = 0 + + # FIXME: Makes the QA hang for some reason + # raise NotImplemented + + # bigger than buffer + if bytenr > self.delaybuf.maxlen: + log.error("Too too long to synchronize, ran out of buffer memory") + 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[:] = self.delaybuf[self.delay] |