From be53a25682cbe1c0868e19a0d0a866062367a65d Mon Sep 17 00:00:00 2001 From: Mark Houtz Date: Wed, 1 Jul 2026 15:08:11 -0600 Subject: [PATCH 1/2] 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/2] 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()