diff options
Diffstat (limited to 'src/gr-fadingui/python')
| -rw-r--r-- | src/gr-fadingui/python/CMakeLists.txt | 8 | ||||
| -rw-r--r-- | src/gr-fadingui/python/__init__.py | 6 | ||||
| -rw-r--r-- | src/gr-fadingui/python/ber.py | 58 | ||||
| -rw-r--r-- | src/gr-fadingui/python/dearpygui_sink.py | 36 | ||||
| -rw-r--r-- | src/gr-fadingui/python/netsink.py | 84 | ||||
| -rwxr-xr-x | src/gr-fadingui/python/qa_ber.py | 48 | ||||
| -rw-r--r-- | src/gr-fadingui/python/qa_xor_frame_sync.py | 57 | ||||
| -rw-r--r-- | src/gr-fadingui/python/xor_frame_sync.py | 152 |
8 files changed, 198 insertions, 251 deletions
diff --git a/src/gr-fadingui/python/CMakeLists.txt b/src/gr-fadingui/python/CMakeLists.txt index 95bb852..bf484cc 100644 --- a/src/gr-fadingui/python/CMakeLists.txt +++ b/src/gr-fadingui/python/CMakeLists.txt @@ -34,11 +34,11 @@ GR_PYTHON_INSTALL( __init__.py logger.py datasource.py - dearpygui_sink.py - xor_frame_sync.py deframer.py frame_obj.py - multipath_fading.py DESTINATION ${GR_PYTHON_DIR}/fadingui + multipath_fading.py + ber.py + netsink.py DESTINATION ${GR_PYTHON_DIR}/fadingui ) ######################################################################## @@ -48,5 +48,5 @@ 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) +GR_ADD_TEST(qa_ber ${PYTHON_EXECUTABLE} ${CMAKE_CURRENT_SOURCE_DIR}/qa_ber.py) diff --git a/src/gr-fadingui/python/__init__.py b/src/gr-fadingui/python/__init__.py index 5a7b546..56dbdd3 100644 --- a/src/gr-fadingui/python/__init__.py +++ b/src/gr-fadingui/python/__init__.py @@ -33,10 +33,12 @@ 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 +from .ber import ber +from .netsink import netsink # diff --git a/src/gr-fadingui/python/ber.py b/src/gr-fadingui/python/ber.py new file mode 100644 index 0000000..e966f17 --- /dev/null +++ b/src/gr-fadingui/python/ber.py @@ -0,0 +1,58 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- +# +# Copyright 2021 Sara Cinzia Halter. +# +# This is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation; either version 3, or (at your option) +# any later version. +# +# This software is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this software; see the file COPYING. If not, write to +# the Free Software Foundation, Inc., 51 Franklin Street, +# Boston, MA 02110-1301, USA. +# + + +import numpy as np +from gnuradio import gr + +from fadingui.logger import get_logger +log = get_logger("ber") + +class ber(gr.sync_block): + """ + docstring for block ber + """ + def __init__(self, vgl, vlen): + gr.sync_block.__init__(self, + name="ber", + in_sig=[np.dtype(str(vlen) + "b")], + out_sig=None) + self.vgl=vgl + self.vlen=vlen + + def work(self, input_items, output_items): + + inp = input_items[0] + ber_tot = 0 + log.debug(f"Length: {len(inp)}") + log.debug(f"Inp_vector:{inp}") + + for i in inp: + log.debug(f"In Schlaufe{i}") + v = np.array(self.vgl, dtype=np.uint8)^np.array(i, dtype=np.uint8) + ber = sum(np.unpackbits(v)) + log.debug(f"BER {ber} in Paket {i}") + ber_tot+=ber + log.debug(f"BER Total{ber_tot}") + + + return len(input_items[0]) + diff --git a/src/gr-fadingui/python/dearpygui_sink.py b/src/gr-fadingui/python/dearpygui_sink.py deleted file mode 100644 index 6153611..0000000 --- a/src/gr-fadingui/python/dearpygui_sink.py +++ /dev/null @@ -1,36 +0,0 @@ -#!/usr/bin/env python -# -*- coding: utf-8 -*- -# -# Copyright 2021 Naoki Pross. - -import socket -from urllib.parse import urlparse - -import numpy as np -from gnuradio import gr - -class dearpygui_sink(gr.sync_block): - """ - DearPyGUI Sink - """ - def __init__(self, sock_addr, ui_element_id): - gr.sync_block.__init__(self, - name="dearpygui_sink", - in_sig=[np.complex64], - out_sig=None) - - # sockets - self.socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) - self.srv = urlparse(sock_addr) - - def send(self, value): - data = value.tobytes() - sent = self.socket.sendto(data, (self.srv.hostname, self.srv.port)) - - return len(data) == sent - - def work(self, input_items, output_items): - in0 = input_items[0] - self.send(in0) - return len(input_items[0]) - diff --git a/src/gr-fadingui/python/netsink.py b/src/gr-fadingui/python/netsink.py new file mode 100644 index 0000000..130e5dc --- /dev/null +++ b/src/gr-fadingui/python/netsink.py @@ -0,0 +1,84 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- +# +# Copyright 2021 Sara Cinzia Halter, Naoki Pross. + +import socket +from urllib.parse import urlparse + +import numpy as np +from gnuradio import gr + +class netsink(gr.sync_block): + """ + Sink that sends the data over the network using UDP. + Keep in mind that is quite slow. + """ + def __init__(self, address, dtype, vlen): + to_numpy = { + "complex": np.complex64, + "float": np.float32, + "int": np.int32, + "short": np.short, + "byte": np.byte, + } + + dt = to_numpy[dtype] + if vlen > 1: + dt = np.dtype(dt, (vlen,)) + print(dt) + + gr.sync_block.__init__(self, + name="Network Sink", + in_sig=[dt], + out_sig=None) + + # Create a socket and parse remote machine url + self.socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + self.url = urlparse(address) + self.srv = (self.url.hostname, self.url.port) + + def send(self, data): + """ + Send the data to self.srv + + @param data Data as python bytes + @return Number of bytes that were actually sent + """ + assert type(data) == bytes + return self.socket.sendto(data, self.srv) + + def encode(self, data): + """ + Encode the data into a dead simple format + + @param data Array like type + @return Bytes of ASCII encoded comma separated string of numbers + """ + # FIXME: this could be (very) slow, is there a faster way with numpy? + values = "[" + ",".join(map(str, data)) + "]" + return bytes(values, "ascii") + + def work(self, input_items, output_items): + # FIXME: it is probably better NOT to send *every* sample + inp = input_items[0] + inp_len = len(inp) + blocksize = 1024 + + # Check that the packet is not huge + if len(inp) < blocksize: + self.send(self.encode(inp)) + else: + # compute how to split inp into blocks + nblocks = inp_len // blocksize + index = blocksize * nblocks + + # send blocks + blocks = np.array(inp[:index]).reshape((blocksize, nblocks)) + for block in blocks: + self.send(self.encode(block)) + + # sent the rest + self.send(self.encode(inp[index:])) + + return len(inp) diff --git a/src/gr-fadingui/python/qa_ber.py b/src/gr-fadingui/python/qa_ber.py new file mode 100755 index 0000000..cb6c198 --- /dev/null +++ b/src/gr-fadingui/python/qa_ber.py @@ -0,0 +1,48 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- +# +# Copyright 2021 Sara Cinzia Halter. +# +# This is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation; either version 3, or (at your option) +# any later version. +# +# This software is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this software; see the file COPYING. If not, write to +# the Free Software Foundation, Inc., 51 Franklin Street, +# Boston, MA 02110-1301, USA. +# + +from gnuradio import gr, gr_unittest +from gnuradio import blocks +from ber import ber +import numpy as np + +class qa_ber(gr_unittest.TestCase): + + def setUp(self): + self.tb = gr.top_block() + + def tearDown(self): + self.tb = None + + def test_001_t(self): + # pattern = np.array([0xaa], dtype=np.uint8) + # testdata = np.array([0xc0, 0xfa, 0xae] * 4, dtype=np.uint8) + + # src = blocks.vector_source_b(testdata) + # op = ber(pattern) + + # self.tb.connect(src, op) + # self.tb.run() + pass + + +if __name__ == '__main__': + gr_unittest.run(qa_ber) 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 - |