83 lines
3.0 KiB
Python
83 lines
3.0 KiB
Python
|
import time
|
||
|
|
|
||
|
|
import numpy as np
|
||
|
|
from PyQt5.QtTest import QTest
|
||
|
|
|
||
|
|
from tests.QtTestCase import QtTestCase
|
||
|
|
from urh import settings
|
||
|
|
from urh.dev.BackendHandler import BackendHandler
|
||
|
|
from urh.plugins.NetworkSDRInterface.NetworkSDRInterfacePlugin import NetworkSDRInterfacePlugin
|
||
|
|
from urh.signalprocessing.IQArray import IQArray
|
||
|
|
from urh.signalprocessing.Modulator import Modulator
|
||
|
|
from urh.signalprocessing.ProtocolAnalyzer import ProtocolAnalyzer
|
||
|
|
from urh.signalprocessing.ProtocolSniffer import ProtocolSniffer
|
||
|
|
from urh.signalprocessing.Signal import Signal
|
||
|
|
from urh.util import util
|
||
|
|
|
||
|
|
|
||
|
|
class TestProtocolSniffer(QtTestCase):
|
||
|
|
def setUp(self):
|
||
|
|
super().setUp()
|
||
|
|
settings.OVERWRITE_RECEIVE_BUFFER_SIZE = 50000
|
||
|
|
|
||
|
|
def test_protocol_sniffer(self):
|
||
|
|
samples_per_symbol = 100
|
||
|
|
center = 0.0942
|
||
|
|
noise = 0.1
|
||
|
|
tolerance = 2
|
||
|
|
modulation_type = "FSK"
|
||
|
|
sample_rate = 1e6
|
||
|
|
device_name = NetworkSDRInterfacePlugin.NETWORK_SDR_NAME
|
||
|
|
sniffer = ProtocolSniffer(samples_per_symbol=samples_per_symbol, center=center, center_spacing=0.1,
|
||
|
|
noise=noise, tolerance=tolerance,
|
||
|
|
modulation_type=modulation_type, bits_per_symbol=1,
|
||
|
|
device=device_name, backend_handler=BackendHandler(),
|
||
|
|
network_raw_mode=True)
|
||
|
|
|
||
|
|
port = util.get_free_port()
|
||
|
|
sniffer.rcv_device.set_server_port(port)
|
||
|
|
|
||
|
|
self.network_sdr_plugin_sender = NetworkSDRInterfacePlugin(raw_mode=True)
|
||
|
|
self.network_sdr_plugin_sender.client_port = port
|
||
|
|
|
||
|
|
sniffer.sniff()
|
||
|
|
QTest.qWait(10)
|
||
|
|
|
||
|
|
data = ["101010", "000111", "1111000"]
|
||
|
|
pause = 10 * samples_per_symbol
|
||
|
|
modulator = Modulator("test")
|
||
|
|
modulator.samples_per_symbol = samples_per_symbol
|
||
|
|
modulator.sample_rate = sample_rate
|
||
|
|
modulator.modulation_type = modulation_type
|
||
|
|
modulator.parameters[1] = 20e3
|
||
|
|
modulator.parameters[0] = 10e3
|
||
|
|
|
||
|
|
packages = []
|
||
|
|
for d in data:
|
||
|
|
packages.append(modulator.modulate(list(map(int, d)), pause))
|
||
|
|
|
||
|
|
# verify modulation was correct
|
||
|
|
pa = ProtocolAnalyzer(None)
|
||
|
|
signal = Signal("", "", sample_rate=sample_rate)
|
||
|
|
signal.iq_array = IQArray.concatenate(packages)
|
||
|
|
signal.modulation_type = modulation_type
|
||
|
|
signal.samples_per_symbol = samples_per_symbol
|
||
|
|
signal.tolerance = tolerance
|
||
|
|
signal.noise_threshold = noise
|
||
|
|
signal.center = center
|
||
|
|
pa.signal = signal
|
||
|
|
pa.get_protocol_from_signal()
|
||
|
|
self.assertEqual(pa.plain_bits_str, data)
|
||
|
|
|
||
|
|
# send data
|
||
|
|
send_data = IQArray.concatenate(packages)
|
||
|
|
self.network_sdr_plugin_sender.send_raw_data(send_data, 1)
|
||
|
|
time.sleep(1)
|
||
|
|
|
||
|
|
# Send enough pauses to end sniffing
|
||
|
|
self.network_sdr_plugin_sender.send_raw_data(IQArray(None, np.float32, 10 * 2 * samples_per_symbol), 1)
|
||
|
|
time.sleep(1)
|
||
|
|
|
||
|
|
sniffer.stop()
|
||
|
|
self.assertEqual(sniffer.plain_bits_str, data)
|