aboutsummaryrefslogtreecommitdiffstats
path: root/src/gr-fadingui/python
diff options
context:
space:
mode:
Diffstat (limited to '')
-rw-r--r--src/gr-fadingui/python/CMakeLists.txt8
-rw-r--r--src/gr-fadingui/python/__init__.py6
-rw-r--r--src/gr-fadingui/python/ber.py58
-rw-r--r--src/gr-fadingui/python/dearpygui_sink.py36
-rw-r--r--src/gr-fadingui/python/netsink.py84
-rwxr-xr-xsrc/gr-fadingui/python/qa_ber.py48
-rw-r--r--src/gr-fadingui/python/qa_xor_frame_sync.py57
-rw-r--r--src/gr-fadingui/python/xor_frame_sync.py152
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
-
generated by cgit v1.2.1 (git 2.18.0) at 2024-11-27 01:47:06 +0100