diff --git a/scriptsapi/superscanner.py b/scriptsapi/superscanner.py index e47ba0da0..5a9fc712e 100644 --- a/scriptsapi/superscanner.py +++ b/scriptsapi/superscanner.py @@ -47,7 +47,9 @@ class SuperScannerAPIError(SuperScannerError): pass # ====================================================================== -def get_input_options(): +def get_input_options(args=None): + if args is None: + args = sys.argv[1:] parser = OptionParser(usage="usage: %%prog [-t]\n") parser.add_option("-a", "--address", dest="address", help="SDRangel web base address. Default: 127.0.0.1", metavar="ADDRESS", type="string") @@ -66,7 +68,7 @@ def get_input_options(): parser.add_option("-r", "--freq-round", dest="freq_round", help="Frequency rounding value in Hz. Default: 1 (no rounding)", metavar="NUM", type="int") parser.add_option("-o", "--freq-offset", dest="freq_offset", help="Frequency rounding offset in Hz. Default: 0 (no offset)", metavar="NUM", type="int") - (options, args) = parser.parse_args() + (options, args) = parser.parse_args(args) if (options.config_file == None): raise SuperScannerOptionsError('A configuration file is required. Option -c or --config-file') @@ -293,7 +295,8 @@ def process_hotspots(scanned_hotspots): used_channels = [channel for channel in channels if channel['usage'] == 1] for channel in used_channels: # loop on used channels distances = [[abs(channel['frequency'] - get_hotspot_frequency(channel, hotspot)), hotspot] for hotspot in hotspots] - sorted(distances, key=operator.itemgetter(0)) + distances = sorted(distances, key=operator.itemgetter(0)) + print(f'channel {channel["index"]} distances: {distances}') if distances: hotspot = distances[0][1] channel['usage'] = 2 # mark channel used on this pass @@ -371,8 +374,6 @@ def get_deviceset_info(deviceset_index): # ====================================================================== def make_config(): global CONFIG - with open(OPTIONS.config_file) as json_file: # get base config - CONFIG = json.load(json_file) deviceset_index = CONFIG['deviceset_index'] deviceset_info = get_deviceset_info(deviceset_index) device_frequency = deviceset_info["samplingDevice"]["centerFrequency"] @@ -392,15 +393,19 @@ def make_config(): def main(): try: global OPTIONS + global CONFIG global API_URI global WS_URI OPTIONS = get_input_options() + with open(OPTIONS.config_file) as json_file: # get base config + CONFIG = json.load(json_file) + API_URI = f'http://{OPTIONS.address}:{OPTIONS.api_port}' WS_URI = f'ws://{OPTIONS.address}:{OPTIONS.ws_port}' - make_config() + make_config() # complete config with device set information from SDRangel ws = websocket.WebSocketApp(WS_URI, on_message = on_ws_message, diff --git a/scriptsapi/test_superscanner.py b/scriptsapi/test_superscanner.py new file mode 100644 index 000000000..f070989f5 --- /dev/null +++ b/scriptsapi/test_superscanner.py @@ -0,0 +1,205 @@ +import unittest +import mock +import superscanner + +def print_hex(bytestring): + print('\\x' + '\\x'.join('{:02x}'.format(x) for x in bytestring)) + +def get_deviceset_info(deviceset_index): + return { + "channelcount": 4, + "channels": [ + { + "deltaFrequency": 170000, + "direction": 0, + "id": "NFMDemod", + "index": 0, + "title": "NFM Demodulator", + "uid": 1590355926650308 + }, + { + "deltaFrequency": -155000, + "direction": 0, + "id": "DSDDemod", + "index": 1, + "title": "DSD Demodulator", + "uid": 1590355926718405 + }, + { + "deltaFrequency": 170000, + "direction": 0, + "id": "NFMDemod", + "index": 2, + "title": "NFM Demodulator", + "uid": 1590355926939766 + }, + { + "deltaFrequency": -95000, + "direction": 0, + "id": "NFMDemod", + "index": 3, + "title": "NFM Demodulator", + "uid": 1590355926945674 + } + ], + "samplingDevice": { + "bandwidth": 768000, + "centerFrequency": 145480000, + "deviceNbStreams": 1, + "deviceStreamIndex": 0, + "direction": 0, + "hwType": "AirspyHF", + "index": 0, + "sequence": 0, + "serial": "c852a98040c73f93", + "state": "running" + } + } + +def set_channel_frequency(channel): + pass + +def set_channel_mute(channel): + pass + +class TestStringMethods(unittest.TestCase): + + def test_upper(self): + self.assertEqual('foo'.upper(), 'FOO') + + def test_isupper(self): + self.assertTrue('FOO'.isupper()) + self.assertFalse('Foo'.isupper()) + + def test_split(self): + s = 'hello world' + self.assertEqual(s.split(), ['hello', 'world']) + # check that s.split fails when the separator is not a string + with self.assertRaises(TypeError): + s.split(2) + +class TestSuperScannerOptions(unittest.TestCase): + + def test_options_minimal(self): + options = superscanner.get_input_options(["-ctoto"]) + self.assertEqual(options.config_file, 'toto') + +class TestSuperScannerDecode(unittest.TestCase): + + def test_decode_bytes(self): + msg_bytes = b'\x40\xd9\xab\x08\x00\x00\x00\x00' + \ + b'\xff\x00\x00\x00\x00\x00\x00\x00' + \ + b'\x69\x63\xbb\x55\x72\x01\x00\x00' + \ + b'\x00\x04\x00\x00' + \ + b'\x00\xb8\x0b\x00' + \ + b'\x04\x00\x00\x00' + for i in range(1024): + msg_bytes += b'\x00\x00\x00\x00' + msg_struct = superscanner.decode_message(msg_bytes) + self.assertEqual(msg_struct['fft_size'], 1024) + +class TestSuperScannerProcessHotspots(unittest.TestCase): + + @mock.patch('superscanner.get_deviceset_info', side_effect=get_deviceset_info) + def setUp(self, urandom_function): + self.options = type('options', (object,), {})() + self.options.address = '127.0.0.1' + self.options.passes = 10 + self.options.api_port = 8091 + self.options.ws_port = 8887 + self.options.config_file = 'toto' + self.options.psd_input_file = None + self.options.psd_output_file = None + self.options.passes = 10 + self.options.margin = 3 + self.options.psd_fixed = None + self.options.psd_exclude_higher = None + self.options.psd_exclude_lower = None + self.options.psd_graph = None + self.options.group_tolerance = 1 + self.options.freq_round = 1 + self.options.freq_offset = 0 + superscanner.OPTIONS = self.options + superscanner.CONFIG = { + "deviceset_index": 0, + "freqrange_exclusions": [ + [145290000, 145335000], + [145692500, 145707500] + ], + "freqrange_inclusions": [ + [145170000, 145800000] + ], + "channel_info": [ + { + "index": 0, + "fc_pos": "center" + }, + { + "index": 2 + }, + { + "index": 3 + } + ] + } + superscanner.make_config() + + def test_make_config(self): + self.assertEqual(superscanner.CONFIG['device_frequency'], 145480000) + + @mock.patch('superscanner.set_channel_frequency', side_effect=set_channel_frequency) + @mock.patch('superscanner.set_channel_mute', side_effect=set_channel_mute) + def test_process_hotspot(self, set_channel_frequency, set_channel_mute): + hotspots1 = [ + { + 'begin': 145550000, + 'end': 145550000, + 'power': -50 + } + ] + superscanner.process_hotspots(hotspots1) + channel_info = superscanner.CONFIG['channel_info'] + print(channel_info) + self.assertEqual(channel_info[0]['usage'], 1) + self.assertEqual(channel_info[1]['usage'], 0) + self.assertEqual(channel_info[2]['usage'], 0) + self.assertEqual(channel_info[0]['frequency'], 145550000) + hotspots2 = [ + { + 'begin': 145200000, + 'end': 145200000, + 'power': -35 + }, + { + 'begin': 145550000, + 'end': 145550000, + 'power': -50 + } + ] + superscanner.process_hotspots(hotspots2) + channel_info = superscanner.CONFIG['channel_info'] + print(channel_info) + self.assertEqual(channel_info[0]['usage'], 1) + self.assertEqual(channel_info[1]['usage'], 1) + self.assertEqual(channel_info[2]['usage'], 0) + self.assertEqual(channel_info[0]['frequency'], 145550000) + self.assertEqual(channel_info[1]['frequency'], 145200000) + hotspots3 = [ + { + 'begin': 145200000, + 'end': 145200000, + 'power': -35 + } + ] + superscanner.process_hotspots(hotspots3) + channel_info = superscanner.CONFIG['channel_info'] + print(channel_info) + self.assertEqual(channel_info[0]['usage'], 0) + self.assertEqual(channel_info[1]['usage'], 1) + self.assertEqual(channel_info[2]['usage'], 0) + self.assertEqual(channel_info[1]['frequency'], 145200000) + + +if __name__ == '__main__': + unittest.main() +