#!/usr/bin/env python3 # -*- coding: utf-8 -*- # TRX Toolkit # Virtual Um-interface (fake transceiver) # # (C) 2017-2019 by Vadim Yanitskiy # # All Rights Reserved # # This program 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 2 of the License, or # (at your option) any later version. # # This program 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. APP_CR_HOLDERS = [("2017-2019", "Vadim Yanitskiy ")] import logging as log import signal import argparse import sys import re import os from app_common import ApplicationBase from clck_gen import CLCKGen from trx_list import TRXList from fake_pm import FakePM from _fake_trx import FakeTRX, Runner class Application(ApplicationBase): def __init__(self): self.app_print_copyright(APP_CR_HOLDERS) self.argv = self.parse_argv() # Set up signal handlers signal.signal(signal.SIGINT, self.sig_handler) # Configure logging self.app_init_logging(self.argv) # List of all transceivers self.trx_list = TRXList() # Init shared clock generator self.clck_gen = CLCKGen([]) # Power measurement emulation # Noise: -120 .. -105 # BTS: -75 .. -50 self.fake_pm = FakePM(-120, -105, -75, -50) self.fake_pm.trx_list = self.trx_list # Init TRX instance for BTS self.append_trx(self.argv.bts_addr, self.argv.bts_base_port, name = "BTS") # Init TRX instance for BB self.append_trx(self.argv.bb_addr, self.argv.bb_base_port, name = "MS", child_mgt = False) # Additional transceivers (optional) if self.argv.trx_list is not None: for trx_def in self.argv.trx_list: (name, addr, port, idx) = trx_def self.append_child_trx(addr, port, name = name, child_idx = idx) log.info("Init complete") def append_trx(self, remote_addr, base_port, **kwargs): trx = FakeTRX(self.argv.trx_bind_addr, remote_addr, base_port, clck_gen = self.clck_gen, pwr_meas = self.fake_pm, **kwargs) self.trx_list.add_trx(trx) def append_child_trx(self, remote_addr, base_port, **kwargs): child_idx = kwargs.get("child_idx", 0) if child_idx == 0: # Index 0 indicates parent transceiver self.append_trx(remote_addr, base_port, **kwargs) return # Find 'parent' transceiver for a new child trx_parent = self.trx_list.find_trx(remote_addr, base_port) if trx_parent is None: raise IndexError("Couldn't find parent transceiver " "for '%s:%d/%d'" % (remote_addr, base_port, child_idx)) # Allocate a new child trx_child = FakeTRX(self.argv.trx_bind_addr, remote_addr, base_port, pwr_meas = self.fake_pm, **kwargs) self.trx_list.add_trx(trx_child) # Link a new 'child' with its 'parent' trx_parent.child_trx_list.add_trx(trx_child) def run(self): if self.argv.sched_rr_prio is not None: sched_param = os.sched_param(self.argv.sched_rr_prio) try: log.info("Setting real time process scheduler to SCHED_RR, priority %u" % (self.argv.sched_rr_prio)) os.sched_setscheduler(0, os.SCHED_RR, sched_param) except OSError: log.error("Failed to set real time process scheduler to SCHED_RR, priority %u" % (self.argv.sched_rr_prio)) runner = Runner(self.clck_gen, self.trx_list) runner.loop() def shutdown(self): log.info("Shutting down...") # Stop clock generator self.clck_gen.stop() # Parses a TRX definition of the following # format: REMOTE_ADDR:BIND_PORT[/TRX_NUM] # e.g. [2001:0db8:85a3:0000:0000:8a2e:0370:7334]:5700/5 # e.g. 127.0.0.1:5700 or 127.0.0.1:5700/1 # e.g. foo@127.0.0.1:5700 or bar@127.0.0.1:5700/1 @staticmethod def trx_def(val): try: result = re.match(r"(.+@)?(.+):([0-9]+)(/[0-9]+)?", val) (name, addr, port, idx) = result.groups() except: raise argparse.ArgumentTypeError("Invalid TRX definition: %s" % val) if idx is not None: idx = int(idx[1:]) else: idx = 0 # Cut '@' from TRX name if name is not None: name = name[:-1] return (name, addr, int(port), idx) def parse_argv(self): parser = argparse.ArgumentParser(prog = "fake_trx", description = "Virtual Um-interface (fake transceiver)") # Register common logging options self.app_reg_logging_options(parser) trx_group = parser.add_argument_group("TRX interface") trx_group.add_argument("-b", "--trx-bind-addr", dest = "trx_bind_addr", type = str, default = "0.0.0.0", help = "Set FakeTRX bind address (default %(default)s)") trx_group.add_argument("-R", "--bts-addr", dest = "bts_addr", type = str, default = "127.0.0.1", help = "Set BTS remote address (default %(default)s)") trx_group.add_argument("-r", "--bb-addr", dest = "bb_addr", type = str, default = "127.0.0.1", help = "Set BB remote address (default %(default)s)") trx_group.add_argument("-P", "--bts-base-port", dest = "bts_base_port", type = int, default = 5700, help = "Set BTS base port number (default %(default)s)") trx_group.add_argument("-p", "--bb-base-port", dest = "bb_base_port", type = int, default = 6700, help = "Set BB base port number (default %(default)s)") trx_group.add_argument("-s", "--sched-rr-prio", dest = "sched_rr_prio", type = int, default = None, help = "Set Scheduler RR Priority (default None)") mtrx_group = parser.add_argument_group("Additional transceivers") mtrx_group.add_argument("--trx", metavar = "REMOTE_ADDR:BASE_PORT[/TRX_NUM]", dest = "trx_list", type = self.trx_def, action = "append", help = "Add a transceiver for BTS or MS (e.g. 127.0.0.1:5703)") argv = parser.parse_args() # Make sure there is no overlap between ports if argv.bts_base_port == argv.bb_base_port: parser.error("BTS and BB base ports shall be different") return argv def sig_handler(self, signum, frame): log.info("Signal %d received" % signum) if signum == signal.SIGINT: self.shutdown() sys.exit(0) if __name__ == '__main__': app = Application() app.run()