aoa-commissionings/bg22_get-iq_samples.py

import time
import sys
import json
import struct
import queue
from datetime import datetime, timezone

import bgapi

# --- Einstellungen, bei Bedarf anpassen ---

COM_PORT = "COM13"
BAUD_RATE = 115200
XAPI_PATH = r"C:\Users\sandr\.silabs\slt\installs\conan\p\simpleca33d691c539\p\bluetooth_le_host\api\sl_bt.xapi"

OUTPUT_FILE = "iq_samples.jsonl"

SLOT_DURATIONS = 2                      # 1 oder 2 (Mikrosekunden pro Slot)
SWITCHING_PATTERN = bytes(range(0, 16))   # welche Antennen der Reihe nach genutzt werden
CTE_COUNT = 0                             # 0 = unbegrenzt weiter empfangen

# --- Ab hier läuft das Programm, nichts mehr anpassen ---

lib = None
output_file = None
sync_handle = None
already_synced = False
pending_actions = queue.Queue()


def on_event(event):
    """Diese Funktion wird jedes Mal aufgerufen, wenn der BG22 uns etwas meldet."""
    global sync_handle

    name = str(event).split("(")[0]

    if name in ("bt_evt_scanner_scan_report", "bt_evt_scanner_extended_advertisement_report"):
        look_for_periodic_advertising(event)

    elif name in ("bt_evt_sync_opened", "bt_evt_sync_scanner_opened", "bt_evt_periodic_sync_opened"):
        sync_handle = event.sync
        print(f"Verbindung zum Tag hergestellt (sync={sync_handle}). Aktiviere CTE ...")
        pending_actions.put(("enable_cte", sync_handle))

    elif name in ("bt_evt_sync_closed", "bt_evt_sync_scanner_closed", "bt_evt_periodic_sync_closed"):
        print("Verbindung zum Tag wurde beendet.")

    elif name in ("bt_evt_cte_receiver_connectionless_iq_report",
                  "bt_evt_periodic_sync_iq_report",
                  "bt_evt_cte_receiver_iq_report"):
        save_iq_report(event)

    elif hasattr(event, "result") and event.result != 0:
        print(f"Fehler vom BG22: {event}")

    # alles andere (andere Bluetooth-Geräte in der Umgebung) wird ignoriert


def start_scan():
    """Sucht nach Bluetooth-Geräten in der Umgebung."""
    print("Suche nach einem Tag ...")
    lib.bt.scanner.set_parameters(1, 160, 160)
    lib.bt.scanner.start(1, 2)


def look_for_periodic_advertising(event):
    """Prüft, ob ein gefundenes Gerät ein Positionierungs-Tag mit CTE ist."""
    global already_synced

    if already_synced:
        return  # wir haben schon ein Tag gefunden, weitere ignorieren

    periodic_interval = getattr(event, "periodic_interval", None)
    adv_sid = getattr(event, "adv_sid", None)

    if not periodic_interval:
        return  # dieses Gerät sendet kein periodisches Advertising -> kein Tag für uns

    print(f"Tag gefunden! RSSI={event.rssi} dBm, Advertising alle {periodic_interval} ms")

    already_synced = True
    pending_actions.put(("connect_to_tag", event.address, event.address_type, adv_sid))


def enable_cte(sync):
    """Schaltet den Empfang der Richtungs-Signale (CTE) ein."""
    lib.bt.cte_receiver.enable_connectionless_cte(sync, SLOT_DURATIONS, CTE_COUNT, SWITCHING_PATTERN)
    print("CTE-Empfang läuft. Warte auf IQ-Daten ...\n")


def save_iq_report(event):
    """Ein IQ-Paket ist angekommen: dekodieren, anzeigen und speichern."""
    raw_bytes = event.samples
    values = list(struct.unpack(f"{len(raw_bytes)}b", raw_bytes))
    i_values = values[0::2]
    q_values = values[1::2]

    print(f"IQ-Paket #{event.event_counter}  Kanal={event.channel}  "
          f"RSSI={event.rssi} dBm  {len(i_values)} Samples")
    print(f"   Rohbytes:   {raw_bytes[:8]} ...")
    print(f"   I/Q-Werte:  {list(zip(i_values[:4], q_values[:4]))} ...")

    entry = {
        "zeit": datetime.now(timezone.utc).isoformat(),
        "counter": event.event_counter,
        "kanal": event.channel,
        "rssi": event.rssi,
        "slot_dauer_us": SLOT_DURATIONS,
        "antennen_pattern": list(SWITCHING_PATTERN),
        "i": i_values,
        "q": q_values,
    }
    output_file.write(json.dumps(entry) + "\n")
    output_file.flush()


def main():
    global lib, output_file

    output_file = open(OUTPUT_FILE, "a", encoding="utf-8")
    print(f"IQ-Samples werden gespeichert in: {OUTPUT_FILE}")

    connector = bgapi.SerialConnector(COM_PORT, baudrate=BAUD_RATE)
    lib = bgapi.BGLib(connector, XAPI_PATH, on_event)
    lib.open()
    print(f"Verbunden mit {COM_PORT}.")

    lib.bt.system.reboot()
    time.sleep(2.0)

    lib.bt.system.hello()
    print("BG22 antwortet, alles bereit.")
    start_scan()

    # Die Ereignisse (Events) kommen im Hintergrund automatisch bei on_event() an.
    # Hier verarbeiten wir nur die Aktionen, die NICHT direkt aus einem Event
    # heraus aufgerufen werden dürfen (sonst blockiert sich das Programm selbst).
    while True:
        try:
            action, *details = pending_actions.get(timeout=0.1)
        except queue.Empty:
            continue

        if action == "connect_to_tag":
            address, address_type, adv_sid = details
            lib.bt.scanner.stop()
            lib.bt.sync_scanner.open(address, address_type, adv_sid)

        elif action == "enable_cte":
            enable_cte(details[0])


if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        print("\nBeendet.")
    except Exception as error:
        print(f"Es ist ein Fehler aufgetreten: {error}")
        sys.exit(1)
    finally:
        if lib is not None:
            lib.close()
        if output_file is not None:
            output_file.close()