From be53a25682cbe1c0868e19a0d0a866062367a65d Mon Sep 17 00:00:00 2001 From: Mark Houtz Date: Wed, 1 Jul 2026 15:08:11 -0600 Subject: [PATCH 1/5] feat: add --cell-kv machine-readable cell measurement output Add a --cell-kv flag (Qualcomm) that emits LTE and 5G-NR serving and neighbor cell measurements as single-line key=value records for machine parsing, instead of the human-readable stdout. Each record has the form: pci=..,earfcn=..,earfcn_ul=..,frequency=..,protocol=lte|nr,cell=scell|ncell, plmn=..,mcc=..,mnc=..,tac=..,cid=..,band=..,bwmhzdl=..,bwmhzul=.. rssi=..,rsrp=..,rsrq=.. Details: - util: dl_earfcn_to_frequency_hz() (3GPP TS 36.101 DL band table) and nrarfcn_to_frequency_hz() (TS 38.104 global raster) give center frequency in Hz; format_cell_kv()/format_plmn() build the record; real LTE UL-EARFCN via calculate_ul_earfcn(). - LTE (diagltelogparser) and NR (diagnrlogparser) serving and neighbor cell measurements emit the record; NR beam parsing is preserved (offsets still advance) but beams are not emitted as records. - Stateful join: RRC SCell Info packets populate a per-radio serving-cell identity cache (plmn/mcc/mnc/tac/cid/band/bandwidth) that is merged onto the matching ML1 serving-cell measurement, guarded on (earfcn, pci) so stale identity is never attached to a different cell. Default (no --cell-kv) output is unchanged; existing tests pass. Co-Authored-By: Claude Opus 4.8 --- src/scat/main.py | 2 + src/scat/parsers/qualcomm/diagltelogparser.py | 36 ++++++++- src/scat/parsers/qualcomm/diagnrlogparser.py | 65 ++++++++++++---- src/scat/parsers/qualcomm/qualcommparser.py | 8 ++ src/scat/util.py | 78 +++++++++++++++++++ 5 files changed, 170 insertions(+), 19 deletions(-) diff --git a/src/scat/main.py b/src/scat/main.py index 641a687..4907583 100755 --- a/src/scat/main.py +++ b/src/scat/main.py @@ -64,6 +64,7 @@ def scat_main(): parser.add_argument('-V', '--version', action='version', version='SCAT {}'.format(__version__)) parser.add_argument('-L', '--layer', help='Specify the layers to see as GSMTAP packets (comma separated).\nAvailable layers: {}, Default: "ip,nas,rrc"'.format(', '.join(valid_layers)), type=str, default='ip,nas,rrc') parser.add_argument('-f', '--format', help='Select display format for LAC/RAC/TAC/CID: [d]ecimal, he[x]adecimal (default), [b]oth.', type=str, default='x', choices=['d', 'x', 'b']) + parser.add_argument('--cell-kv', help='Emit LTE/5G-NR serving and neighbor cell measurements as key=value lines to stdout for machine parsing.', action='store_true') parser.add_argument('-3', '--gsmtapv3', help='Enable GSMTAPv3 for 2G/3G/4G. Default: enabled only for 5G NR', action='store_true') input_group = parser.add_mutually_exclusive_group(required=True) @@ -195,6 +196,7 @@ def scat_main(): 'disable-crc-check': args.disable_crc_check, 'layer': layers, 'format': args.format, + 'cell-kv': args.cell_kv, 'gsmtapv3': args.gsmtapv3}) elif args.type == 'sec': if args.trace: diff --git a/src/scat/parsers/qualcomm/diagltelogparser.py b/src/scat/parsers/qualcomm/diagltelogparser.py index 7e877ae..be9740a 100644 --- a/src/scat/parsers/qualcomm/diagltelogparser.py +++ b/src/scat/parsers/qualcomm/diagltelogparser.py @@ -186,8 +186,16 @@ def parse_lte_ml1_scell_meas(self, pkt_header, pkt_body: bytes, args: dict): real_rssi = self.parse_rssi(meas_rssi) real_rsrq = self.parse_rsrq(meas_rsrq) - return {'stdout': 'LTE SCell: EARFCN: {}, PCI: {:3d}, Measured RSRP: {:.2f}, Measured RSSI: {:.2f}, Measured RSRQ: {:.2f}'.format(item.earfcn, pci, real_rsrp, real_rssi, real_rsrq), - 'ts': pkt_ts} + if getattr(self.parent, 'cell_kv', False): + radio_id = args['radio_id'] if (args and 'radio_id' in args) else 0 + cache = self.parent.lte_serving_cell[radio_id] if (self.parent and radio_id in (0, 1)) else {} + ident = util.serving_identity_fields(cache, item.earfcn, pci) + stdout = util.format_cell_kv('lte', 'scell', pci, item.earfcn, + util.calculate_ul_earfcn(item.earfcn), util.dl_earfcn_to_frequency_hz(item.earfcn), + rssi=real_rssi, rsrp=real_rsrp, rsrq=real_rsrq, **ident) + else: + stdout = 'LTE SCell: EARFCN: {}, PCI: {:3d}, Measured RSRP: {:.2f}, Measured RSSI: {:.2f}, Measured RSRQ: {:.2f}'.format(item.earfcn, pci, real_rsrp, real_rssi, real_rsrq) + return {'stdout': stdout, 'ts': pkt_ts} def parse_lte_ml1_ncell_meas(self, pkt_header, pkt_body: bytes, args: dict): pkt_ts = util.parse_qxdm_ts(pkt_header.timestamp) @@ -218,7 +226,12 @@ def parse_lte_ml1_ncell_meas(self, pkt_header, pkt_body: bytes, args: dict): q_rxlevmin = item.q_rxlevmin_n_cells & 0x3f n_cells = item.q_rxlevmin_n_cells >> 6 - stdout += 'LTE NCell: EARFCN: {}, number of cells: {}\n'.format(item.earfcn, n_cells) + cell_kv = getattr(self.parent, 'cell_kv', False) + n_earfcn_ul = util.calculate_ul_earfcn(item.earfcn) + n_frequency = util.dl_earfcn_to_frequency_hz(item.earfcn) + kv_lines = [] + if not cell_kv: + stdout += 'LTE NCell: EARFCN: {}, number of cells: {}\n'.format(item.earfcn, n_cells) for i in range(n_cells): n_cell_pkt = pkt_body[pos + 32 * i:pos + 32 * (i + 1)] @@ -251,7 +264,13 @@ def parse_lte_ml1_ncell_meas(self, pkt_header, pkt_body: bytes, args: dict): n_real_rssi = self.parse_rssi(n_meas_rssi) n_real_rsrq = self.parse_rsrq(n_meas_rsrq) - stdout += '└── Neighbor cell {}: PCI: {:3d}, RSRP: {:.2f}, RSSI: {:.2f}, RSRQ: {:.2f}\n'.format(i, n_pci, n_real_rsrp, n_real_rssi, n_real_rsrq) + if cell_kv: + kv_lines.append(util.format_cell_kv('lte', 'ncell', n_pci, item.earfcn, + n_earfcn_ul, n_frequency, rssi=n_real_rssi, rsrp=n_real_rsrp, rsrq=n_real_rsrq)) + else: + stdout += '└── Neighbor cell {}: PCI: {:3d}, RSRP: {:.2f}, RSSI: {:.2f}, RSRQ: {:.2f}\n'.format(i, n_pci, n_real_rsrp, n_real_rssi, n_real_rsrq) + if cell_kv: + return {'stdout': '\n'.join(kv_lines), 'ts': pkt_ts} return {'stdout': stdout.rstrip(), 'ts': pkt_ts} def parse_lte_ml1_scell_meas_response_cell_v36(self, cell_id: int, cell_bytes: bytes, rsrp_offset: int=16, snr_offset: int=80, sir_cinr_offset: int=104): @@ -1242,6 +1261,15 @@ def parse_lte_rrc_cell_info(self, pkt_header, pkt_body: bytes, args: dict): self.parent.lte_last_earfcn_ul[radio_id] = item.ul_earfcn self.parent.lte_last_bw_dl[radio_id] = item.dl_bw self.parent.lte_last_bw_ul[radio_id] = item.ul_bw + if radio_id in (0, 1): + self.parent.lte_serving_cell[radio_id] = { + 'pci': item.pci, 'earfcn': item.dl_earfcn, + 'plmn': util.format_plmn(item.mcc, item.mnc, item.mnc_digit), + 'mcc': item.mcc, 'mnc': item.mnc, 'tac': item.tac, + 'cid': item.cell_id, 'band': item.band, + 'bwmhzdl': prb_to_mhz.get(item.dl_bw, 0), + 'bwmhzul': prb_to_mhz.get(item.ul_bw, 0), + } bw_str = '' if item.dl_bw in prb_to_mhz and item.ul_bw in prb_to_mhz: diff --git a/src/scat/parsers/qualcomm/diagnrlogparser.py b/src/scat/parsers/qualcomm/diagnrlogparser.py index 0c4e8f3..8ba498b 100644 --- a/src/scat/parsers/qualcomm/diagnrlogparser.py +++ b/src/scat/parsers/qualcomm/diagnrlogparser.py @@ -73,6 +73,10 @@ def parse_float_q7(self, data_to_convert: int) -> float: # ML1 def parse_nr_ml1_meas_db_update(self, pkt_header, pkt_body: bytes, args: dict): stdout = '' + cell_kv = getattr(self.parent, 'cell_kv', False) + kv_lines = [] + radio_id = args['radio_id'] if (args and 'radio_id' in args) else 0 + nr_cache = self.parent.nr_serving_cell[radio_id] if (self.parent and cell_kv and radio_id in (0, 1)) else {} pkt_ver = self.nr_pkt_ver._make(struct.unpack(' 0x00 and meas_carrier_list.serv_cell_index < 0xff: @@ -156,17 +171,24 @@ def parse_nr_ml1_meas_db_update(self, pkt_header, pkt_body: bytes, args: dict): cell_list_struct = namedtuple('QcDiagNrMl1Packet', 'pci pbch_sfn num_beams null_0 cell_quality_rsrp cell_quality_rsrq') cell_list = cell_list_struct._make(struct.unpack(' None: self.parse_msgs = params[p] elif p == 'cacombos': self.cacombos = params[p] + elif p == 'cell-kv': + self.cell_kv = params[p] elif p == 'combine-stdout': self.combine_stdout = params[p] elif p == 'disable-crc-check': diff --git a/src/scat/util.py b/src/scat/util.py index 17d8f5c..1031825 100644 --- a/src/scat/util.py +++ b/src/scat/util.py @@ -717,6 +717,84 @@ def calculate_dl_earfcn(ul_earfcn: int) -> int: offset = 0 return ul_earfcn - offset +# E-UTRA DL band table: (ndl_low, ndl_high, fdl_low_mhz). +# NOffs-DL equals ndl_low for every band, and the DL raster is always 0.1 MHz, +# so FDL = fdl_low + 0.1 * (earfcn - ndl_low). Source: 3GPP TS 36.101 Table 5.7.3-1. +_LTE_DL_EARFCN_BANDS = ( + (0, 599, 2110.0), (600, 1199, 1930.0), (1200, 1949, 1805.0), + (1950, 2399, 2110.0), (2400, 2649, 869.0), (2650, 2749, 875.0), + (2750, 3449, 2620.0), (3450, 3799, 925.0), (3800, 4149, 1844.9), + (4150, 4749, 2110.0), (4750, 4949, 1475.9), (5010, 5179, 729.0), + (5180, 5279, 746.0), (5280, 5379, 758.0), (5730, 5849, 734.0), + (5850, 5999, 860.0), (6000, 6149, 875.0), (6150, 6449, 791.0), + (6450, 6599, 1495.9), (6600, 7399, 3510.0), (7500, 7699, 2180.0), + (7700, 8039, 1525.0), (8040, 8689, 1930.0), (8690, 9039, 859.0), + (9040, 9209, 852.0), (9210, 9659, 758.0), (9660, 9769, 717.0), + (9770, 9869, 2350.0), (9870, 9919, 462.5), (9920, 10359, 1452.0), + (36000, 36199, 1900.0), (36200, 36349, 2010.0), (36350, 36949, 1850.0), + (36950, 37549, 1930.0), (37550, 37749, 1910.0), (37750, 38249, 2570.0), + (38250, 38649, 1880.0), (38650, 39649, 2300.0), (39650, 41589, 2496.0), + (41590, 43589, 3400.0), (43590, 45589, 3600.0), (45590, 46589, 703.0), + (46590, 46789, 1447.0), (46790, 54539, 5150.0), (54540, 55239, 5855.0), + (55240, 56739, 3550.0), (56740, 58239, 3550.0), (58240, 59089, 1432.0), + (59090, 59139, 1427.0), (59140, 60139, 3300.0), (60140, 60254, 2483.5), + (65536, 66435, 2110.0), (66436, 67335, 2110.0), (67336, 67535, 738.0), + (67536, 67835, 753.0), (67836, 68335, 2570.0), (68336, 68585, 1995.0), + (68586, 68935, 617.0), (68936, 68985, 461.0), (68986, 69035, 460.0), + (69036, 69465, 1475.0), (69466, 70315, 1432.0), (70316, 70365, 1427.0), + (70366, 70545, 728.0), (70546, 70595, 420.0), (70596, 70645, 422.0), +) + +# Returns the DL center frequency in Hz for a DL-EARFCN, or 0 if out of range. +def dl_earfcn_to_frequency_hz(dl_earfcn: int) -> int: + for ndl_low, ndl_high, fdl_low in _LTE_DL_EARFCN_BANDS: + if ndl_low <= dl_earfcn <= ndl_high: + freq_mhz = round(fdl_low + 0.1 * (dl_earfcn - ndl_low), 1) + return int(round(freq_mhz * 1_000_000)) + return 0 + +# Returns the center frequency in Hz for an NR-ARFCN (3GPP TS 38.104 Table 5.4.2.1-1 +# global frequency raster), or 0 if out of range. +def nrarfcn_to_frequency_hz(nrarfcn: int) -> int: + if 0 <= nrarfcn < 600000: # 0 - 3000 MHz, dF_global = 5 kHz + return 5000 * nrarfcn + elif 600000 <= nrarfcn < 2016667: # 3000 - 24250 MHz, dF_global = 15 kHz + return 3_000_000_000 + 15000 * (nrarfcn - 600000) + elif 2016667 <= nrarfcn <= 3279165: # 24250 - 100000 MHz, dF_global = 60 kHz + return 24_250_080_000 + 60000 * (nrarfcn - 2016667) + return 0 + +# Formats a single serving/neighbor cell measurement as a key=value line for +# machine parsing (enabled by SCAT's --cell-kv flag). Identity fields not present +# in ML1 measurement packets (plmn/mcc/mnc/tac/cid/band/bandwidth) default to 0. +def format_cell_kv(protocol: str, cell: str, pci, earfcn, earfcn_ul, frequency, + rssi=0, rsrp=0, rsrq=0, plmn=0, mcc=0, mnc=0, tac=0, cid=0, + band=0, bwmhzdl=0, bwmhzul=0) -> str: + return ('pci={},earfcn={},earfcn_ul={},frequency={},protocol={},cell={},' + 'plmn={},mcc={},mnc={},tac={},cid={},band={},bwmhzdl={},bwmhzul={} ' + 'rssi={},rsrp={},rsrq={}').format(pci, earfcn, earfcn_ul, frequency, + protocol, cell, plmn, mcc, mnc, tac, cid, band, bwmhzdl, bwmhzul, + rssi, rsrp, rsrq) + +# Combined PLMN identifier (3-digit MCC + 2/3-digit MNC), e.g. mcc=310,mnc=260 -> '310260'. +# Returns '0' if the MCC/MNC are missing or not numeric. +def format_plmn(mcc, mnc, mnc_digit=0) -> str: + try: + mcc_i, mnc_i = int(mcc), int(mnc) + except (ValueError, TypeError): + return '0' + width = mnc_digit if mnc_digit in (2, 3) else (3 if mnc_i > 99 else 2) + return '{:03d}{:0{}d}'.format(mcc_i, mnc_i, width) + +# Returns the identity kwargs (plmn/mcc/mnc/tac/cid/band/bwmhzdl/bwmhzul) from a +# cached serving-cell entry, but only if it matches the given (earfcn, pci) — so +# stale identity is never joined onto a different cell. Empty dict otherwise. +def serving_identity_fields(cache_entry: dict, earfcn: int, pci: int) -> dict: + if not cache_entry or cache_entry.get('earfcn') != earfcn or cache_entry.get('pci') != pci: + return {} + return {k: cache_entry[k] for k in ('plmn', 'mcc', 'mnc', 'tac', 'cid', + 'band', 'bwmhzdl', 'bwmhzul') if k in cache_entry} + def convert_mcc(mcc_digit_2: int, mcc_digit_1: int, mcc_digit_0: int) -> str: if mcc_digit_2 > 0x09 or mcc_digit_1 > 0x09 or mcc_digit_0 > 0x09: raise ValueError('Invalid digit in MCC') From a3c91a794f33bb07a5d61bcdad8ca3b1999d8205 Mon Sep 17 00:00:00 2001 From: Mark Houtz Date: Wed, 1 Jul 2026 15:30:08 -0600 Subject: [PATCH 2/5] test: add coverage for --cell-kv output and RRC/ML1 join Cover the util helpers (dl_earfcn_to_frequency_hz, nrarfcn_to_frequency_hz, format_plmn, serving_identity_fields match-guard), LTE/NR serving and neighbor cell key=value records, the RRC SCell Info -> serving-cell cache population, and the stateful identity join (hit, mismatch-guard, and neighbor-not-enriched) using existing real capture vectors. Also asserts default (no --cell-kv) output is unchanged. Co-Authored-By: Claude Opus 4.8 --- tests/test_cell_kv.py | 145 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 145 insertions(+) create mode 100644 tests/test_cell_kv.py diff --git a/tests/test_cell_kv.py b/tests/test_cell_kv.py new file mode 100644 index 0000000..a3b225a --- /dev/null +++ b/tests/test_cell_kv.py @@ -0,0 +1,145 @@ +#!/usr/bin/env python3 + +import unittest +import binascii +import logging +from collections import namedtuple + +import scat.util as util +import scat.parsers.qualcomm.diagcmd as diagcmd +from scat.parsers.qualcomm.diagltelogparser import DiagLteLogParser +from scat.parsers.qualcomm.diagnrlogparser import DiagNrLogParser + + +class FakeParent: + """Minimal stand-in for QualcommParser carrying the --cell-kv flag and the + per-radio serving-cell identity caches consumed by the log parsers.""" + def __init__(self): + self.cell_kv = True + self.display_format = 'x' + self.gsmtapv3 = False + self.logger = logging.getLogger('scat.test') + self.lte_last_cell_id = [0, 0] + self.lte_last_earfcn_dl = [0, 0] + self.lte_last_earfcn_ul = [0, 0] + self.lte_last_bw_dl = [0, 0] + self.lte_last_bw_ul = [0, 0] + self.lte_serving_cell = [{}, {}] + self.nr_serving_cell = [{}, {}] + + +log_header = namedtuple('QcDiagLogHeader', 'cmd_code reserved length1 length2 log_id timestamp') + + +def lte_hdr(code): + return log_header(0x10, 0, 0, 0, diagcmd.diag_log_get_lte_item_id(code), 0) + + +class TestCellKvUtil(unittest.TestCase): + def test_dl_earfcn_to_frequency_hz(self): + self.assertEqual(util.dl_earfcn_to_frequency_hz(1850), 1870000000) # B3 + self.assertEqual(util.dl_earfcn_to_frequency_hz(6300), 806000000) # B20 + self.assertEqual(util.dl_earfcn_to_frequency_hz(2400), 869000000) # B5 + self.assertEqual(util.dl_earfcn_to_frequency_hz(999999), 0) # out of range + + def test_nrarfcn_to_frequency_hz(self): + self.assertEqual(util.nrarfcn_to_frequency_hz(519953), 2599765000) + self.assertEqual(util.nrarfcn_to_frequency_hz(632628), 3489420000) # n78 + self.assertEqual(util.nrarfcn_to_frequency_hz(9999999), 0) # out of range + + def test_format_plmn(self): + self.assertEqual(util.format_plmn(460, 11, 2), '46011') + self.assertEqual(util.format_plmn(302, 220, 3), '302220') + self.assertEqual(util.format_plmn(None, None), '0') + + def test_serving_identity_fields_match_guard(self): + cache = {'earfcn': 6300, 'pci': 214, 'plmn': '310260', 'mcc': 310, 'mnc': 260, + 'tac': 7, 'cid': 1, 'band': 20, 'bwmhzdl': 10, 'bwmhzul': 10} + self.assertEqual(util.serving_identity_fields(cache, 6300, 214)['band'], 20) + self.assertEqual(util.serving_identity_fields(cache, 6300, 999), {}) # pci mismatch + self.assertEqual(util.serving_identity_fields(cache, 100, 214), {}) # earfcn mismatch + self.assertEqual(util.serving_identity_fields({}, 6300, 214), {}) # empty cache + + +class TestLteCellKv(unittest.TestCase): + def setUp(self): + self.parent = FakeParent() + self.parser = DiagLteLogParser(parent=self.parent) + + def test_scell_meas_kv(self): + payload = binascii.unhexlify('040100009C18D60AECC44E00E2244E00FFFCE30FFED80A0047AD56021D310100A2624100') + r = self.parser.parse_lte_ml1_scell_meas(lte_hdr(diagcmd.diag_log_code_lte.LOG_LTE_ML1_SERVING_CELL_MEAS_AND_EVAL), payload, dict()) + self.assertEqual(r['stdout'], 'pci=214,earfcn=6300,earfcn_ul=24300,frequency=806000000,protocol=lte,cell=scell,plmn=0,mcc=0,mnc=0,tac=0,cid=0,band=0,bwmhzdl=0,bwmhzul=0 rssi=-66.625,rsrp=-101.25,rsrq=-14.0625') + + def test_ncell_meas_kv(self): + payload = binascii.unhexlify('040100009C1847008348E44DDEA44C00CAB4CC32B6D8420300000000FF773301FF77330122020100') + r = self.parser.parse_lte_ml1_ncell_meas(lte_hdr(diagcmd.diag_log_code_lte.LOG_LTE_ML1_NEIGHBOR_MEASUREMENTS), payload, dict()) + self.assertEqual(r['stdout'], 'pci=131,earfcn=6300,earfcn_ul=24300,frequency=806000000,protocol=lte,cell=ncell,plmn=0,mcc=0,mnc=0,tac=0,cid=0,band=0,bwmhzdl=0,bwmhzul=0 rssi=-75.75,rsrp=-102.125,rsrq=-17.3125') + + def test_rrc_cell_info_populates_cache(self): + payload = binascii.unhexlify('034D0021070000714D00004B4B33C8B009159B03000000CC01020B0000') + self.parser.parse_lte_rrc_cell_info(lte_hdr(diagcmd.diag_log_code_lte.LOG_LTE_RRC_SERVING_CELL_INFO), payload, dict()) + self.assertEqual(self.parent.lte_serving_cell[0], { + 'pci': 77, 'earfcn': 1825, 'plmn': '46011', 'mcc': 460, 'mnc': 11, + 'tac': 39701, 'cid': 162580531, 'band': 3, 'bwmhzdl': 15, 'bwmhzul': 15}) + + def test_scell_meas_kv_join_hit(self): + # Serving cell identity from a prior RRC SCell Info is joined onto the ML1 measurement. + self.parent.lte_serving_cell[0] = {'pci': 214, 'earfcn': 6300, 'plmn': '310260', + 'mcc': 310, 'mnc': 260, 'tac': 7, 'cid': 123456, 'band': 20, 'bwmhzdl': 10, 'bwmhzul': 10} + payload = binascii.unhexlify('040100009C18D60AECC44E00E2244E00FFFCE30FFED80A0047AD56021D310100A2624100') + r = self.parser.parse_lte_ml1_scell_meas(lte_hdr(diagcmd.diag_log_code_lte.LOG_LTE_ML1_SERVING_CELL_MEAS_AND_EVAL), payload, dict()) + self.assertEqual(r['stdout'], 'pci=214,earfcn=6300,earfcn_ul=24300,frequency=806000000,protocol=lte,cell=scell,plmn=310260,mcc=310,mnc=260,tac=7,cid=123456,band=20,bwmhzdl=10,bwmhzul=10 rssi=-66.625,rsrp=-101.25,rsrq=-14.0625') + + def test_scell_meas_kv_join_mismatch_guard(self): + # Cached serving cell is a *different* cell -> identity must NOT be joined. + self.parent.lte_serving_cell[0] = {'pci': 77, 'earfcn': 1825, 'plmn': '46011', + 'mcc': 460, 'mnc': 11, 'tac': 39701, 'cid': 1, 'band': 3, 'bwmhzdl': 15, 'bwmhzul': 15} + payload = binascii.unhexlify('040100009C18D60AECC44E00E2244E00FFFCE30FFED80A0047AD56021D310100A2624100') + r = self.parser.parse_lte_ml1_scell_meas(lte_hdr(diagcmd.diag_log_code_lte.LOG_LTE_ML1_SERVING_CELL_MEAS_AND_EVAL), payload, dict()) + self.assertIn('plmn=0,mcc=0,mnc=0,tac=0,cid=0,band=0,bwmhzdl=0,bwmhzul=0', r['stdout']) + + def test_default_output_unchanged(self): + # parent=None => cell_kv defaults off => original human-readable output. + parser = DiagLteLogParser(parent=None) + payload = binascii.unhexlify('040100009C18D60AECC44E00E2244E00FFFCE30FFED80A0047AD56021D310100A2624100') + r = parser.parse_lte_ml1_scell_meas(lte_hdr(diagcmd.diag_log_code_lte.LOG_LTE_ML1_SERVING_CELL_MEAS_AND_EVAL), payload, dict()) + self.assertTrue(r['stdout'].startswith('LTE SCell: EARFCN: 6300')) + + +class TestNrCellKv(unittest.TestCase): + ML1_MEAS = '070002000114000026ffffff44000000991006000100c602000000000000000000000000ffffffffffff0000ffffffffc6027e000100000017caffff0afaffff000000000000000000000000a5a1dbbd4199a005a3bcffff17caffff17caffff0afaffff0000000000000000' + + def setUp(self): + self.parent = FakeParent() + self.parser = DiagNrLogParser(parent=self.parent) + + def test_meas_db_update_kv(self): + r = self.parser.parse_nr_ml1_meas_db_update(log_header(0x10, 0, 0, 0, 0, 0), binascii.unhexlify(self.ML1_MEAS), dict()) + self.assertEqual(r['stdout'], + 'pci=710,earfcn=397465,earfcn_ul=397465,frequency=1987325000,protocol=nr,cell=scell,plmn=0,mcc=0,mnc=0,tac=0,cid=0,band=0,bwmhzdl=0,bwmhzul=0 rssi=0,rsrp=0,rsrq=0\n' + 'pci=710,earfcn=397465,earfcn_ul=397465,frequency=1987325000,protocol=nr,cell=ncell,plmn=0,mcc=0,mnc=0,tac=0,cid=0,band=0,bwmhzdl=0,bwmhzul=0 rssi=0,rsrp=-107.8203125,rsrq=-11.921875') + + def test_rrc_scell_info_populates_cache(self): + payload = binascii.unhexlify('040000009d02e0ca0900d6c609005a005a0000127df204000000060102010001297900004e00') + pkt_header = log_header(0x10, 0, 0, 0, diagcmd.diag_log_get_lte_item_id(diagcmd.diag_log_code_5gnr.LOG_5GNR_RRC_SERVING_CELL_INFO), 0) + self.parser.parse_nr_rrc_scell_info(pkt_header, payload, dict()) + self.assertEqual(self.parent.nr_serving_cell[0], { + 'pci': 669, 'earfcn': 641760, 'earfcn_ul': 640726, 'plmn': '26201', + 'mcc': 262, 'mnc': 1, 'tac': 31017, 'cid': 21248152064, 'band': 78, + 'bwmhzdl': 90, 'bwmhzul': 90}) + + def test_meas_db_update_kv_join_hit(self): + # Serving cell enriched from cache (incl. real UL NR-ARFCN); neighbor cell left untouched + # even though it shares the same PCI/NR-ARFCN in this capture. + self.parent.nr_serving_cell[0] = {'pci': 710, 'earfcn': 397465, 'earfcn_ul': 390000, + 'plmn': '26201', 'mcc': 262, 'mnc': 1, 'tac': 31017, 'cid': 999, 'band': 41, + 'bwmhzdl': 100, 'bwmhzul': 100} + r = self.parser.parse_nr_ml1_meas_db_update(log_header(0x10, 0, 0, 0, 0, 0), binascii.unhexlify(self.ML1_MEAS), dict()) + lines = r['stdout'].split('\n') + self.assertEqual(lines[0], 'pci=710,earfcn=397465,earfcn_ul=390000,frequency=1987325000,protocol=nr,cell=scell,plmn=26201,mcc=262,mnc=1,tac=31017,cid=999,band=41,bwmhzdl=100,bwmhzul=100 rssi=0,rsrp=0,rsrq=0') + self.assertIn('cell=ncell,plmn=0,mcc=0,mnc=0,tac=0,cid=0,band=0,bwmhzdl=0,bwmhzul=0', lines[1]) + + +if __name__ == '__main__': + unittest.main() From 62e8b55cc562ef7d28d7beeadf778e02004b1178 Mon Sep 17 00:00:00 2001 From: Mark Houtz Date: Fri, 4 Sep 2026 15:00:39 -0600 Subject: [PATCH 3/5] fix(nr): guard NR RRC SCell Info version dispatch; log unknown versions parse_nr_rrc_scell_info handled versions 0.04 / 3.00 / 3.02 / 3.03, but a rel_maj==0x03 packet with any other rel_min fell through the inner if/elif with `item` never assigned and then crashed with NameError at the nr_serving_cell assignment. Flatten the dispatch and add a single `item is None` guard that logs the version + body and returns cleanly. This unblocks NR serving-cell identity (TAC/CID/PLMN) for modems whose SCell Info version isn't listed yet (e.g. the Quectel RG650 in the GL-E5800): the warning now prints the exact version + raw bytes needed to add its struct, instead of silently yielding empty NR identity in --cell-kv output. Co-Authored-By: Claude Opus 4.8 --- src/scat/parsers/qualcomm/diagnrlogparser.py | 21 ++++++++++++-------- 1 file changed, 13 insertions(+), 8 deletions(-) diff --git a/src/scat/parsers/qualcomm/diagnrlogparser.py b/src/scat/parsers/qualcomm/diagnrlogparser.py index 8ba498b..7a41812 100644 --- a/src/scat/parsers/qualcomm/diagnrlogparser.py +++ b/src/scat/parsers/qualcomm/diagnrlogparser.py @@ -258,17 +258,22 @@ def parse_nr_rrc_scell_info(self, pkt_header, pkt_body: bytes, args: dict): item_struct = namedtuple('QcDiagNrScellInfo', 'pci dl_nrarfcn ul_nrarfcn dl_bandwidth ul_bandwidth cell_id mcc mnc_digit mnc allowed_access tac band') item_struct_v30000 = namedtuple('QcDiagNrScellInfoV30000', 'pci nr_cgi dl_nrarfcn ul_nrarfcn dl_bandwidth ul_bandwidth cell_id mcc mnc_digit mnc allowed_access tac band') + item = None if pkt_ver.rel_maj == 0x00 and pkt_ver.rel_min == 0x04: # PCI 2b, DL NR-ARFCN 4b, UL NR-ARFCN 4b, DLBW 2b, ULBW 2b, Cell ID 8b, MCC 2b, MCC digit 1b, MNC 2b, MNC digit 1b, TAC 4b, ? item = item_struct._make(struct.unpack(' Date: Fri, 4 Sep 2026 16:41:57 -0600 Subject: [PATCH 4/5] fix(nr): don't emit an scell kv-row when there's no serving cell (pci 0xffff) On SA 5G the NR ML1 measurement reports one entry per measured carrier; a carrier the UE isn't camped on reports serv_cell_pci == 0xffff. SCAT was emitting an all-zero `protocol=nr,cell=scell` row for each of those (pci=65535, tac=0, cid=0, rsrp=0), which drowned the real serving cell and made NR captures look like they had no TAC/CID. The real serving cell's identity join already works (verified live: n25 PCI 796 -> tac/cid populated from RRC SCell Info). Guard the scell kv-row on serv_cell_pci != 0xffff so only actual serving cells are emitted; neighbors are unaffected. Adds a regression test. Co-Authored-By: Claude Opus 4.8 --- src/scat/parsers/qualcomm/diagnrlogparser.py | 5 ++++- tests/test_cell_kv.py | 13 +++++++++++++ 2 files changed, 17 insertions(+), 1 deletion(-) diff --git a/src/scat/parsers/qualcomm/diagnrlogparser.py b/src/scat/parsers/qualcomm/diagnrlogparser.py index 7a41812..a1a21bd 100644 --- a/src/scat/parsers/qualcomm/diagnrlogparser.py +++ b/src/scat/parsers/qualcomm/diagnrlogparser.py @@ -145,7 +145,10 @@ def parse_nr_ml1_meas_db_update(self, pkt_header, pkt_body: bytes, args: dict): else: rsrp_str = '' nr_frequency = util.nrarfcn_to_frequency_hz(meas_carrier_list.raster_arfcn) - if cell_kv: + # serv_cell_pci == 0xffff means the UE has no serving cell on this + # measured carrier (neighbor-only) — don't emit an all-zero scell row + # for it. Its neighbor cells are still reported below. + if cell_kv and meas_carrier_list.serv_cell_pci != 0xffff: nr_ident = util.serving_identity_fields(nr_cache, meas_carrier_list.raster_arfcn, meas_carrier_list.serv_cell_pci) nr_earfcn_ul = nr_cache.get('earfcn_ul', meas_carrier_list.raster_arfcn) if nr_ident else meas_carrier_list.raster_arfcn kv_lines.append(util.format_cell_kv('nr', 'scell', meas_carrier_list.serv_cell_pci, diff --git a/tests/test_cell_kv.py b/tests/test_cell_kv.py index a3b225a..322cc85 100644 --- a/tests/test_cell_kv.py +++ b/tests/test_cell_kv.py @@ -120,6 +120,19 @@ def test_meas_db_update_kv(self): 'pci=710,earfcn=397465,earfcn_ul=397465,frequency=1987325000,protocol=nr,cell=scell,plmn=0,mcc=0,mnc=0,tac=0,cid=0,band=0,bwmhzdl=0,bwmhzul=0 rssi=0,rsrp=0,rsrq=0\n' 'pci=710,earfcn=397465,earfcn_ul=397465,frequency=1987325000,protocol=nr,cell=ncell,plmn=0,mcc=0,mnc=0,tac=0,cid=0,band=0,bwmhzdl=0,bwmhzul=0 rssi=0,rsrp=-107.8203125,rsrq=-11.921875') + def test_meas_db_update_kv_skips_absent_serving_cell(self): + # serv_cell_pci == 0xffff means the UE has no serving cell on this measured + # carrier — SCAT must not emit an all-zero scell row for it. (These bogus + # rows are what made SA-NR captures look like "no TAC/CID".) + baseline = self.parser.parse_nr_ml1_meas_db_update( + log_header(0x10, 0, 0, 0, 0, 0), binascii.unhexlify(self.ML1_MEAS), dict()) + self.assertIn('cell=scell', baseline['stdout']) # normally emits one + b = bytearray(binascii.unhexlify(self.ML1_MEAS)) + b[22:24] = b'\xff\xff' # serv_cell_pci -> 0xffff + r = self.parser.parse_nr_ml1_meas_db_update( + log_header(0x10, 0, 0, 0, 0, 0), bytes(b), dict()) + self.assertNotIn('cell=scell', r['stdout']) + def test_rrc_scell_info_populates_cache(self): payload = binascii.unhexlify('040000009d02e0ca0900d6c609005a005a0000127df204000000060102010001297900004e00') pkt_header = log_header(0x10, 0, 0, 0, diagcmd.diag_log_get_lte_item_id(diagcmd.diag_log_code_5gnr.LOG_5GNR_RRC_SERVING_CELL_INFO), 0) From f99216dacb6df0b6c5eff7db0d379e2f73a0aca8 Mon Sep 17 00:00:00 2001 From: Mark Houtz Date: Fri, 4 Sep 2026 17:00:58 -0600 Subject: [PATCH 5/5] feat: machine-readable LTE/5G-NR cell measurements (--cell-kv, --meas-gsmtap) Consolidates the cell-measurement work (supersedes the stacked PRs #150, #1, #2). --cell-kv: emit LTE and 5G-NR serving and neighbor cell measurements as key=value lines to stdout for machine parsing. Frequency in Hz (3GPP TS 36.101 / 38.104 tables), real LTE UL-EARFCN, and a stateful RRC<->ML1 identity join (per-radio serving-cell cache from RRC SCell Info merged onto ML1 measurements, guarded on (earfcn, pci) so no stale identity; neighbors stay 0). --meas-gsmtap: also write the measurements into the capture as GSMTAPv3 SIGNAL_STATUS_REPORT packets, so RSRP/RSRQ/RSSI/SINR are visible in Wireshark (NR uses the SS_* tags). Fixes a latent float-packing bug in create_gsmtap_header's v3 metadata (float tags only worked for int values). CellID capture is order-independent: cache the latest ML1 serving signal so the RRC SCell Info parser emits its own identity-bearing serving-cell line (CellID/TAC/PLMN) joined to that signal, even when the RRC packet arrives before the next measurement. Default (no flags) output is unchanged. util helpers: dl_earfcn_to_frequency_hz, nrarfcn_to_frequency_hz, format_cell_kv, format_plmn, serving_identity_fields, build_signal_status_report. Full suite: 107 pass. Co-Authored-By: Claude Opus 4.8 --- src/scat/main.py | 2 + src/scat/parsers/qualcomm/diagltelogparser.py | 51 +++++- src/scat/parsers/qualcomm/diagnrlogparser.py | 61 +++++-- src/scat/parsers/qualcomm/qualcommparser.py | 8 + src/scat/util.py | 45 +++-- tests/test_cell_kv.py | 42 +++++ tests/test_meas_gsmtap.py | 160 ++++++++++++++++++ 7 files changed, 338 insertions(+), 31 deletions(-) create mode 100644 tests/test_meas_gsmtap.py diff --git a/src/scat/main.py b/src/scat/main.py index 4907583..23ea4cc 100755 --- a/src/scat/main.py +++ b/src/scat/main.py @@ -65,6 +65,7 @@ def scat_main(): parser.add_argument('-L', '--layer', help='Specify the layers to see as GSMTAP packets (comma separated).\nAvailable layers: {}, Default: "ip,nas,rrc"'.format(', '.join(valid_layers)), type=str, default='ip,nas,rrc') parser.add_argument('-f', '--format', help='Select display format for LAC/RAC/TAC/CID: [d]ecimal, he[x]adecimal (default), [b]oth.', type=str, default='x', choices=['d', 'x', 'b']) parser.add_argument('--cell-kv', help='Emit LTE/5G-NR serving and neighbor cell measurements as key=value lines to stdout for machine parsing.', action='store_true') + parser.add_argument('--meas-gsmtap', help='Emit LTE/5G-NR serving and neighbor cell measurements (RSRP/RSRQ/RSSI/SINR) into the capture as GSMTAPv3 signal status reports.', action='store_true') parser.add_argument('-3', '--gsmtapv3', help='Enable GSMTAPv3 for 2G/3G/4G. Default: enabled only for 5G NR', action='store_true') input_group = parser.add_mutually_exclusive_group(required=True) @@ -197,6 +198,7 @@ def scat_main(): 'layer': layers, 'format': args.format, 'cell-kv': args.cell_kv, + 'meas-gsmtap': args.meas_gsmtap, 'gsmtapv3': args.gsmtapv3}) elif args.type == 'sec': if args.trace: diff --git a/src/scat/parsers/qualcomm/diagltelogparser.py b/src/scat/parsers/qualcomm/diagltelogparser.py index be9740a..c6d1671 100644 --- a/src/scat/parsers/qualcomm/diagltelogparser.py +++ b/src/scat/parsers/qualcomm/diagltelogparser.py @@ -186,16 +186,27 @@ def parse_lte_ml1_scell_meas(self, pkt_header, pkt_body: bytes, args: dict): real_rssi = self.parse_rssi(meas_rssi) real_rsrq = self.parse_rsrq(meas_rsrq) + radio_id = args['radio_id'] if (args and 'radio_id' in args) else 0 + cache = self.parent.lte_serving_cell[radio_id] if (self.parent and radio_id in (0, 1)) else {} + ident = util.serving_identity_fields(cache, item.earfcn, pci) + if getattr(self.parent, 'cell_kv', False): - radio_id = args['radio_id'] if (args and 'radio_id' in args) else 0 - cache = self.parent.lte_serving_cell[radio_id] if (self.parent and radio_id in (0, 1)) else {} - ident = util.serving_identity_fields(cache, item.earfcn, pci) + if self.parent and radio_id in (0, 1): + self.parent.lte_serving_signal[radio_id] = {'earfcn': item.earfcn, 'pci': pci, + 'rssi': real_rssi, 'rsrp': real_rsrp, 'rsrq': real_rsrq} stdout = util.format_cell_kv('lte', 'scell', pci, item.earfcn, util.calculate_ul_earfcn(item.earfcn), util.dl_earfcn_to_frequency_hz(item.earfcn), rssi=real_rssi, rsrp=real_rsrp, rsrq=real_rsrq, **ident) else: stdout = 'LTE SCell: EARFCN: {}, PCI: {:3d}, Measured RSRP: {:.2f}, Measured RSSI: {:.2f}, Measured RSRQ: {:.2f}'.format(item.earfcn, pci, real_rsrp, real_rssi, real_rsrq) - return {'stdout': stdout, 'ts': pkt_ts} + + result = {'stdout': stdout, 'ts': pkt_ts} + if getattr(self.parent, 'meas_gsmtap', False): + ts_sec = calendar.timegm(pkt_ts.timetuple()) + result['cp'] = [util.build_signal_status_report(item.earfcn, pci=pci, + band=ident.get('band'), ts_sec=ts_sec, ts_usec=pkt_ts.microsecond, + rsrp=real_rsrp, rsrq=real_rsrq, rssi=real_rssi)] + return result def parse_lte_ml1_ncell_meas(self, pkt_header, pkt_body: bytes, args: dict): pkt_ts = util.parse_qxdm_ts(pkt_header.timestamp) @@ -227,9 +238,11 @@ def parse_lte_ml1_ncell_meas(self, pkt_header, pkt_body: bytes, args: dict): q_rxlevmin = item.q_rxlevmin_n_cells & 0x3f n_cells = item.q_rxlevmin_n_cells >> 6 cell_kv = getattr(self.parent, 'cell_kv', False) + meas_gsmtap = getattr(self.parent, 'meas_gsmtap', False) n_earfcn_ul = util.calculate_ul_earfcn(item.earfcn) n_frequency = util.dl_earfcn_to_frequency_hz(item.earfcn) kv_lines = [] + cp_pkts = [] if not cell_kv: stdout += 'LTE NCell: EARFCN: {}, number of cells: {}\n'.format(item.earfcn, n_cells) @@ -269,9 +282,14 @@ def parse_lte_ml1_ncell_meas(self, pkt_header, pkt_body: bytes, args: dict): n_earfcn_ul, n_frequency, rssi=n_real_rssi, rsrp=n_real_rsrp, rsrq=n_real_rsrq)) else: stdout += '└── Neighbor cell {}: PCI: {:3d}, RSRP: {:.2f}, RSSI: {:.2f}, RSRQ: {:.2f}\n'.format(i, n_pci, n_real_rsrp, n_real_rssi, n_real_rsrq) - if cell_kv: - return {'stdout': '\n'.join(kv_lines), 'ts': pkt_ts} - return {'stdout': stdout.rstrip(), 'ts': pkt_ts} + if meas_gsmtap: + cp_pkts.append(util.build_signal_status_report(item.earfcn, pci=n_pci, + ts_sec=calendar.timegm(pkt_ts.timetuple()), ts_usec=pkt_ts.microsecond, + rsrp=n_real_rsrp, rsrq=n_real_rsrq, rssi=n_real_rssi)) + result = {'stdout': '\n'.join(kv_lines) if cell_kv else stdout.rstrip(), 'ts': pkt_ts} + if cp_pkts: + result['cp'] = cp_pkts + return result def parse_lte_ml1_scell_meas_response_cell_v36(self, cell_id: int, cell_bytes: bytes, rsrp_offset: int=16, snr_offset: int=80, sir_cinr_offset: int=104): interim = struct.unpack(' float: def parse_nr_ml1_meas_db_update(self, pkt_header, pkt_body: bytes, args: dict): stdout = '' cell_kv = getattr(self.parent, 'cell_kv', False) + meas_gsmtap = getattr(self.parent, 'meas_gsmtap', False) kv_lines = [] + cp_pkts = [] radio_id = args['radio_id'] if (args and 'radio_id' in args) else 0 - nr_cache = self.parent.nr_serving_cell[radio_id] if (self.parent and cell_kv and radio_id in (0, 1)) else {} + nr_cache = self.parent.nr_serving_cell[radio_id] if (self.parent and (cell_kv or meas_gsmtap) and radio_id in (0, 1)) else {} + pkt_ts = util.parse_qxdm_ts(pkt_header.timestamp) + ts_sec = calendar.timegm(pkt_ts.timetuple()) pkt_ver = self.nr_pkt_ver._make(struct.unpack(' 0x00 and meas_carrier_list.serv_cell_index < 0xff: @@ -184,6 +199,11 @@ def parse_nr_ml1_meas_db_update(self, pkt_header, pkt_body: bytes, args: dict): cell, cell_list.pci, cell_list.pbch_sfn, self.parse_float_q7(cell_list.cell_quality_rsrp), self.parse_float_q7(cell_list.cell_quality_rsrq), cell_list.num_beams) + if meas_gsmtap: + cp_pkts.append(util.build_signal_status_report(meas_carrier_list.raster_arfcn, + pci=cell_list.pci, ts_sec=ts_sec, ts_usec=pkt_ts.microsecond, + rsrp=self.parse_float_q7(cell_list.cell_quality_rsrp), + rsrq=self.parse_float_q7(cell_list.cell_quality_rsrq), is_nr=True)) for beam in range(cell_list.num_beams): beam_meas_struct = namedtuple('QcDiagNrMl1Packet', 'ssb_index null_0 rx_beam_0 rx_beam_1 null_1 ssb_ref_timing rx_beam_info_rsrp_0 rx_beam_info_rsrp_1 nr2nr_filtered_beam_rsrp_l3 nr2nr_filtered_beam_rsrq_l3 l_2_nr_filtered_tx_beam_rsrp_l3 l_2_nr_filtered_tx_beam_rsrq_l3') beam_meas_struct_v3 = namedtuple('QcDiagNrMl1PacketV3', 'ssb_index null_0 rx_beam_0 rx_beam_1 null_1 ssb_ref_timing rx_beam_info_rsrp_0 rx_beam_info_rsrp_1 unk_0 unk_1 unk_2 unk_3 unk_4 unk_5 unk_6 unk_7 unk_8 unk_9 nr2nr_filtered_beam_rsrp_l3 nr2nr_filtered_beam_rsrq_l3 l_2_nr_filtered_tx_beam_rsrp_l3 l_2_nr_filtered_tx_beam_rsrq_l3') @@ -211,10 +231,10 @@ def parse_nr_ml1_meas_db_update(self, pkt_header, pkt_body: bytes, args: dict): self.parse_float_q7(beam_meas.l_2_nr_filtered_tx_beam_rsrp_l3), self.parse_float_q7(beam_meas.l_2_nr_filtered_tx_beam_rsrq_l3), ) - pkt_ts = util.parse_qxdm_ts(pkt_header.timestamp) - if cell_kv: - return {'stdout': '\n'.join(kv_lines), 'ts': pkt_ts} - return {'stdout': stdout.rstrip(), 'ts': pkt_ts} + result = {'stdout': '\n'.join(kv_lines) if cell_kv else stdout.rstrip(), 'ts': pkt_ts} + if cp_pkts: + result['cp'] = cp_pkts + return result # RRC def parse_nr_mib_info(self, pkt_header, pkt_body: bytes, args: dict): @@ -292,6 +312,21 @@ def parse_nr_rrc_scell_info(self, pkt_header, pkt_body: bytes, args: dict): 'bwmhzdl': item.dl_bandwidth, 'bwmhzul': item.ul_bandwidth, } + if getattr(self.parent, 'cell_kv', False): + # Emit an identity-bearing serving-cell line (CellID/TAC/PLMN) joined to + # the last ML1 signal, so CellID is captured order-independently. Skip + # when no signal is cached yet (the ML1 join fills it onto the next meas). + sig = self.parent.nr_serving_signal[radio_id] if radio_id in (0, 1) else {} + if sig.get('earfcn') == item.dl_nrarfcn and sig.get('pci') == item.pci: + stdout = util.format_cell_kv('nr', 'scell', item.pci, item.dl_nrarfcn, + item.ul_nrarfcn, util.nrarfcn_to_frequency_hz(item.dl_nrarfcn), + rsrp=sig['rsrp'], plmn=util.format_plmn(item.mcc, item.mnc, item.mnc_digit), + mcc=item.mcc, mnc=item.mnc, tac=item.tac, cid=item.cell_id, + band=item.band, bwmhzdl=item.dl_bandwidth, bwmhzul=item.ul_bandwidth) + else: + stdout = '' + return {'stdout': stdout, 'ts': pkt_ts} + if self.display_format == 'd': tac_cid_fmt = 'TAC/CID: {}/{}'.format(item.tac, item.cell_id) elif self.display_format == 'x': diff --git a/src/scat/parsers/qualcomm/qualcommparser.py b/src/scat/parsers/qualcomm/qualcommparser.py index 53283e0..7fe7af3 100644 --- a/src/scat/parsers/qualcomm/qualcommparser.py +++ b/src/scat/parsers/qualcomm/qualcommparser.py @@ -71,6 +71,11 @@ def __init__(self): # LTE/NR RRC SCell Info packets and joined onto ML1 measurement reports. self.lte_serving_cell = [{}, {}] self.nr_serving_cell = [{}, {}] + # Latest ML1 serving-cell signal (per radio) so an RRC SCell Info packet + # can emit an identity-bearing --cell-kv line even when it arrives before + # the next ML1 measurement (makes CellID capture order-independent). + self.lte_serving_signal = [{}, {}] + self.nr_serving_signal = [{}, {}] self.io_device: AbstractIO self.writer: AbstractWriter @@ -82,6 +87,7 @@ def __init__(self): self.log_id_range = {} self.cacombos = False self.cell_kv = False + self.meas_gsmtap = False self.combine_stdout = False self.check_crc = True self.layers = [] @@ -296,6 +302,8 @@ def set_parameter(self, params: dict[str, Any]) -> None: self.cacombos = params[p] elif p == 'cell-kv': self.cell_kv = params[p] + elif p == 'meas-gsmtap': + self.meas_gsmtap = params[p] elif p == 'combine-stdout': self.combine_stdout = params[p] elif p == 'disable-crc-check': diff --git a/src/scat/util.py b/src/scat/util.py index 1031825..418473d 100644 --- a/src/scat/util.py +++ b/src/scat/util.py @@ -475,18 +475,17 @@ def create_gsmtap_header(version: int = 2, payload_type: int = 0, timeslot: int gsmtap_v3_metadata += struct.pack('!HHL', t.CHANNEL_NUMBER, 4, arfcn) header_len += 8 for k, v in metadata.items(): - if type(v) == int: - if k in (t.BAND_INDICATOR, t.BSIC_PSC_PCI, t.SUBFN, t.HFN): - buf = struct.pack('!H', v) - elif k in (t.GSM_TIMESLOT, t.GSM_SUBSLOT, t.ANT_NUM): - buf = struct.pack('!B', v) - elif k in (t.SIGNAL_LEVEL, t.RSSI, t.SNR, t.SINR, t.RSCP, t.ECIO, t.RSRP, t.RSRQ, - t.SS_RSRP, t.CSI_RSRP, t.SRS_RSRP, t.SS_RSRQ, t.CSI_RSRQ, t.SS_SINR, t.CSI_SINR): - buf = struct.pack('!f', v) - else: - buf = struct.pack('!L', v) - else: + if isinstance(v, (bytes, bytearray)): buf = v + elif k in (t.BAND_INDICATOR, t.BSIC_PSC_PCI, t.SUBFN, t.HFN): + buf = struct.pack('!H', v) + elif k in (t.GSM_TIMESLOT, t.GSM_SUBSLOT, t.ANT_NUM): + buf = struct.pack('!B', v) + elif k in (t.SIGNAL_LEVEL, t.RSSI, t.SNR, t.SINR, t.RSCP, t.ECIO, t.RSRP, t.RSRQ, + t.SS_RSRP, t.CSI_RSRP, t.SRS_RSRP, t.SS_RSRQ, t.CSI_RSRQ, t.SS_SINR, t.CSI_SINR): + buf = struct.pack('!f', v) + else: + buf = struct.pack('!L', v) gsmtap_v3_metadata += struct.pack('!HH', k, len(buf)) gsmtap_v3_metadata += buf header_len += (4 + len(buf)) @@ -795,6 +794,30 @@ def serving_identity_fields(cache_entry: dict, earfcn: int, pci: int) -> dict: return {k: cache_entry[k] for k in ('plmn', 'mcc', 'mnc', 'tac', 'cid', 'band', 'bwmhzdl', 'bwmhzul') if k in cache_entry} +# Builds a GSMTAPv3 SIGNAL_STATUS_REPORT packet (header + metadata, no payload) +# carrying a single cell's measurement. arfcn goes in CHANNEL_NUMBER; signal +# metrics are emitted as float metadata (SS_* tags for NR, plain tags for LTE). +# Any metric passed as None is omitted. Returns the GSMTAPv3 bytes. +def build_signal_status_report(arfcn: int, pci=None, band=None, ts_sec: int = 0, + ts_usec: int = 0, rsrp=None, rsrq=None, rssi=None, sinr=None, is_nr: bool = False) -> bytes: + t = gsmtapv3_metadata_tags + md = {} + if pci is not None: + md[t.BSIC_PSC_PCI] = int(pci) + if band: + md[t.BAND_INDICATOR] = int(band) + if is_nr: + if rsrp is not None: md[t.SS_RSRP] = float(rsrp) + if rsrq is not None: md[t.SS_RSRQ] = float(rsrq) + if sinr is not None: md[t.SS_SINR] = float(sinr) + else: + if rsrp is not None: md[t.RSRP] = float(rsrp) + if rsrq is not None: md[t.RSRQ] = float(rsrq) + if rssi is not None: md[t.RSSI] = float(rssi) + if sinr is not None: md[t.SINR] = float(sinr) + return create_gsmtap_header(version=3, payload_type=gsmtapv3_types.SIGNAL_STATUS_REPORT, + arfcn=arfcn, device_sec=ts_sec, device_usec=ts_usec, metadata=md) + def convert_mcc(mcc_digit_2: int, mcc_digit_1: int, mcc_digit_0: int) -> str: if mcc_digit_2 > 0x09 or mcc_digit_1 > 0x09 or mcc_digit_0 > 0x09: raise ValueError('Invalid digit in MCC') diff --git a/tests/test_cell_kv.py b/tests/test_cell_kv.py index 322cc85..e3a32c6 100644 --- a/tests/test_cell_kv.py +++ b/tests/test_cell_kv.py @@ -26,6 +26,8 @@ def __init__(self): self.lte_last_bw_ul = [0, 0] self.lte_serving_cell = [{}, {}] self.nr_serving_cell = [{}, {}] + self.lte_serving_signal = [{}, {}] + self.nr_serving_signal = [{}, {}] log_header = namedtuple('QcDiagLogHeader', 'cmd_code reserved length1 length2 log_id timestamp') @@ -154,5 +156,45 @@ def test_meas_db_update_kv_join_hit(self): self.assertIn('cell=ncell,plmn=0,mcc=0,mnc=0,tac=0,cid=0,band=0,bwmhzdl=0,bwmhzul=0', lines[1]) +class TestCellIdFromRrc(unittest.TestCase): + """Order-independent CellID capture: an RRC SCell Info packet emits a full + identity-bearing kv line joined to the last ML1 serving signal, so CellID is + captured even when the RRC packet arrives before the next ML1 measurement.""" + + def setUp(self): + self.parent = FakeParent() + + def test_lte_rrc_emits_kv_with_cellid_when_signal_cached(self): + parser = DiagLteLogParser(parent=self.parent) + self.parent.lte_serving_signal[0] = {'earfcn': 1825, 'pci': 77, + 'rssi': -60.0, 'rsrp': -95.5, 'rsrq': -11.0} + payload = binascii.unhexlify('034D0021070000714D00004B4B33C8B009159B03000000CC01020B0000') + r = parser.parse_lte_rrc_cell_info(lte_hdr(diagcmd.diag_log_code_lte.LOG_LTE_RRC_SERVING_CELL_INFO), payload, dict()) + self.assertEqual(r['stdout'], 'pci=77,earfcn=1825,earfcn_ul=19825,frequency=1867500000,protocol=lte,cell=scell,plmn=46011,mcc=460,mnc=11,tac=39701,cid=162580531,band=3,bwmhzdl=15,bwmhzul=15 rssi=-60.0,rsrp=-95.5,rsrq=-11.0') + + def test_lte_rrc_emits_nothing_without_cached_signal(self): + parser = DiagLteLogParser(parent=self.parent) # no signal cached + payload = binascii.unhexlify('034D0021070000714D00004B4B33C8B009159B03000000CC01020B0000') + r = parser.parse_lte_rrc_cell_info(lte_hdr(diagcmd.diag_log_code_lte.LOG_LTE_RRC_SERVING_CELL_INFO), payload, dict()) + self.assertEqual(r['stdout'], '') # all-zero-metric line suppressed + + def test_nr_rrc_emits_kv_with_cellid_when_signal_cached(self): + parser = DiagNrLogParser(parent=self.parent) + self.parent.nr_serving_signal[0] = {'earfcn': 641760, 'pci': 669, 'rsrp': -88.0} + payload = binascii.unhexlify('040000009d02e0ca0900d6c609005a005a0000127df204000000060102010001297900004e00') + pkt_header = log_header(0x10, 0, 0, 0, diagcmd.diag_log_get_lte_item_id(diagcmd.diag_log_code_5gnr.LOG_5GNR_RRC_SERVING_CELL_INFO), 0) + r = parser.parse_nr_rrc_scell_info(pkt_header, payload, dict()) + self.assertEqual(r['stdout'], 'pci=669,earfcn=641760,earfcn_ul=640726,frequency=3626400000,protocol=nr,cell=scell,plmn=26201,mcc=262,mnc=1,tac=31017,cid=21248152064,band=78,bwmhzdl=90,bwmhzul=90 rssi=0,rsrp=-88.0,rsrq=0') + + def test_lte_ml1_scell_caches_signal(self): + parser = DiagLteLogParser(parent=self.parent) + payload = binascii.unhexlify('040100009C18D60AECC44E00E2244E00FFFCE30FFED80A0047AD56021D310100A2624100') + parser.parse_lte_ml1_scell_meas(lte_hdr(diagcmd.diag_log_code_lte.LOG_LTE_ML1_SERVING_CELL_MEAS_AND_EVAL), payload, dict()) + sig = self.parent.lte_serving_signal[0] + self.assertEqual(sig['earfcn'], 6300) + self.assertEqual(sig['pci'], 214) + self.assertAlmostEqual(sig['rsrp'], -101.25, places=4) + + if __name__ == '__main__': unittest.main() diff --git a/tests/test_meas_gsmtap.py b/tests/test_meas_gsmtap.py new file mode 100644 index 0000000..6bd7c98 --- /dev/null +++ b/tests/test_meas_gsmtap.py @@ -0,0 +1,160 @@ +#!/usr/bin/env python3 + +import unittest +import binascii +import logging +import struct +from collections import namedtuple + +import scat.util as util +import scat.parsers.qualcomm.diagcmd as diagcmd +from scat.parsers.qualcomm.diagltelogparser import DiagLteLogParser +from scat.parsers.qualcomm.diagnrlogparser import DiagNrLogParser + +_T = util.gsmtapv3_metadata_tags +_NAME = {int(t): t.name for t in _T} +_FLOAT_TAGS = {int(getattr(_T, x)) for x in ('SIGNAL_LEVEL', 'RSSI', 'SNR', 'SINR', 'RSCP', + 'ECIO', 'RSRP', 'RSRQ', 'SS_RSRP', 'CSI_RSRP', 'SRS_RSRP', 'SS_RSRQ', 'CSI_RSRQ', + 'SS_SINR', 'CSI_SINR')} +_H2_TAGS = {int(_T.BSIC_PSC_PCI), int(_T.BAND_INDICATOR), int(_T.SUBFN), int(_T.HFN)} + + +def decode_gsmtapv3(pkt): + """Decode a GSMTAPv3 packet into {'_type': name, tag_name: value, ...}.""" + version, _reserved, _hlen, ptype, _stype = struct.unpack('!BBHHH', pkt[0:8]) + assert version == 3, 'expected GSMTAPv3, got version %d' % version + out = {'_type': util.gsmtapv3_types(ptype).name} + o = 8 + while o < len(pkt): + tag = struct.unpack('!H', pkt[o:o+2])[0] + o += 2 + if tag == int(_T.END_OF_METADATA): + break + ln = struct.unpack('!H', pkt[o:o+2])[0] + o += 2 + val = pkt[o:o+ln] + o += ln + if tag in _FLOAT_TAGS: + v = struct.unpack('!f', val)[0] + elif tag in _H2_TAGS: + v = struct.unpack('!H', val)[0] + elif tag == int(_T.CHANNEL_NUMBER): + v = struct.unpack('!L', val)[0] + else: + v = val + out[_NAME.get(tag, hex(tag))] = v + return out + + +class FakeParent: + def __init__(self): + self.cell_kv = False + self.meas_gsmtap = True + self.display_format = 'x' + self.gsmtapv3 = False + self.logger = logging.getLogger('scat.test') + self.lte_serving_cell = [{}, {}] + self.nr_serving_cell = [{}, {}] + + +log_header = namedtuple('QcDiagLogHeader', 'cmd_code reserved length1 length2 log_id timestamp') + + +def lte_hdr(code): + return log_header(0x10, 0, 0, 0, diagcmd.diag_log_get_lte_item_id(code), 0) + + +class TestBuildSignalStatusReport(unittest.TestCase): + def test_lte_metrics_and_tags(self): + pkt = util.build_signal_status_report(6300, pci=214, band=20, rsrp=-101.25, rsrq=-14.0625, rssi=-66.625) + d = decode_gsmtapv3(pkt) + self.assertEqual(d['_type'], 'SIGNAL_STATUS_REPORT') + self.assertEqual(d['CHANNEL_NUMBER'], 6300) + self.assertEqual(d['BSIC_PSC_PCI'], 214) + self.assertEqual(d['BAND_INDICATOR'], 20) + self.assertAlmostEqual(d['RSRP'], -101.25, places=4) + self.assertAlmostEqual(d['RSRQ'], -14.0625, places=4) + self.assertAlmostEqual(d['RSSI'], -66.625, places=4) + + def test_nr_uses_ss_tags(self): + pkt = util.build_signal_status_report(632628, pci=500, rsrp=-88.5, rsrq=-11.5, is_nr=True) + d = decode_gsmtapv3(pkt) + self.assertIn('SS_RSRP', d) + self.assertIn('SS_RSRQ', d) + self.assertNotIn('RSRP', d) # NR must not use the LTE tag + + def test_none_metrics_omitted(self): + pkt = util.build_signal_status_report(6300, pci=1, rsrp=-90.0) + d = decode_gsmtapv3(pkt) + self.assertIn('RSRP', d) + self.assertNotIn('RSRQ', d) + self.assertNotIn('RSSI', d) + + +class TestLteMeasGsmtap(unittest.TestCase): + def setUp(self): + self.parent = FakeParent() + self.parser = DiagLteLogParser(parent=self.parent) + + def test_scell_emits_signal_report(self): + # band comes from the RRC serving-cell cache join + self.parent.lte_serving_cell[0] = {'pci': 214, 'earfcn': 6300, 'band': 20, + 'plmn': '0', 'mcc': 0, 'mnc': 0, 'tac': 0, 'cid': 0, 'bwmhzdl': 0, 'bwmhzul': 0} + payload = binascii.unhexlify('040100009C18D60AECC44E00E2244E00FFFCE30FFED80A0047AD56021D310100A2624100') + r = self.parser.parse_lte_ml1_scell_meas(lte_hdr(diagcmd.diag_log_code_lte.LOG_LTE_ML1_SERVING_CELL_MEAS_AND_EVAL), payload, dict()) + self.assertEqual(len(r['cp']), 1) + d = decode_gsmtapv3(r['cp'][0]) + self.assertEqual(d['_type'], 'SIGNAL_STATUS_REPORT') + self.assertEqual(d['CHANNEL_NUMBER'], 6300) + self.assertEqual(d['BSIC_PSC_PCI'], 214) + self.assertEqual(d['BAND_INDICATOR'], 20) + self.assertAlmostEqual(d['RSRP'], -101.25, places=4) + self.assertAlmostEqual(d['RSRQ'], -14.0625, places=4) + self.assertAlmostEqual(d['RSSI'], -66.625, places=4) + + def test_ncell_emits_one_report_per_cell(self): + payload = binascii.unhexlify('040100009C1847008348E44DDEA44C00CAB4CC32B6D8420300000000FF773301FF77330122020100') + r = self.parser.parse_lte_ml1_ncell_meas(lte_hdr(diagcmd.diag_log_code_lte.LOG_LTE_ML1_NEIGHBOR_MEASUREMENTS), payload, dict()) + self.assertEqual(len(r['cp']), 1) + d = decode_gsmtapv3(r['cp'][0]) + self.assertEqual(d['CHANNEL_NUMBER'], 6300) + self.assertEqual(d['BSIC_PSC_PCI'], 131) + self.assertAlmostEqual(d['RSRP'], -102.125, places=4) + self.assertAlmostEqual(d['RSRQ'], -17.3125, places=4) + + def test_disabled_by_default(self): + parser = DiagLteLogParser(parent=None) + payload = binascii.unhexlify('040100009C18D60AECC44E00E2244E00FFFCE30FFED80A0047AD56021D310100A2624100') + r = parser.parse_lte_ml1_scell_meas(lte_hdr(diagcmd.diag_log_code_lte.LOG_LTE_ML1_SERVING_CELL_MEAS_AND_EVAL), payload, dict()) + self.assertNotIn('cp', r) + + +class TestNrMeasGsmtap(unittest.TestCase): + ML1_MEAS = '070002000114000026ffffff44000000991006000100c602000000000000000000000000ffffffffffff0000ffffffffc6027e000100000017caffff0afaffff000000000000000000000000a5a1dbbd4199a005a3bcffff17caffff17caffff0afaffff0000000000000000' + + def setUp(self): + self.parent = FakeParent() + self.parser = DiagNrLogParser(parent=self.parent) + + def test_emits_ss_reports(self): + r = self.parser.parse_nr_ml1_meas_db_update(log_header(0x10, 0, 0, 0, 0, 0), binascii.unhexlify(self.ML1_MEAS), dict()) + # one serving-layer report + one detected-cell report + self.assertEqual(len(r['cp']), 2) + serv = decode_gsmtapv3(r['cp'][0]) + self.assertEqual(serv['_type'], 'SIGNAL_STATUS_REPORT') + self.assertEqual(serv['CHANNEL_NUMBER'], 397465) + self.assertEqual(serv['BSIC_PSC_PCI'], 710) + self.assertIn('SS_RSRP', serv) + ncell = decode_gsmtapv3(r['cp'][1]) + self.assertEqual(ncell['BSIC_PSC_PCI'], 710) + self.assertAlmostEqual(ncell['SS_RSRP'], -107.8203125, places=4) + self.assertAlmostEqual(ncell['SS_RSRQ'], -11.921875, places=4) + + def test_disabled_by_default(self): + parser = DiagNrLogParser(parent=None) + r = parser.parse_nr_ml1_meas_db_update(log_header(0x10, 0, 0, 0, 0, 0), binascii.unhexlify(self.ML1_MEAS), dict()) + self.assertNotIn('cp', r) + + +if __name__ == '__main__': + unittest.main()