Skip to content

Add --cell-kv machine-readable LTE/5G-NR cell measurement output #150

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
shark-fi wants to merge 2 commits into
base: master
Choose a base branch
from
Open

Add --cell-kv machine-readable LTE/5G-NR cell measurement output #150

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
be53a25
feat: add --cell-kv machine-readable cell measurement output
shark-fi Jul 1, 2026
a3c91a7
test: add coverage for --cell-kv output and RRC/ML1 join
shark-fi Jul 1, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions src/scat/main.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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:
Expand Down
36 changes: 32 additions & 4 deletions src/scat/parsers/qualcomm/diagltelogparser.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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)]
Expand Down Expand Up @@ -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):
Expand Down Expand Up @@ -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:
Expand Down
65 changes: 50 additions & 15 deletions src/scat/parsers/qualcomm/diagnrlogparser.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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('<HH', pkt_body[0:4]))
num_layers = 0
current_offset = 0
Expand All @@ -83,12 +87,14 @@ def parse_nr_ml1_meas_db_update(self, pkt_header, pkt_body: bytes, args: dict):
if pkt_ver.rel_min == 0x07:
ml1_2_7 = ml1_shared_struct_v2_7._make(struct.unpack('<BB2sII', pkt_body[4:16]))
num_layers = ml1_2_7.num_layers
stdout += "NR ML1 Meas Packet: Layers: {}, ssb_periocity: {}\n".format(ml1_2_7.num_layers, ml1_2_7.ssb_periocity)
if not cell_kv:
stdout += "NR ML1 Meas Packet: Layers: {}, ssb_periocity: {}\n".format(ml1_2_7.num_layers, ml1_2_7.ssb_periocity)
current_offset = 16
elif pkt_ver.rel_min in (0x09, 0x0a):
ml1_2_9 = ml1_shared_struct_v2_9._make(struct.unpack('<IBBHII', pkt_body[4:20]))
num_layers = ml1_2_9.num_layers
stdout += "NR ML1 Meas Packet: Layers: {}, ssb_periocity: {}\n".format(ml1_2_9.num_layers, ml1_2_9.ssb_periocity)
if not cell_kv:
stdout += "NR ML1 Meas Packet: Layers: {}, ssb_periocity: {}\n".format(ml1_2_9.num_layers, ml1_2_9.ssb_periocity)
current_offset = 20
else:
if self.parent:
Expand All @@ -99,7 +105,8 @@ def parse_nr_ml1_meas_db_update(self, pkt_header, pkt_body: bytes, args: dict):
if pkt_ver.rel_min == 0x00:
ml1_3_0 = ml1_shared_struct_v2_9._make(struct.unpack('<IBBHII', pkt_body[4:20]))
num_layers = ml1_3_0.num_layers
stdout += "NR ML1 Meas Packet: Layers: {}, ssb_periocity: {}\n".format(ml1_3_0.num_layers, ml1_3_0.ssb_periocity)
if not cell_kv:
stdout += "NR ML1 Meas Packet: Layers: {}, ssb_periocity: {}\n".format(ml1_3_0.num_layers, ml1_3_0.ssb_periocity)
current_offset = 20
else:
if self.parent:
Expand Down Expand Up @@ -137,12 +144,20 @@ def parse_nr_ml1_meas_db_update(self, pkt_header, pkt_body: bytes, args: dict):
)
else:
rsrp_str = ''
stdout += "Layer {}: NR-ARFCN: {}, SCell PCI: {:4d}/SSB: {}, {}, RX beam: {}/{}, Num Cells: {} (S: {})\n".format(
layer, meas_carrier_list.raster_arfcn, meas_carrier_list.serv_cell_pci, meas_carrier_list.serv_ssb & 0xf,
rsrp_str,
meas_carrier_list.serv_rx_beam_0 if meas_carrier_list.serv_rx_beam_0 != 0xffff else 'NA',
meas_carrier_list.serv_rx_beam_1 if meas_carrier_list.serv_rx_beam_1 != 0xffff else 'NA',
meas_carrier_list.num_cells, meas_carrier_list.serv_cell_index)
nr_frequency = util.nrarfcn_to_frequency_hz(meas_carrier_list.raster_arfcn)
if cell_kv:
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,
meas_carrier_list.raster_arfcn, nr_earfcn_ul, nr_frequency,
rsrp=self.parse_float_q7(meas_carrier_list.serv_rsrp_rx_0), **nr_ident))
else:
stdout += "Layer {}: NR-ARFCN: {}, SCell PCI: {:4d}/SSB: {}, {}, RX beam: {}/{}, Num Cells: {} (S: {})\n".format(
layer, meas_carrier_list.raster_arfcn, meas_carrier_list.serv_cell_pci, meas_carrier_list.serv_ssb & 0xf,
rsrp_str,
meas_carrier_list.serv_rx_beam_0 if meas_carrier_list.serv_rx_beam_0 != 0xffff else 'NA',
meas_carrier_list.serv_rx_beam_1 if meas_carrier_list.serv_rx_beam_1 != 0xffff else 'NA',
meas_carrier_list.num_cells, meas_carrier_list.serv_cell_index)

