MCPcopy
hub / github.com/smittix/intercept / _parse_sbs_line

Method _parse_sbs_line

intercept_agent.py:1470–1528  ·  view source on GitHub ↗

Parse SBS format line and update aircraft dict.

(self, line: str)

Source from the content-addressed store, hash-verified

1468 logger.info("ADS-B SBS reader stopped")
1469
1470 def _parse_sbs_line(self, line: str):
1471 """Parse SBS format line and update aircraft dict."""
1472 if not line:
1473 return
1474
1475 parts = line.split(',')
1476 if len(parts) < 11 or parts[0] != 'MSG':
1477 return
1478
1479 msg_type = parts[1]
1480 icao = parts[4].upper()
1481 if not icao:
1482 return
1483
1484 aircraft = self.adsb_aircraft.get(icao) or {'icao': icao}
1485 aircraft['last_seen'] = datetime.now(timezone.utc).isoformat()
1486
1487 # Add GPS
1488 gps_pos = gps_manager.position
1489 if gps_pos:
1490 aircraft['agent_gps'] = gps_pos
1491
1492 try:
1493 if msg_type == '1' and len(parts) > 10:
1494 callsign = parts[10].strip()
1495 if callsign:
1496 aircraft['callsign'] = callsign
1497
1498 elif msg_type == '3' and len(parts) > 15:
1499 if parts[11]:
1500 aircraft['altitude'] = int(float(parts[11]))
1501 if parts[14] and parts[15]:
1502 aircraft['lat'] = float(parts[14])
1503 aircraft['lon'] = float(parts[15])
1504
1505 elif msg_type == '4' and len(parts) > 16:
1506 if parts[12]:
1507 aircraft['speed'] = int(float(parts[12]))
1508 if parts[13]:
1509 aircraft['heading'] = int(float(parts[13]))
1510 if parts[16]:
1511 aircraft['vertical_rate'] = int(float(parts[16]))
1512
1513 elif msg_type == '5' and len(parts) > 11:
1514 if parts[10]:
1515 callsign = parts[10].strip()
1516 if callsign:
1517 aircraft['callsign'] = callsign
1518 if parts[11]:
1519 aircraft['altitude'] = int(float(parts[11]))
1520
1521 elif msg_type == '6' and len(parts) > 17:
1522 if parts[17]:
1523 aircraft['squawk'] = parts[17]
1524
1525 except (ValueError, IndexError):
1526 pass
1527

Calls 1

getMethod · 0.80