Vadim Yanitskiy has uploaded this change for review. (
https://gerrit.osmocom.org/12185
Change subject: trx_toolkit: use generic logging module instead of print()
......................................................................
trx_toolkit: use generic logging module instead of print()
There are multiple advantages of using Python's logging module:
- advanced message formatting (file name, line number, etc.),
- multiple logging targets (e.g. stderr, file, socket),
- logging levels (e.g. DEBUG, INFO, ERROR),
- the pythonic way ;)
so, let's replace multiple print() calls by logging calls,
add use the following logging message format by default:
[%(levelname)s] %(filename)s:%(lineno)d %(message)s
Examples:
[INFO] ctrl_if_bts.py:57 Starting transceiver...
[DEBUG] clck_gen.py:87 IND CLOCK 26826
[DEBUG] ctrl_if_bts.py:71 Recv POWEROFF cmd
[INFO] ctrl_if_bts.py:73 Stopping transceiver...
[INFO] fake_trx.py:127 Shutting down...
Please note that there is no way to filter messages by logging
level yet. This is to be introduced soon, together with argparse.
Change-Id: I7fcafabafe8323b58990997a47afdd48b6d1f357
---
M src/target/trx_toolkit/burst_fwd.py
M src/target/trx_toolkit/burst_gen.py
M src/target/trx_toolkit/burst_send.py
M src/target/trx_toolkit/clck_gen.py
M src/target/trx_toolkit/ctrl_cmd.py
M src/target/trx_toolkit/ctrl_if.py
M src/target/trx_toolkit/ctrl_if_bb.py
M src/target/trx_toolkit/ctrl_if_bts.py
M src/target/trx_toolkit/data_dump.py
M src/target/trx_toolkit/data_msg.py
M src/target/trx_toolkit/fake_trx.py
M src/target/trx_toolkit/trx_sniff.py
12 files changed, 145 insertions(+), 95 deletions(-)
git pull ssh://gerrit.osmocom.org:29418/osmocom-bb refs/changes/85/12185/1
diff --git a/src/target/trx_toolkit/burst_fwd.py
b/src/target/trx_toolkit/burst_fwd.py
index 31f882e..3cb6acd 100644
--- a/src/target/trx_toolkit/burst_fwd.py
+++ b/src/target/trx_toolkit/burst_fwd.py
@@ -22,6 +22,7 @@
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
+import logging as log
import random
from data_msg import *
@@ -214,7 +215,7 @@
# Burst dropping
if self.burst_dl_drop_amount > 0:
if msg.fn % self.burst_dl_drop_period == 0:
- print("[~] Simulation: dropping DL burst (fn=%u
%% %u == 0)"
+ log.info("Simulation: dropping DL burst (fn=%u
%% %u == 0)"
% (msg.fn, self.burst_dl_drop_period))
self.burst_dl_drop_amount -= 1
return None
@@ -226,7 +227,7 @@
# Burst dropping
if self.burst_ul_drop_amount > 0:
if msg.fn % self.burst_ul_drop_period == 0:
- print("[~] Simulation: dropping UL burst (fn=%u
%% %u == 0)"
+ log.info("Simulation: dropping UL burst (fn=%u
%% %u == 0)"
% (msg.fn, self.burst_ul_drop_period))
self.burst_ul_drop_amount -= 1
return None
@@ -254,7 +255,7 @@
msg_l12trx = DATAMSG_L12TRX()
msg_l12trx.parse_msg(bytearray(msg_raw))
except:
- print("[!] Dropping unhandled DL message...")
+ log.error("Dropping unhandled DL message...")
return None
# Compose a new message for L1
@@ -320,7 +321,7 @@
# Timeslot filter
if msg.tn not in self.ts_pass_list:
- print("[!] TS %u is not configured, dropping UL
burst..." % msg.tn)
+ log.warning("TS %u is not configured, dropping UL
burst..." % msg.tn)
return None
# Path loss simulation
diff --git a/src/target/trx_toolkit/burst_gen.py
b/src/target/trx_toolkit/burst_gen.py
index d83f137..9a17ffa 100755
--- a/src/target/trx_toolkit/burst_gen.py
+++ b/src/target/trx_toolkit/burst_gen.py
@@ -26,6 +26,7 @@
from copyright import print_copyright
CR_HOLDERS = [("2017-2018", "Vadim Yanitskiy <[email protected]>")]
+import logging as log
import signal
import getopt
import sys
@@ -64,6 +65,10 @@
# Set up signal handlers
signal.signal(signal.SIGINT, self.sig_handler)
+ # Configure logging
+ log.basicConfig(level = log.DEBUG,
+ format = "[%(levelname)s] %(filename)s:%(lineno)d
%(message)s")
+
# Open requested capture file
if self.output_file is not None:
self.ddf = DATADumpFile(self.output_file)
@@ -130,7 +135,7 @@
# Set burst
msg.burst = burst
- print("[i] Sending %d/%d %s burst %s to %s..."
+ log.info("Sending %d/%d %s burst %s to %s..."
% (i + 1, self.burst_count, self.burst_type,
msg.desc_hdr(), self.conn_mode))
@@ -239,7 +244,7 @@
sys.exit(2)
def sig_handler(self, signum, frame):
- print("Signal %d received" % signum)
+ log.info("Signal %d received" % signum)
if signum is signal.SIGINT:
sys.exit(0)
diff --git a/src/target/trx_toolkit/burst_send.py
b/src/target/trx_toolkit/burst_send.py
index f6c85ba..787e0fc 100755
--- a/src/target/trx_toolkit/burst_send.py
+++ b/src/target/trx_toolkit/burst_send.py
@@ -25,6 +25,7 @@
from copyright import print_copyright
CR_HOLDERS = [("2017-2018", "Vadim Yanitskiy <[email protected]>")]
+import logging as log
import signal
import getopt
import sys
@@ -60,6 +61,10 @@
# Set up signal handlers
signal.signal(signal.SIGINT, self.sig_handler)
+ # Configure logging
+ log.basicConfig(level = log.DEBUG,
+ format = "[%(levelname)s] %(filename)s:%(lineno)d
%(message)s")
+
# Open requested capture file
self.ddf = DATADumpFile(self.capture_file)
@@ -88,7 +93,7 @@
if not self.msg_pass_filter(l12trx, msg):
continue
- print("[i] Sending a burst %s to %s..."
+ log.info("Sending a burst %s to %s..."
% (msg.desc_hdr(), self.conn_mode))
# Send message
@@ -209,7 +214,7 @@
sys.exit(2)
def sig_handler(self, signum, frame):
- print("Signal %d received" % signum)
+ log.info("Signal %d received" % signum)
if signum is signal.SIGINT:
sys.exit(0)
diff --git a/src/target/trx_toolkit/clck_gen.py
b/src/target/trx_toolkit/clck_gen.py
index b488770..56207f4 100755
--- a/src/target/trx_toolkit/clck_gen.py
+++ b/src/target/trx_toolkit/clck_gen.py
@@ -25,6 +25,7 @@
from copyright import print_copyright
CR_HOLDERS = [("2017-2018", "Vadim Yanitskiy <[email protected]>")]
+import logging as log
import signal
import time
import sys
@@ -83,7 +84,7 @@
link.send(payload)
# Debug print
- print("[T] %s" % payload)
+ log.debug(payload)
# Increase frame count
self.clck_src += self.ind_period
@@ -101,13 +102,17 @@
# Set up signal handlers
signal.signal(signal.SIGINT, self.sig_handler)
+ # Configure logging
+ log.basicConfig(level = log.DEBUG,
+ format = "[%(levelname)s] %(filename)s:%(lineno)d
%(message)s")
+
def run(self):
self.link = UDPLink("127.0.0.1", 5800, "0.0.0.0", 5700)
self.clck = CLCKGen([self.link], ind_period = 51)
self.clck.start()
def sig_handler(self, signum, frame):
- print("Signal %d received" % signum)
+ log.info("Signal %d received" % signum)
if signum is signal.SIGINT:
self.clck.stop()
diff --git a/src/target/trx_toolkit/ctrl_cmd.py
b/src/target/trx_toolkit/ctrl_cmd.py
index e56105a..ec683be 100755
--- a/src/target/trx_toolkit/ctrl_cmd.py
+++ b/src/target/trx_toolkit/ctrl_cmd.py
@@ -26,6 +26,7 @@
from copyright import print_copyright
CR_HOLDERS = [("2017-2018", "Vadim Yanitskiy <[email protected]>")]
+import logging as log
import signal
import getopt
import select
@@ -48,12 +49,16 @@
# Set up signal handlers
signal.signal(signal.SIGINT, self.sig_handler)
+ # Configure logging
+ log.basicConfig(level = log.DEBUG,
+ format = "[%(levelname)s] %(filename)s:%(lineno)d
%(message)s")
+
# Init UDP connection
self.ctrl_link = UDPLink(self.remote_addr, self.base_port + 1,
self.bind_addr, self.bind_port)
# Debug print
- print("[i] Init CTRL interface (%s)" \
+ log.info("Init CTRL interface (%s)" \
% self.ctrl_link.desc_link())
def print_help(self, msg = None):
@@ -138,7 +143,7 @@
sys.stdout.flush()
def sig_handler(self, signum, frame):
- print("\n\nSignal %d received" % signum)
+ log.info("Signal %d received" % signum)
if signum is signal.SIGINT:
sys.exit(0)
diff --git a/src/target/trx_toolkit/ctrl_if.py
b/src/target/trx_toolkit/ctrl_if.py
index 1e569a6..b533746 100644
--- a/src/target/trx_toolkit/ctrl_if.py
+++ b/src/target/trx_toolkit/ctrl_if.py
@@ -22,12 +22,14 @@
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
+import logging as log
+
from udp_link import UDPLink
class CTRLInterface(UDPLink):
def handle_rx(self, data, remote):
if not self.verify_req(data):
- print("[!] Wrong data on CTRL interface")
+ log.error("Wrong data on CTRL interface")
return
# Attempt to parse a command
diff --git a/src/target/trx_toolkit/ctrl_if_bb.py
b/src/target/trx_toolkit/ctrl_if_bb.py
index 97a3d9d..aaa12f1 100644
--- a/src/target/trx_toolkit/ctrl_if_bb.py
+++ b/src/target/trx_toolkit/ctrl_if_bb.py
@@ -22,6 +22,8 @@
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
+import logging as log
+
from ctrl_if import CTRLInterface
class CTRLInterfaceBB(CTRLInterface):
@@ -34,37 +36,37 @@
def __init__(self, remote_addr, remote_port, bind_addr, bind_port):
CTRLInterface.__init__(self, remote_addr, remote_port,
bind_addr, bind_port)
- print("[i] Init CTRL interface for BB (%s)" % self.desc_link())
+ log.info("Init CTRL interface for BB (%s)" % self.desc_link())
def parse_cmd(self, request):
# Power control
if self.verify_cmd(request, "POWERON", 0):
- print("[i] Recv POWERON CMD")
+ log.debug("Recv POWERON CMD")
# Ensure transceiver isn't working
if self.trx_started:
- print("[!] Transceiver already started")
+ log.error("Transceiver already started")
return -1
# Ensure RX / TX freq. are set
if (self.rx_freq is None) or (self.tx_freq is None):
- print("[!] RX / TX freq. are not set")
+ log.error("RX / TX freq. are not set")
return -1
- print("[i] Starting transceiver...")
+ log.info("Starting transceiver...")
self.trx_started = True
return 0
elif self.verify_cmd(request, "POWEROFF", 0):
- print("[i] Recv POWEROFF cmd")
+ log.debug("Recv POWEROFF cmd")
- print("[i] Stopping transceiver...")
+ log.info("Stopping transceiver...")
self.trx_started = False
return 0
# Tuning Control
elif self.verify_cmd(request, "RXTUNE", 1):
- print("[i] Recv RXTUNE cmd")
+ log.debug("Recv RXTUNE cmd")
# TODO: check freq range
self.rx_freq = int(request[1]) * 1000
@@ -72,7 +74,7 @@
return 0
elif self.verify_cmd(request, "TXTUNE", 1):
- print("[i] Recv TXTUNE cmd")
+ log.debug("Recv TXTUNE cmd")
# TODO: check freq range
self.tx_freq = int(request[1]) * 1000
@@ -80,7 +82,7 @@
# Power measurement
elif self.verify_cmd(request, "MEASURE", 1):
- print("[i] Recv MEASURE cmd")
+ log.debug("Recv MEASURE cmd")
if self.pm is None:
return -1
@@ -92,7 +94,7 @@
return (0, [meas_dbm])
elif self.verify_cmd(request, "SETSLOT", 2):
- print("[i] Recv SETSLOT cmd")
+ log.debug("Recv SETSLOT cmd")
if self.burst_fwd is None:
return -1
@@ -100,7 +102,7 @@
# Obtain TS index
ts = int(request[1])
if ts not in range(0, 8):
- print("[!] TS index should be in range: 0..7")
+ log.error("TS index should be in range: 0..7")
return -1
# Parse TS type
@@ -113,7 +115,7 @@
if ts in self.burst_fwd.ts_pass_list:
self.burst_fwd.ts_pass_list.remove(ts)
else:
- print("[!] TS %u was not activated
before" % ts)
+ log.warning("TS %u was not activated
before" % ts)
# TODO: uncomment as soon as RESET is
introduced
# return -1
else:
@@ -121,7 +123,7 @@
if ts not in self.burst_fwd.ts_pass_list:
self.burst_fwd.ts_pass_list.append(ts)
else:
- print("[!] TS %u was already activated
before" % ts)
+ log.warning("TS %u was already
activated before" % ts)
# TODO: uncomment as soon as RESET is
introduced
# return -1
@@ -129,7 +131,7 @@
# Timing Advance
elif self.verify_cmd(request, "SETTA", 1):
- print("[i] Recv SETTA cmd")
+ log.debug("Recv SETTA cmd")
# Save to the BurstForwarder instance
self.burst_fwd.ta = ta
@@ -138,7 +140,7 @@
# Timing of Arrival simulation for Uplink
# Absolute form: CMD FAKE_TOA <BASE> <THRESH>
elif self.verify_cmd(request, "FAKE_TOA", 2):
- print("[i] Recv FAKE_TOA cmd")
+ log.debug("Recv FAKE_TOA cmd")
# Parse and apply both base and threshold
self.burst_fwd.toa256_ul_base = int(request[1])
@@ -149,7 +151,7 @@
# Timing of Arrival simulation for Uplink
# Relative form: CMD FAKE_TOA <+-BASE_DELTA>
elif self.verify_cmd(request, "FAKE_TOA", 1):
- print("[i] Recv FAKE_TOA cmd")
+ log.debug("Recv FAKE_TOA cmd")
# Parse and apply delta
self.burst_fwd.toa256_ul_base += int(request[1])
@@ -159,7 +161,7 @@
# RSSI simulation for Uplink
# Absolute form: CMD FAKE_RSSI <BASE> <THRESH>
elif self.verify_cmd(request, "FAKE_RSSI", 2):
- print("[i] Recv FAKE_RSSI cmd")
+ log.debug("Recv FAKE_RSSI cmd")
# Parse and apply both base and threshold
self.burst_fwd.rssi_ul_base = int(request[1])
@@ -170,7 +172,7 @@
# RSSI simulation for Uplink
# Relative form: CMD FAKE_RSSI <+-BASE_DELTA>
elif self.verify_cmd(request, "FAKE_RSSI", 1):
- print("[i] Recv FAKE_RSSI cmd")
+ log.debug("Recv FAKE_RSSI cmd")
# Parse and apply delta
self.burst_fwd.rssi_ul_base += int(request[1])
@@ -181,12 +183,12 @@
# Syntax: CMD FAKE_DROP <AMOUNT>
# Dropping pattern: fn % 1 == 0
elif self.verify_cmd(request, "FAKE_DROP", 1):
- print("[i] Recv FAKE_DROP cmd")
+ log.debug("Recv FAKE_DROP cmd")
# Parse / validate amount of bursts
num = int(request[1])
if num < 0:
- print("[!] FAKE_DROP amount shall not be
negative")
+ log.error("FAKE_DROP amount shall not be
negative")
return -1
self.burst_fwd.burst_ul_drop_amount = num
@@ -198,18 +200,18 @@
# Syntax: CMD FAKE_DROP <AMOUNT> <FN_PERIOD>
# Dropping pattern: fn % period == 0
elif self.verify_cmd(request, "FAKE_DROP", 2):
- print("[i] Recv FAKE_DROP cmd")
+ log.debug("Recv FAKE_DROP cmd")
# Parse / validate amount of bursts
num = int(request[1])
if num < 0:
- print("[!] FAKE_DROP amount shall not be
negative")
+ log.error("FAKE_DROP amount shall not be
negative")
return -1
# Parse / validate period
period = int(request[2])
if period <= 0:
- print("[!] FAKE_DROP period shall be greater
than zero")
+ log.error("FAKE_DROP period shall be greater
than zero")
return -1
self.burst_fwd.burst_ul_drop_amount = num
@@ -221,5 +223,5 @@
else:
# We don't care about other commands,
# so let's merely ignore them ;)
- print("[i] Ignore CMD %s" % request[0])
+ log.debug("Ignore CMD %s" % request[0])
return 0
diff --git a/src/target/trx_toolkit/ctrl_if_bts.py
b/src/target/trx_toolkit/ctrl_if_bts.py
index 6ac8ffb..2dde3e3 100644
--- a/src/target/trx_toolkit/ctrl_if_bts.py
+++ b/src/target/trx_toolkit/ctrl_if_bts.py
@@ -22,6 +22,8 @@
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
+import logging as log
+
from ctrl_if import CTRLInterface
class CTRLInterfaceBTS(CTRLInterface):
@@ -35,24 +37,24 @@
def __init__(self, remote_addr, remote_port, bind_addr, bind_port):
CTRLInterface.__init__(self, remote_addr, remote_port,
bind_addr, bind_port)
- print("[i] Init CTRL interface for BTS (%s)" % self.desc_link())
+ log.info("Init CTRL interface for BTS (%s)" % self.desc_link())
def parse_cmd(self, request):
# Power control
if self.verify_cmd(request, "POWERON", 0):
- print("[i] Recv POWERON CMD")
+ log.debug("Recv POWERON CMD")
# Ensure transceiver isn't working
if self.trx_started:
- print("[!] Transceiver already started")
+ log.error("Transceiver already started")
return -1
# Ensure RX / TX freq. are set
if (self.rx_freq is None) or (self.tx_freq is None):
- print("[!] RX / TX freq. are not set")
+ log.error("RX / TX freq. are not set")
return -1
- print("[i] Starting transceiver...")
+ log.info("Starting transceiver...")
self.trx_started = True
# Power emulation
@@ -66,9 +68,9 @@
return 0
elif self.verify_cmd(request, "POWEROFF", 0):
- print("[i] Recv POWEROFF cmd")
+ log.debug("Recv POWEROFF cmd")
- print("[i] Stopping transceiver...")
+ log.info("Stopping transceiver...")
self.trx_started = False
# Power emulation
@@ -83,14 +85,14 @@
# Tuning Control
elif self.verify_cmd(request, "RXTUNE", 1):
- print("[i] Recv RXTUNE cmd")
+ log.debug("Recv RXTUNE cmd")
# TODO: check freq range
self.rx_freq = int(request[1]) * 1000
return 0
elif self.verify_cmd(request, "TXTUNE", 1):
- print("[i] Recv TXTUNE cmd")
+ log.debug("Recv TXTUNE cmd")
# TODO: check freq range
self.tx_freq = int(request[1]) * 1000
@@ -100,7 +102,7 @@
# Timing of Arrival simulation for Downlink
# Absolute form: CMD FAKE_TOA <BASE> <THRESH>
elif self.verify_cmd(request, "FAKE_TOA", 2):
- print("[i] Recv FAKE_TOA cmd")
+ log.debug("Recv FAKE_TOA cmd")
# Parse and apply both base and threshold
self.burst_fwd.toa256_dl_base = int(request[1])
@@ -111,7 +113,7 @@
# Timing of Arrival simulation for Downlink
# Relative form: CMD FAKE_TOA <+-BASE_DELTA>
elif self.verify_cmd(request, "FAKE_TOA", 1):
- print("[i] Recv FAKE_TOA cmd")
+ log.debug("Recv FAKE_TOA cmd")
# Parse and apply delta
self.burst_fwd.toa256_dl_base += int(request[1])
@@ -121,7 +123,7 @@
# RSSI simulation for Downlink
# Absolute form: CMD FAKE_RSSI <BASE> <THRESH>
elif self.verify_cmd(request, "FAKE_RSSI", 2):
- print("[i] Recv FAKE_RSSI cmd")
+ log.debug("Recv FAKE_RSSI cmd")
# Parse and apply both base and threshold
self.burst_fwd.rssi_dl_base = int(request[1])
@@ -132,7 +134,7 @@
# RSSI simulation for Downlink
# Relative form: CMD FAKE_RSSI <+-BASE_DELTA>
elif self.verify_cmd(request, "FAKE_RSSI", 1):
- print("[i] Recv FAKE_RSSI cmd")
+ log.debug("Recv FAKE_RSSI cmd")
# Parse and apply delta
self.burst_fwd.rssi_dl_base += int(request[1])
@@ -143,12 +145,12 @@
# Syntax: CMD FAKE_DROP <AMOUNT>
# Dropping pattern: fn % 1 == 0
elif self.verify_cmd(request, "FAKE_DROP", 1):
- print("[i] Recv FAKE_DROP cmd")
+ log.debug("Recv FAKE_DROP cmd")
# Parse / validate amount of bursts
num = int(request[1])
if num < 0:
- print("[!] FAKE_DROP amount shall not be
negative")
+ log.error("FAKE_DROP amount shall not be
negative")
return -1
self.burst_fwd.burst_dl_drop_amount = num
@@ -160,18 +162,18 @@
# Syntax: CMD FAKE_DROP <AMOUNT> <FN_PERIOD>
# Dropping pattern: fn % period == 0
elif self.verify_cmd(request, "FAKE_DROP", 2):
- print("[i] Recv FAKE_DROP cmd")
+ log.debug("Recv FAKE_DROP cmd")
# Parse / validate amount of bursts
num = int(request[1])
if num < 0:
- print("[!] FAKE_DROP amount shall not be
negative")
+ log.error("FAKE_DROP amount shall not be
negative")
return -1
# Parse / validate period
period = int(request[2])
if period <= 0:
- print("[!] FAKE_DROP period shall be greater
than zero")
+ log.error("FAKE_DROP period shall be greater
than zero")
return -1
self.burst_fwd.burst_dl_drop_amount = num
@@ -183,5 +185,5 @@
else:
# We don't care about other commands,
# so let's merely ignore them ;)
- print("[i] Ignore CMD %s" % request[0])
+ log.debug("Ignore CMD %s" % request[0])
return 0
diff --git a/src/target/trx_toolkit/data_dump.py
b/src/target/trx_toolkit/data_dump.py
index 1d7805e..71e1261 100644
--- a/src/target/trx_toolkit/data_dump.py
+++ b/src/target/trx_toolkit/data_dump.py
@@ -22,6 +22,7 @@
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
+import logging as log
import struct
from data_msg import *
@@ -77,13 +78,15 @@
def __init__(self, capture):
# Check if capture file is already opened
if isinstance(capture, str):
- print("[i] Opening capture file '%s'..." % capture)
+ log.info("Opening capture file '%s'..." % capture)
self.f = open(capture, "a+b")
else:
self.f = capture
def __del__(self):
- print("[i] Closing the capture file")
+ # FIXME: this causes an Exception in Python 2 (but not in
Python 3)
+ # AttributeError: 'NoneType' object has no attribute 'info'
+ log.info("Closing the capture file")
self.f.close()
# Moves the file descriptor before a specified message
@@ -104,7 +107,7 @@
# Attempt to parse it
rc = self.parse_hdr(hdr_raw)
if rc is False:
- print("[!] Couldn't parse a message header")
+ log.error("Couldn't parse a message header")
return False
# Expand the header
@@ -129,7 +132,7 @@
# Attempt to parse it
rc = self.parse_hdr(hdr_raw)
if rc is False:
- print("[!] Couldn't parse a message header")
+ log.error("Couldn't parse a message header")
return None
# Expand the header
@@ -138,7 +141,7 @@
# Attempt to read a message
msg_raw = self.f.read(msg_len)
if len(msg_raw) != msg_len:
- print("[!] Message length mismatch")
+ log.error("Message length mismatch")
return None
# Attempt to parse a message
@@ -146,7 +149,7 @@
msg_raw = bytearray(msg_raw)
msg.parse_msg(msg_raw)
except:
- print("[!] Couldn't parse a message, skipping...")
+ log.error("Couldn't parse a message, skipping...")
return False
# Success
@@ -161,7 +164,7 @@
# Move descriptor to the begining of requested message
rc = self._seek2msg(idx)
if not rc:
- print("[!] Couldn't find requested message")
+ log.error("Couldn't find requested message")
return False
# Attempt to parse a message
@@ -181,7 +184,7 @@
else:
rc = self._seek2msg(skip)
if not rc:
- print("[!] Couldn't find requested message")
+ log.error("Couldn't find requested message")
return False
# Read the capture in loop...
@@ -224,6 +227,10 @@
from gsm_shared import *
import random
+ # Configure logging
+ log.basicConfig(level = log.DEBUG,
+ format = "[%(levelname)s] %(filename)s:%(lineno)d %(message)s")
+
# Create a temporary file
tf = TemporaryFile()
@@ -242,7 +249,7 @@
burst_trx2l1.append(sbit)
# Generate a basic list of random messages
- print("[i] Generating the reference messages")
+ log.info("Generating the reference messages")
messages_ref = []
for i in range(100):
@@ -260,9 +267,9 @@
# Append
messages_ref.append(msg)
- print("[i] Adding the following messages to the capture:")
+ log.info("Adding the following messages to the capture:")
for msg in messages_ref[:3]:
- print(" %s: burst_len=%d"
+ log.info("%s: burst_len=%d"
% (msg.desc_hdr(), len(msg.burst)))
# Check single message appending
@@ -273,9 +280,9 @@
# Read the written messages back
messages_check = ddf.parse_all()
- print("[i] Read the following messages back:")
+ log.info("Read the following messages back:")
for msg in messages_check:
- print(" %s: burst_len=%d"
+ log.info("%s: burst_len=%d"
% (msg.desc_hdr(), len(msg.burst)))
# Expecting three messages
@@ -291,7 +298,7 @@
# Validate a message
assert(messages_check[i].validate())
- print("[?] Check append_msg(): OK")
+ log.info("Check append_msg(): OK")
# Append the pending reference messages
@@ -313,7 +320,7 @@
# Validate a message
assert(messages_check[i].validate())
- print("[?] Check append_all(): OK")
+ log.info("Check append_all(): OK")
# Check parse_msg()
@@ -336,7 +343,7 @@
assert(msg0.validate())
assert(msg10.validate())
- print("[?] Check parse_msg(): OK")
+ log.info("Check parse_msg(): OK")
# Check parse_all() with range
@@ -357,4 +364,4 @@
# Validate a message
assert(messages_check[i].validate())
- print("[?] Check parse_all(): OK")
+ log.info("Check parse_all(): OK")
diff --git a/src/target/trx_toolkit/data_msg.py
b/src/target/trx_toolkit/data_msg.py
index ea415ab..95ec9dc 100644
--- a/src/target/trx_toolkit/data_msg.py
+++ b/src/target/trx_toolkit/data_msg.py
@@ -431,6 +431,12 @@
# Regression test
if __name__ == '__main__':
+ import logging as log
+
+ # Configure logging
+ log.basicConfig(level = log.DEBUG,
+ format = "[%(levelname)s] %(filename)s:%(lineno)d %(message)s")
+
# Common reference data
fn = 1024
tn = 0
@@ -446,7 +452,7 @@
sbit = random.randint(-127, 127)
burst_trx2l1_ref.append(sbit)
- print("[i] Generating the reference messages")
+ log.info("Generating the reference messages")
# Create messages of both types
msg_l12trx_ref = DATAMSG_L12TRX(fn = fn, tn = tn)
@@ -461,13 +467,13 @@
msg_l12trx_ref.burst = burst_l12trx_ref
msg_trx2l1_ref.burst = burst_trx2l1_ref
- print("[i] Encoding the reference messages")
+ log.info("Encoding the reference messages")
# Encode DATA messages
l12trx_raw = msg_l12trx_ref.gen_msg()
trx2l1_raw = msg_trx2l1_ref.gen_msg()
- print("[i] Parsing generated messages back")
+ log.info("Parsing generated messages back")
# Parse generated DATA messages
msg_l12trx_dec = DATAMSG_L12TRX()
@@ -475,13 +481,13 @@
msg_l12trx_dec.parse_msg(l12trx_raw)
msg_trx2l1_dec.parse_msg(trx2l1_raw)
- print("[i] Comparing decoded messages with the reference")
+ log.info("Comparing decoded messages with the reference")
# Compare bursts
assert(msg_l12trx_dec.burst == burst_l12trx_ref)
assert(msg_trx2l1_dec.burst == burst_trx2l1_ref)
- print("[?] Compare bursts: OK")
+ log.info("Compare bursts: OK")
# Compare both parsed messages with the reference data
assert(msg_l12trx_dec.fn == fn)
@@ -489,14 +495,14 @@
assert(msg_l12trx_dec.tn == tn)
assert(msg_trx2l1_dec.tn == tn)
- print("[?] Compare FN / TN: OK")
+ log.info("Compare FN / TN: OK")
# Compare message specific parts
assert(msg_trx2l1_dec.rssi == msg_trx2l1_ref.rssi)
assert(msg_l12trx_dec.pwr == msg_l12trx_ref.pwr)
assert(msg_trx2l1_dec.toa256 == msg_trx2l1_ref.toa256)
- print("[?] Compare message specific data: OK")
+ log.info("Compare message specific data: OK")
# Validate header randomization
for i in range(0, 100):
@@ -506,7 +512,7 @@
assert(msg_l12trx_ref.validate())
assert(msg_trx2l1_ref.validate())
- print("[?] Validate header randomization: OK")
+ log.info("Validate header randomization: OK")
# Bit conversation test
usbits_ref = list(range(0, 256))
@@ -518,7 +524,7 @@
assert(usbits[:255] == usbits_ref[:255])
assert(usbits[255] == 254)
- print("[?] Check both usbit2sbit() and sbit2usbit(): OK")
+ log.info("Check both usbit2sbit() and sbit2usbit(): OK")
# Test both sbit2ubit() and ubit2sbit()
ubits = msg_trx2l1_ref.sbit2ubit(sbits_ref)
@@ -527,7 +533,7 @@
sbits = msg_trx2l1_ref.ubit2sbit(ubits)
assert(sbits == ([-127] * 127 + [127] * 128))
- print("[?] Check both sbit2ubit() and ubit2sbit(): OK")
+ log.info("Check both sbit2ubit() and ubit2sbit(): OK")
# Test message transformation
msg_l12trx_dec = msg_trx2l1_ref.gen_l12trx()
@@ -542,4 +548,4 @@
assert(msg_l12trx_dec.burst ==
msg_l12trx_dec.sbit2ubit(burst_trx2l1_ref))
assert(msg_trx2l1_dec.burst ==
msg_trx2l1_dec.ubit2sbit(burst_l12trx_ref))
- print("[?] Check L12TRX <-> TRX2L1 type transformations: OK")
+ log.info("Check L12TRX <-> TRX2L1 type transformations: OK")
diff --git a/src/target/trx_toolkit/fake_trx.py
b/src/target/trx_toolkit/fake_trx.py
index d99186b..ced1238 100755
--- a/src/target/trx_toolkit/fake_trx.py
+++ b/src/target/trx_toolkit/fake_trx.py
@@ -25,6 +25,7 @@
from copyright import print_copyright
CR_HOLDERS = [("2017-2018", "Vadim Yanitskiy <[email protected]>")]
+import logging as log
import signal
import getopt
import select
@@ -53,6 +54,10 @@
# Set up signal handlers
signal.signal(signal.SIGINT, self.sig_handler)
+ # Configure logging
+ log.basicConfig(level = log.DEBUG,
+ format = "[%(levelname)s] %(filename)s:%(lineno)d
%(message)s")
+
def run(self):
# Init TRX CTRL interface for BTS
self.bts_ctrl = CTRLInterfaceBTS(self.bts_addr,
self.bts_base_port + 101,
@@ -90,7 +95,7 @@
self.clck_gen = CLCKGen([self.bts_clck])
self.bts_ctrl.clck_gen = self.clck_gen
- print("[i] Init complete")
+ log.info("Init complete")
# Enter main loop
while True:
@@ -119,7 +124,7 @@
self.bb_ctrl.handle_rx(data.decode(), addr)
def shutdown(self):
- print("[i] Shutting down...")
+ log.info("Shutting down...")
# Stop clock generator
self.clck_gen.stop()
@@ -198,7 +203,7 @@
sys.exit(2)
def sig_handler(self, signum, frame):
- print("Signal %d received" % signum)
+ log.info("Signal %d received" % signum)
if signum is signal.SIGINT:
self.shutdown()
sys.exit(0)
diff --git a/src/target/trx_toolkit/trx_sniff.py
b/src/target/trx_toolkit/trx_sniff.py
index 535bb3f..9850983 100755
--- a/src/target/trx_toolkit/trx_sniff.py
+++ b/src/target/trx_toolkit/trx_sniff.py
@@ -25,6 +25,7 @@
from copyright import print_copyright
CR_HOLDERS = [("2018", "Vadim Yanitskiy <[email protected]>")]
+import logging as log
import signal
import getopt
import sys
@@ -67,6 +68,10 @@
print_copyright(CR_HOLDERS)
self.parse_argv()
+ # Configure logging
+ log.basicConfig(level = log.DEBUG,
+ format = "[%(levelname)s] %(filename)s:%(lineno)d
%(message)s")
+
# Open requested capture file
if self.output_file is not None:
self.ddf = DATADumpFile(self.output_file)
@@ -76,7 +81,7 @@
pkt_filter = "udp and (port %d or port %d)" \
% (self.sniff_base_port + 2, self.sniff_base_port + 102)
- print("[i] Listening on interface '%s'..." %
self.sniff_interface)
+ log.info("Listening on interface '%s'..." %
self.sniff_interface)
# Start sniffing...
scapy.all.sniff(iface = self.sniff_interface, store = 0,
@@ -110,7 +115,7 @@
try:
msg.parse_msg(msg_raw)
except:
- print("[!] Failed to parse message, dropping...")
+ log.warning("Failed to parse message, dropping...")
self.cnt_burst_dropped_num += 1
return
@@ -121,7 +126,7 @@
return
# Debug print
- print("[i] %s burst: %s" \
+ log.debug("%s burst: %s" \
% ("L1 -> TRX" if l12trx else "TRX -> L1",
msg.desc_hdr()))
# Poke message handler
@@ -177,22 +182,22 @@
# Stop sniffing after N bursts
if self.cnt_burst_break is not None:
if self.cnt_burst_num == self.cnt_burst_break:
- print("[i] Collected required amount of bursts")
+ log.info("Collected required amount of bursts")
return True
# Stop sniffing after N frames
if self.cnt_frame_break is not None:
if self.cnt_frame_num == self.cnt_frame_break:
- print("[i] Collected required amount of frames")
+ log.info("Collected required amount of frames")
return True
return False
def shutdown(self):
- print("[i] Shutting down...")
+ log.info("Shutting down...")
# Print statistics
- print("[i] %u bursts handled, %u dropped" \
+ log.info("%u bursts handled, %u dropped" \
% (self.cnt_burst_num, self.cnt_burst_dropped_num))
# Exit
--
To view, visit https://gerrit.osmocom.org/12185
To unsubscribe, or for help writing mail filters, visit
https://gerrit.osmocom.org/settings
Gerrit-Project: osmocom-bb
Gerrit-Branch: master
Gerrit-MessageType: newchange
Gerrit-Change-Id: I7fcafabafe8323b58990997a47afdd48b6d1f357
Gerrit-Change-Number: 12185
Gerrit-PatchSet: 1
Gerrit-Owner: Vadim Yanitskiy <[email protected]>