if meas_carrier_list.num_cells == 0xff or meas_carrier_list.num_cells == 0x00:
if meas_carrier_list.serv_cell_index > 0x00 and meas_carrier_list.serv_cell_index < 0xff:
Expand All @@ -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('<HHB3sII', pkt_body[current_offset:current_offset+16]))
current_offset += 16
stdout += "└── Cell {}: PCI: {:4d}, PBCH SFN: {}, RSRP: {:.2f}, RSRQ: {:.2f}, Num Beams: {}\n".format(
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 cell_kv:
kv_lines.append(util.format_cell_kv('nr', 'ncell', cell_list.pci,
meas_carrier_list.raster_arfcn, meas_carrier_list.raster_arfcn, nr_frequency,
rsrp=self.parse_float_q7(cell_list.cell_quality_rsrp),
rsrq=self.parse_float_q7(cell_list.cell_quality_rsrq)))
else:
stdout += "└── Cell {}: PCI: {:4d}, PBCH SFN: {}, RSRP: {:.2f}, RSRQ: {:.2f}, Num Beams: {}\n".format(
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)
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')
if pkt_ver.rel_maj == 0x02 and pkt_ver.rel_min in (0x07, 0x09):
beam_meas = beam_meas_struct._make(struct.unpack('<HHHHIQIIIIII', pkt_body[current_offset: current_offset+44]))
current_offset += 44
stdout += " └── Beam {}: SSB[{}] Beam ID: {}/{}, RSRP: {:.2f}/{:.2f}, Filtered RSRP/RSRQ (Nr2Nr): {:.2f}/{:.2f}, Filtered RSRP/RSRQ (L2Nr): {:.2f}/{:.2f}\n".format(
if not cell_kv:
stdout += " └── Beam {}: SSB[{}] Beam ID: {}/{}, RSRP: {:.2f}/{:.2f}, Filtered RSRP/RSRQ (Nr2Nr): {:.2f}/{:.2f}, Filtered RSRP/RSRQ (L2Nr): {:.2f}/{:.2f}\n".format(
beam, beam_meas.ssb_index,
beam_meas.rx_beam_0, beam_meas.rx_beam_1,
self.parse_float_q7(beam_meas.rx_beam_info_rsrp_0), self.parse_float_q7(beam_meas.rx_beam_info_rsrp_1),
Expand All @@ -176,7 +198,8 @@ def parse_nr_ml1_meas_db_update(self, pkt_header, pkt_body: bytes, args: dict):
elif (pkt_ver.rel_maj == 0x02 and pkt_ver.rel_min in (0x0a, )) or pkt_ver.rel_maj == 0x03:
beam_meas = beam_meas_struct_v3._make(struct.unpack('<HHHHIQIIIIIIIIIIIIIIII', pkt_body[current_offset: current_offset+84]))
current_offset += 84
stdout += " └── Beam {}: SSB[{}] Beam ID: {}/{}, RSRP: {:.2f}/{:.2f}, RSRQ: {:.2f}/{:.2f}, Filtered RSRP/RSRQ (Nr2Nr): {:.2f}/{:.2f}, Filtered RSRP/RSRQ (L2Nr): {:.2f}/{:.2f}\n".format(
if not cell_kv:
stdout += " └── Beam {}: SSB[{}] Beam ID: {}/{}, RSRP: {:.2f}/{:.2f}, RSRQ: {:.2f}/{:.2f}, Filtered RSRP/RSRQ (Nr2Nr): {:.2f}/{:.2f}, Filtered RSRP/RSRQ (L2Nr): {:.2f}/{:.2f}\n".format(
beam, beam_meas.ssb_index,
beam_meas.rx_beam_0, beam_meas.rx_beam_1,
self.parse_float_q7(beam_meas.rx_beam_info_rsrp_0), self.parse_float_q7(beam_meas.rx_beam_info_rsrp_1),
Expand All @@ -186,6 +209,8 @@ def parse_nr_ml1_meas_db_update(self, pkt_header, pkt_body: bytes, args: dict):
)

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}

# RRC
Expand Down Expand Up @@ -249,6 +274,16 @@ def parse_nr_rrc_scell_info(self, pkt_header, pkt_body: bytes, args: dict):
self.parent.logger.log(logging.WARNING, "Body: {}".format(util.xxd_oneline(pkt_body)))
return None

radio_id = args['radio_id'] if (args and 'radio_id' in args) else 0
if self.parent and radio_id in (0, 1):
self.parent.nr_serving_cell[radio_id] = {
'pci': item.pci, 'earfcn': item.dl_nrarfcn, 'earfcn_ul': item.ul_nrarfcn,
'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,
}

if self.display_format == 'd':
tac_cid_fmt = 'TAC/CID: {}/{}'.format(item.tac, item.cell_id)
elif self.display_format == 'x':
Expand Down
8 changes: 8 additions & 0 deletions src/scat/parsers/qualcomm/qualcommparser.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -67,6 +67,11 @@ def __init__(self):
self.lte_last_band_ind = [0, 0]
self.lte_last_tcrnti = [1, 1]

# Serving-cell identity cache (per radio) for --cell-kv, populated from
# LTE/NR RRC SCell Info packets and joined onto ML1 measurement reports.
self.lte_serving_cell = [{}, {}]
self.nr_serving_cell = [{}, {}]

self.io_device: AbstractIO
self.writer: AbstractWriter
self.parse_msgs = False
Expand All @@ -76,6 +81,7 @@ def __init__(self):
self.emr_id_range = []
self.log_id_range = {}
self.cacombos = False
self.cell_kv = False
self.combine_stdout = False
self.check_crc = True
self.layers = []
Expand Down Expand Up @@ -288,6 +294,8 @@ def set_parameter(self, params: dict[str, Any]) -> 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':
Expand Down
Loading