Machine-readable LTE/5G-NR cell measurements (--cell-kv, --meas-gsmtap)

src/scat/main.py

@@ -64,6 +64,8 @@ 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('--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)
@@ -195,6 +197,8 @@ def scat_main():
             'disable-crc-check': args.disable_crc_check,
             '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:

src/scat/parsers/qualcomm/diagltelogparser.py

@@ -186,8 +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)
 
-        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}
+        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):
+            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)
+
+        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)
@@ -218,7 +237,14 @@ 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)
+        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)
 
         for i in range(n_cells):
             n_cell_pkt = pkt_body[pos + 32 * i:pos + 32 * (i + 1)]
@@ -251,8 +277,19 @@ 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)
-        return {'stdout': stdout.rstrip(), 'ts': pkt_ts}
+            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 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('<HHH', cell_bytes[0:6])
@@ -1242,6 +1279,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:
@@ -1258,6 +1304,25 @@ def parse_lte_rrc_cell_info(self, pkt_header, pkt_body: bytes, args: dict):
         else:
             tac_cid_fmt = 'xTAC/xCID: {:x}/{:x}'.format(item.tac, item.cell_id)
 
+        if getattr(self.parent, 'cell_kv', False):
+            # Emit an identity-bearing serving-cell line (CellID/TAC/PLMN) joined to
+            # the last ML1 signal for this cell, so CellID is captured even when the
+            # RRC packet arrives before the next ML1 measurement. Skip when no signal
+            # is cached yet (all-zero metric lines are dropped downstream); the ML1
+            # join then fills identity onto the next measurement instead.
+            sig = self.parent.lte_serving_signal[radio_id] if radio_id in (0, 1) else {}
+            if sig.get('earfcn') == item.dl_earfcn and sig.get('pci') == item.pci:
+                stdout = util.format_cell_kv('lte', 'scell', item.pci, item.dl_earfcn,
+                    item.ul_earfcn, util.dl_earfcn_to_frequency_hz(item.dl_earfcn),
+                    rssi=sig['rssi'], rsrp=sig['rsrp'], rsrq=sig['rsrq'],
+                    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))
+            else:
+                stdout = ''
+            return {'stdout': stdout, 'ts': pkt_ts}
+
         if item.mnc_digit == 2:
             stdout = 'LTE RRC SCell Info: EARFCN: {}/{}, Band: {}, Bandwidth: {}, PCI: {}, MCC: {}, MNC: {:02}, {}'.format(item.dl_earfcn,
                 item.ul_earfcn, item.band, bw_str, item.pci, item.mcc, item.mnc, tac_cid_fmt)