UE/gNB, multiple UEs, and schedulers

Choose between a one-UE real-PHY stack and a faster MAC/FAPI simulation with up to 16 UEs. Both live in one Python application.

One UE through real PHY

"""Attach one UE and exchange SDUs through the complete PHY stack mode."""

from oai_python.bindings.nr.phy.stack import NrPhyStack


def main() -> None:
    with NrPhyStack(history_limit=64) as simulation:
        simulation.start()
        attached = simulation.run_until_connected(140)
        assert attached.terminal_state == "connected"

        simulation.enqueue_downlink_sdu(0, b"hello from the gNB")
        simulation.enqueue_uplink_sdu(0, b"hello from the UE")
        for _ in range(500):
            simulation.step()
            result = simulation.result()
            if result.downlink_sdus and result.uplink_sdus:
                break
        else:
            raise AssertionError("full-PHY stack traffic did not complete")

    print(f"connected RNTI: {result.connected_rntis[0]:#06x}")
    print(f"downlink SDU: {result.downlink_sdus[-1]!r}")
    print(f"uplink SDU: {result.uplink_sdus[-1]!r}")


if __name__ == "__main__":
    main()

Successful output reports the connected RNTI and both decoded SDUs. MAC grants drive the supported waveform paths; only receiver results drive feedback.

Multiple UEs through MAC/FAPI

"""Run four UEs through the abstract MAC/FAPI link mode."""

from oai_python.bindings.nr.mac.link import UeGnbLink


def main() -> None:
    with UeGnbLink(ue_count=4, history_limit=64) as simulation:
        simulation.start()
        attached = simulation.run_until_connected(1_000)
        assert attached.terminal_state == "connected"

        for ue_id in range(4):
            simulation.enqueue_downlink_sdu(ue_id, f"downlink-{ue_id}".encode())
            simulation.enqueue_uplink_sdu(ue_id, f"uplink-{ue_id}".encode())

        for _ in range(1_000):
            simulation.step()
            result = simulation.result()
            if len(result.downlink_sdus) == len(result.uplink_sdus) == 4:
                break
        else:
            raise AssertionError("MAC/FAPI traffic did not complete")

    for delivery in result.deliveries:
        print(
            f"UE {delivery.ue_id} RNTI {delivery.rnti:#06x} "
            f"{delivery.direction} SDU: {delivery.payload!r}"
        )


if __name__ == "__main__":
    main()

Successful output contains one attributed uplink and downlink delivery for each UE. This mode abstracts PHY outcomes as copied FAPI messages, so use it for scheduler and protocol experiments rather than waveform or BLER claims.

Prototype a scheduler

The ordinary pinned-OAI scheduler remains the default. Passing a Python policy opts into a PREPARE → decision → COMMIT seam: OAI retains protocol state, validation, HARQ bookkeeping, and grant construction while Python chooses from the copied observation.

"""Run the high-level UE/gNB MAC workflow with an opt-in Python scheduler."""

from __future__ import annotations

from oai_python.bindings.nr.mac.gnb.scheduler_sdk import (
    RoundRobinPolicy,
    SchedulerDecision,
    SchedulerObservation,
)
from oai_python.bindings.nr.mac.link import UeGnbLink


class MyScheduler:
    """Small policy scaffold: replace the delegate with your own algorithm."""

    def __init__(self) -> None:
        self._baseline = RoundRobinPolicy()
        self.slots_seen = 0

    def decide(self, observation: SchedulerObservation) -> SchedulerDecision:
        self.slots_seen += 1
        return self._baseline.decide(observation)


def main() -> None:
    policy = MyScheduler()
    with UeGnbLink(ue_count=4, scheduler=policy) as link:
        link.start()
        attached = link.run_until_connected(1_000)
        assert attached.terminal_state == "connected"

        for ue_id in range(4):
            link.enqueue_downlink_sdu(ue_id, f"downlink-{ue_id}".encode())
            link.enqueue_uplink_sdu(ue_id, f"uplink-{ue_id}".encode())

        for _ in range(1_000):
            link.step()
            result = link.result()
            if len(result.downlink_sdus) == len(result.uplink_sdus) == 4:
                break
        assert len(result.downlink_sdus) == len(result.uplink_sdus) == 4
        assert policy.slots_seen == result.slots

    print(f"scheduled {len(result.connected_ues)} UEs for {policy.slots_seen} slots")
    for delivery in result.deliveries:
        print(f"UE {delivery.ue_id} {delivery.direction} SDU: {delivery.payload!r}")


if __name__ == "__main__":
    main()

Expected output starts with scheduled 4 UEs and then lists attributed uplink/downlink deliveries. The same policy interface can be passed to NrPhyStack, connecting the decision to supported real-PHY execution.

Several independent cells

"""Run independent MAC/FAPI and full-PHY cells in one Python process."""

from oai_python.bindings.nr.multicell import CellConfiguration, IndependentMultiCell


def main() -> None:
    cells = (
        CellConfiguration(101, mode="mac-fapi", ue_count=2, history_limit=64),
        CellConfiguration(202, mode="full-phy", history_limit=64),
    )
    with IndependentMultiCell(cells) as network:
        network.start()
        attached = network.run_until_connected(512)
        assert all(cell.result.terminal_state == "connected" for cell in attached)

        for cell in cells:
            for ue_id in range(cell.ue_count):
                network.enqueue_downlink_sdu(cell.cell_id, ue_id, b"downlink")
                network.enqueue_uplink_sdu(cell.cell_id, ue_id, b"uplink")

        for _ in range(512):
            network.step()
            current = network.result()
            if all(
                len(cell.result.deliveries) == 2 * network.configuration(cell.cell_id).ue_count
                for cell in current
            ):
                break
        else:
            raise AssertionError("independent-cell traffic did not complete")

    for cell in current:
        for delivery in cell.result.deliveries:
            print(
                f"cell {cell.cell_id}, UE {delivery.ue_id}, "
                f"{delivery.direction} SDU: {delivery.payload!r}"
            )


if __name__ == "__main__":
    main()

Each result carries a cell ID, UE ID, RNTI, direction, and payload. Cells are isolated and stepped sequentially; this release does not model shared RF, inter-cell interference, handover, or coordinated scheduling.