diff --git a/common/platform_constants.h b/common/platform_constants.h index 2faadc3dcd..091ad08a4f 100644 --- a/common/platform_constants.h +++ b/common/platform_constants.h @@ -83,7 +83,7 @@ #define MAX_NUM_SLICES 1024 // RLC Entity -#define RLC_TX_MAXSIZE 10000000 +#define RLC_TX_MAXSIZE 2000000 #define RLC_RX_MAXSIZE 10000000 #define SEND_MRW_ON 240 #define MAX_ANT 8 diff --git a/docker/Dockerfile.nakaolab.test b/docker/Dockerfile.nakaolab.test index 3216fa2b96..3f75050a2a 100644 --- a/docker/Dockerfile.nakaolab.test +++ b/docker/Dockerfile.nakaolab.test @@ -36,7 +36,19 @@ RUN mkdir -p /usr/local/etc/flexric /usr/local/lib/flexric && \ echo "/usr/local/lib/flexric" > /etc/ld.so.conf.d/flexric.conf && ldconfig && \ ls -la /usr/local/lib/flexric/ -# Install netcat for antenna control testing -RUN apt-get update && apt-get install -y netcat-openbsd && rm -rf /var/lib/apt/lists/* +# Install netcat for antenna control testing + paramiko for M-Plane NETCONF server +RUN apt-get update && apt-get install -y --no-install-recommends \ + netcat-openbsd python3-pip \ + && pip3 install --break-system-packages paramiko \ + && rm -rf /var/lib/apt/lists/* + +# O-RU M-Plane NETCONF server (o-ran-uplane-conf antenna mask control) +# Listens on port 831, translates NETCONF edit-config to internal antenna control +COPY mplane/mplane_server.py /opt/mplane/mplane_server.py +COPY mplane/wg4-capability.json /opt/mplane/wg4-capability.json +COPY mplane/gnb-mplane-entrypoint.sh /opt/mplane/gnb-mplane-entrypoint.sh +RUN chmod +x /opt/mplane/gnb-mplane-entrypoint.sh + +EXPOSE 831 ENTRYPOINT ["/bin/bash"] \ No newline at end of file diff --git a/mplane/gnb-mplane-entrypoint.sh b/mplane/gnb-mplane-entrypoint.sh new file mode 100644 index 0000000000..b17fdb1dab --- /dev/null +++ b/mplane/gnb-mplane-entrypoint.sh @@ -0,0 +1,26 @@ +#!/bin/bash +# gNB entrypoint with M-Plane NETCONF server +# +# Starts the M-Plane server alongside the gNB process. +# The M-Plane server acts as an O-RU NETCONF endpoint, +# translating o-ran-uplane-conf edit-config to internal antenna control. +# +# Usage: gnb-mplane-entrypoint.sh +# Example: gnb-mplane-entrypoint.sh /oai-ran/.../nr-softmodem -O /opt/.../gnb.conf --rfsim + +set -e + +echo "[gnb-mplane] Starting M-Plane NETCONF server (port ${MPLANE_PORT:-831})..." +python3 /opt/mplane/mplane_server.py & +MPLANE_PID=$! + +# Give the M-Plane server a moment to start +sleep 1 + +if ! kill -0 $MPLANE_PID 2>/dev/null; then + echo "[gnb-mplane] WARNING: M-Plane server failed to start, continuing without it" +fi + +# Run the gNB command (passed as arguments) +echo "[gnb-mplane] Starting gNB: $@" +exec "$@" diff --git a/mplane/mplane_server.py b/mplane/mplane_server.py new file mode 100644 index 0000000000..49bb230429 --- /dev/null +++ b/mplane/mplane_server.py @@ -0,0 +1,602 @@ +#!/usr/bin/env python3 + +""" +O-RU M-Plane NETCONF Server for gNB Antenna Control + +Lightweight NETCONF/SSH server implementing the O-RAN WG4 M-Plane interface +for antenna mask configuration (o-ran-uplane-conf.yang). + +The gNB acts as an O-RU for antenna control purposes: + - Advertises mplane-supported-trx-control-masks (capability report) + - Accepts tx-array-antenna-mask-config via NETCONF edit-config + - Translates received mask to internal antenna control (localhost:9999) + - Sends mplane-trx-control-ant-mask-update notification (logged) + +Protocol: + SSH transport (paramiko) -> NETCONF base:1.0 framing -> XML RPC processing + +Architecture: + o1-adapter (O-DU) --NETCONF:831--> mplane_server.py (O-RU) --localhost:9999--> nr_antenna_control.c +""" + +import base64 +import json +import logging +import os +import re +import signal +import socket +import sys +import threading +import time +import xml.etree.ElementTree as ET + +import paramiko + +# --------------------------------------------------------------------------- +# Configuration +# --------------------------------------------------------------------------- + +MPLANE_PORT = 831 +MPLANE_USER = MPLANE_PASS = "netconf" +GNB_CTRL_HOST = "127.0.0.1" +GNB_CTRL_PORT = 9999 +WG4_CAPABILITY_FILE = "/opt/mplane/wg4-capability.json" +SSH_HOST_KEY = "/opt/mplane/ssh_host_rsa_key" + +NS_NC = "urn:ietf:params:xml:ns:netconf:base:1.0" +NS_UPLANE = "urn:o-ran:uplane-conf:1.0" +NETCONF_END = "]]>]]>" # https://www.rfc-editor.org/rfc/rfc4742 NETCONF base:1.0 message delimiter + +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s %(levelname)s %(name)s: %(message)s", + datefmt="%Y-%m-%d %H:%M:%S", + stream=sys.stdout, +) +log = logging.getLogger("mplane-server") + +# WG4 Capability (O-RU supported masks) +def b64_to_mask(b64_str): + return int.from_bytes(base64.b64decode(b64_str), byteorder="big") + +def mask_to_b64(mask_int): + byte_len = max(1, (mask_int.bit_length() + 7) // 8) + return base64.b64encode(mask_int.to_bytes(byte_len, byteorder="big")).decode() + + +class Capability: + """O-RU capability report: tx-arrays with supported masks.""" + + def __init__(self): + self.tx_arrays = {} # name -> {"masks": [{index, b64, int}], ...} + self.current_config = {} # array-name -> {"index": i, "mask_int": m} + + def load(self, path): + if not (path and os.path.exists(path)): + # Fallback: generate for 4 antennas + masks = [] + for n in range(4, 0, -1): + m = (1 << n) - 1 + masks.append({"index": 4 - n, "b64": mask_to_b64(m), "int": m}) + self.tx_arrays["tx-array-0"] = {"masks": masks, "columns": 4, "layers": 4} + self.current_config["tx-array-0"] = {"index": 0, "mask_int": 0x0F} + log.info("Generated fallback capability (4 antennas)") + return + + with open(path) as f: + data = json.load(f) # returns a dict + for arr in data.get("tx-arrays", []): # blank list for fallback + name = arr["name"] + cap = arr.get("mplane-trx-control-txarr-capability-info", {}) + masks = [] + for m in cap.get("mplane-supported-trx-control-masks", []): + masks.append({ + "index": m["index"], + "b64": m["antenna-mask"], + "int": b64_to_mask(m["antenna-mask"]), + }) + self.tx_arrays[name] = { + "masks": masks, + "columns": arr.get("number-of-columns", 4), + "layers": arr.get("number-of-array-layers", 4), + } + # Default: all active (index 0) + if masks: + self.current_config[name] = { + "index": masks[0]["index"], + "mask_int": masks[0]["int"], + } + log.info("Loaded capability: %d tx-arrays from %s", len(self.tx_arrays), path) + + def resolve_index(self, array_name, index): + arr = self.tx_arrays.get(array_name) + if not arr: + return None, f"unknown tx-array '{array_name}'" + for m in arr["masks"]: + if m["index"] == index: + return m, "" + return None, f"index {index} not in supported masks" + + def resolve_mask(self, array_name, mask_int): + arr = self.tx_arrays.get(array_name) + if not arr: + return None, f"unknown tx-array '{array_name}'" + for m in arr["masks"]: + if m["int"] == mask_int: + return m, "" + return None, f"mask 0x{mask_int:02X} not in supported masks" + + def get_capability_xml(self): + """Generate o-ran-uplane-conf operational data XML for response.""" + parts = [f''] + for name, arr in self.tx_arrays.items(): + parts.append(f"{name}") + parts.append(f"{arr['columns']}") + parts.append(f"{arr['layers']}") + parts.append("") + for m in arr["masks"]: + parts.append("") + parts.append(f"{m['index']}") + parts.append(f"{m['b64']}") + parts.append("") + parts.append("") + parts.append("") + # Current config + for aname, cfg in self.current_config.items(): + parts.append("") + parts.append(f"{aname}") + parts.append(f"{cfg['index']}") + parts.append("") + parts.append("") + return "".join(parts) + + +# --------------------------------------------------------------------------- +# gNB internal antenna control client (localhost:9999) +# --------------------------------------------------------------------------- + +class GnbCtrlClient: + """Connects to the gNB internal test socket to apply antenna masks.""" + + def __init__(self, host, port): + self.host = host + self.port = port + self.sock = None + + def connect(self): + for attempt in range(30): + try: + self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + self.sock.settimeout(5.0) + self.sock.connect((self.host, self.port)) + self._recv() # consume welcome + log.info("Connected to gNB control at %s:%d", self.host, self.port) + return True + except (socket.error, ConnectionRefusedError) as e: + log.debug("gNB ctrl connect attempt %d/30: %s", attempt + 1, e) + if self.sock: + self.sock.close() + self.sock = None + time.sleep(2) + log.warning("Cannot connect to gNB control at %s:%d", self.host, self.port) + return False + + def send_command(self, cmd): + if not self.sock: + if not self.connect(): + return "ERROR: not connected" + try: + self.sock.sendall((cmd + "\n").encode()) + time.sleep(0.1) + return self._recv() + except (socket.error, BrokenPipeError): + log.warning("gNB ctrl connection lost, reconnecting...") + self.sock = None + if not self.connect(): + return "ERROR: reconnect failed" + self.sock.sendall((cmd + "\n").encode()) + time.sleep(0.1) + return self._recv() + + def _recv(self): + self.sock.settimeout(0.5) + data = b"" + try: + while True: + chunk = self.sock.recv(4096) + if not chunk: + break + data += chunk + except socket.timeout: + pass + return data.decode("utf-8", errors="ignore") + + def close(self): + if self.sock: + try: + self.sock.sendall(b"QUIT\n") + except Exception: + pass + self.sock.close() + self.sock = None + + +# --------------------------------------------------------------------------- +# NETCONF Session Handler +# --------------------------------------------------------------------------- + +class NetconfSession: + """Handle a single NETCONF session on an SSH channel.""" + + def __init__(self, channel, capability, gnb_ctrl): + self.channel = channel + self.cap = capability + self.gnb = gnb_ctrl + self.session_id = 1 + self.running = True + + def run(self): + try: + self._send_hello() + hello_msg = self._recv_message() + if hello_msg is None: + log.warning("Client did not send hello") + return + log.info("NETCONF hello exchange complete") + + while self.running: + msg = self._recv_message() + if msg is None: + break + self._process_rpc(msg) + except Exception as e: + log.exception("Session error: %s", e) + finally: + try: + self.channel.close() + except Exception: + pass + log.info("NETCONF session closed") + + def _send_hello(self): + hello = ( + '' + f'' + "" + "urn:ietf:params:netconf:base:1.0" + "urn:ietf:params:netconf:capability:writable-running:1.0" + "" + f"{self.session_id}" + "" + ) + self._send_message(hello) + + def _send_message(self, xml_str): + data = (xml_str + NETCONF_END).encode("utf-8") + self.channel.sendall(data) + + def _recv_message(self): + buf = b"" + while self.running: + try: + self.channel.settimeout(300) + chunk = self.channel.recv(65536) + if not chunk: + return None + buf += chunk + text = buf.decode("utf-8", errors="ignore") + if NETCONF_END in text: + return text.split(NETCONF_END)[0].strip() + except socket.timeout: + continue + except Exception: + return None + return None + + def _process_rpc(self, xml_str): + # Strip XML declaration if present + xml_str = re.sub(r"<\?xml[^?]*\?>", "", xml_str).strip() + try: + root = ET.fromstring(xml_str) + except ET.ParseError as e: + log.warning("Malformed RPC: %s", e) + self._send_error("0", "malformed-message", "rpc") + return + + msg_id = root.get("message-id", "0") + + # Find the operation element + for child in root: + local_tag = child.tag.split("}")[-1] if "}" in child.tag else child.tag + + if local_tag == "edit-config": + self._handle_edit_config(msg_id, child) + elif local_tag == "get": + self._handle_get(msg_id, child) + elif local_tag == "get-config": + self._handle_get_config(msg_id, child) + elif local_tag == "close-session": + self._send_ok(msg_id) + self.running = False + elif local_tag == "lock" or local_tag == "unlock": + self._send_ok(msg_id) + else: + self._send_error(msg_id, "operation-not-supported", "protocol") + return # Only process first operation + + def _handle_edit_config(self, msg_id, edit_elem): + """Handle edit-config for tx-array-antenna-mask-config.""" + # Find element + config_elem = None + for child in edit_elem: + local = child.tag.split("}")[-1] if "}" in child.tag else child.tag + if local == "config": + config_elem = child + break + + if config_elem is None: + self._send_error(msg_id, "missing-element", "protocol", + "Missing element") + return + + # Find tx-array-antenna-mask-config under user-plane-configuration + mask_config = None + for elem in config_elem.iter(): + local = elem.tag.split("}")[-1] if "}" in elem.tag else elem.tag + if local == "tx-array-antenna-mask-config": + mask_config = elem + break + + if mask_config is None: + # No antenna mask config in this edit - accept silently + self._send_ok(msg_id) + return + + # Extract array-name, antenna-bitmask-index or antenna-bitmask + array_name = None + bitmask_index = None + bitmask_b64 = None + + for child in mask_config: + local = child.tag.split("}")[-1] if "}" in child.tag else child.tag + if local == "array-name": + array_name = child.text.strip() if child.text else None + elif local == "antenna-bitmask-index": + bitmask_index = int(child.text.strip()) if child.text else None + elif local == "antenna-bitmask": + bitmask_b64 = child.text.strip() if child.text else None + + if not array_name: + self._send_error(msg_id, "missing-element", "application", + "Missing array-name in tx-array-antenna-mask-config") + return + + # Resolve the mask + resolved = None + if bitmask_index is not None: + resolved, err = self.cap.resolve_index(array_name, bitmask_index) + if not resolved: + self._send_error(msg_id, "invalid-value", "application", err) + return + log.info("M-Plane edit-config: array=%s, antenna-bitmask-index=%d", + array_name, bitmask_index) + elif bitmask_b64 is not None: + mask_int = b64_to_mask(bitmask_b64) + resolved, err = self.cap.resolve_mask(array_name, mask_int) + if not resolved: + self._send_error(msg_id, "invalid-value", "application", err) + return + log.info("M-Plane edit-config: array=%s, antenna-bitmask=%s (0x%02X)", + array_name, bitmask_b64, mask_int) + else: + self._send_error(msg_id, "missing-element", "application", + "Need antenna-bitmask or antenna-bitmask-index") + return + + # Apply to gNB via internal control socket + mask_int = resolved["int"] + resp = self.gnb.send_command(f"SET TX MASK 0x{mask_int:02X}") + log.info("Applied mask 0x%02X to gNB: %s", mask_int, resp.strip()) + + # Update current config + self.cap.current_config[array_name] = { + "index": resolved["index"], + "mask_int": mask_int, + } + + # Log WG4 mplane-trx-control-ant-mask-update notification + log.info("WG4 mplane-trx-control-ant-mask-update: " + "array-name='%s', array-type=TX-ARRAY, " + "updated-antenna-mask-index=%d, updated-antenna-mask=%s", + array_name, resolved["index"], resolved["b64"]) + + self._send_ok(msg_id) + + def _handle_get(self, msg_id, get_elem): + """Handle - return operational data (capability + current config).""" + data_xml = self.cap.get_capability_xml() + reply = ( + f'' + f"{data_xml}" + "" + ) + self._send_message(reply) + + def _handle_get_config(self, msg_id, get_elem): + """Handle - return running config (current mask config).""" + parts = [f''] + for aname, cfg in self.cap.current_config.items(): + parts.append("") + parts.append(f"{aname}") + parts.append(f"{cfg['index']}") + parts.append("") + parts.append("") + reply = ( + f'' + f'{"".join(parts)}' + "" + ) + self._send_message(reply) + + def _send_ok(self, msg_id): + reply = ( + f'' + "" + "" + ) + self._send_message(reply) + + def _send_error(self, msg_id, error_tag, error_type, error_message=""): + msg_part = "" + if error_message: + msg_part = f"{error_message}" + reply = ( + f'' + "" + f"{error_type}" + f"{error_tag}" + "error" + f"{msg_part}" + "" + "" + ) + self._send_message(reply) + + +# --------------------------------------------------------------------------- +# SSH Server Interface +# --------------------------------------------------------------------------- + +class MplaneSSHServer(paramiko.ServerInterface): + def __init__(self): + self.subsystem_event = threading.Event() + + def check_channel_request(self, kind, chanid): + if kind == "session": + return paramiko.OPEN_SUCCEEDED + return paramiko.OPEN_FAILED_ADMINISTRATIVELY_PROHIBITED + + def get_allowed_auths(self, username): + return "password" + + def check_auth_password(self, username, password): + if username == MPLANE_USER and password == MPLANE_PASS: + return paramiko.AUTH_SUCCESSFUL + return paramiko.AUTH_FAILED + + def check_channel_subsystem_request(self, channel, name): + if name == "netconf": + self.subsystem_event.set() + return True + return False + + +# --------------------------------------------------------------------------- +# Main Server +# --------------------------------------------------------------------------- + +def handle_connection(client_sock, addr, host_key, capability, gnb_ctrl): + """Handle a single SSH/NETCONF client connection.""" + log.info("SSH connection from %s:%d", addr[0], addr[1]) + transport = None + try: + transport = paramiko.Transport(client_sock) + transport.add_server_key(host_key) + server = MplaneSSHServer() + transport.start_server(server=server) + + channel = transport.accept(30) + if channel is None: + log.warning("No channel opened by client") + return + + # Wait for NETCONF subsystem request + if not server.subsystem_event.wait(10): + log.warning("Client did not request netconf subsystem") + channel.close() + return + + session = NetconfSession(channel, capability, gnb_ctrl) + session.run() + + except paramiko.SSHException as e: + log.warning("SSH error from %s: %s", addr[0], e) + except Exception as e: + log.exception("Connection error from %s: %s", addr[0], e) + finally: + if transport: + try: + transport.close() + except Exception: + pass + + +def main(): + log.info("O-RU M-Plane NETCONF server starting") + log.info(" Port: %d", MPLANE_PORT) + log.info(" gNB control: %s:%d (internal)", GNB_CTRL_HOST, GNB_CTRL_PORT) + + # Load/generate SSH host key + if os.path.exists(SSH_HOST_KEY): + host_key = paramiko.RSAKey.from_private_key_file(SSH_HOST_KEY) + log.info("Loaded SSH host key from %s", SSH_HOST_KEY) + else: + host_key = paramiko.RSAKey.generate(2048) + os.makedirs(os.path.dirname(SSH_HOST_KEY), exist_ok=True) + host_key.write_private_key_file(SSH_HOST_KEY) + log.info("Generated SSH host key at %s", SSH_HOST_KEY) + + # Load WG4 capability + capability = Capability() + capability.load(WG4_CAPABILITY_FILE) + for name, arr in capability.tx_arrays.items(): + log.info(" tx-array '%s': %d supported masks", name, len(arr["masks"])) + for m in arr["masks"]: + log.info(" index=%d: %s (0x%02X, %d active)", + m["index"], m["b64"], m["int"], bin(m["int"]).count("1")) + + # Connect to gNB internal control + gnb_ctrl = GnbCtrlClient(GNB_CTRL_HOST, GNB_CTRL_PORT) + gnb_ctrl.connect() + + # Create server socket + server_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + server_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + server_sock.bind(("0.0.0.0", MPLANE_PORT)) + server_sock.listen(5) + server_sock.settimeout(1.0) + + shutdown = threading.Event() + + def handle_signal(signum, frame): + log.info("Shutdown signal received") + shutdown.set() + + signal.signal(signal.SIGTERM, handle_signal) + signal.signal(signal.SIGINT, handle_signal) + + log.info("M-Plane NETCONF server ready on port %d", MPLANE_PORT) + + while not shutdown.is_set(): + try: + client_sock, addr = server_sock.accept() + t = threading.Thread( + target=handle_connection, + args=(client_sock, addr, host_key, capability, gnb_ctrl), + daemon=True, + ) + t.start() + except socket.timeout: + continue + except Exception as e: + if not shutdown.is_set(): + log.exception("Accept error: %s", e) + + log.info("Shutting down...") + server_sock.close() + gnb_ctrl.close() + + +if __name__ == "__main__": + main() diff --git a/mplane/wg4-capability.json b/mplane/wg4-capability.json new file mode 100644 index 0000000000..551ba2f4de --- /dev/null +++ b/mplane/wg4-capability.json @@ -0,0 +1,17 @@ +{ + "tx-arrays": [ + { + "name": "tx-array-0", + "number-of-columns": 4, + "number-of-array-layers": 4, + "mplane-trx-control-txarr-capability-info": { + "mplane-supported-trx-control-masks": [ + {"index": 0, "antenna-mask": "Dw=="}, + {"index": 1, "antenna-mask": "Bw=="}, + {"index": 2, "antenna-mask": "Aw=="}, + {"index": 3, "antenna-mask": "AQ=="} + ] + } + } + ] +} diff --git a/openair1/PHY/NR_TRANSPORT/nr_antenna_control.c b/openair1/PHY/NR_TRANSPORT/nr_antenna_control.c index 02750b02ca..d79f64ea4d 100644 --- a/openair1/PHY/NR_TRANSPORT/nr_antenna_control.c +++ b/openair1/PHY/NR_TRANSPORT/nr_antenna_control.c @@ -365,7 +365,7 @@ bool nr_antenna_control_start_test_server(nr_antenna_state_t *ant_state, int por // Bind struct sockaddr_in addr = { .sin_family = AF_INET, - .sin_addr.s_addr = INADDR_ANY, + .sin_addr.s_addr = htonl(INADDR_ANY), .sin_port = htons(g_ctrl.port) }; diff --git a/radio/rfsimulator/simulator.cpp b/radio/rfsimulator/simulator.cpp index ef6be135e6..9003cfcf36 100644 --- a/radio/rfsimulator/simulator.cpp +++ b/radio/rfsimulator/simulator.cpp @@ -545,7 +545,9 @@ static void rfsimulator_readconfig(rfsimulator_state_t *rfsimulator) break; } else if (strcmp(rfsimu_params[p].strlistptr[i], "chanmod") == 0) { init_channelmod(); - load_channellist(rfsimulator->tx_num_channels, + // Use max(tx,rx) for channel model TX dimension so the DL path works + // when peer has more TX antennas than our TX (e.g., UE tx=1, gNB tx=4) + load_channellist(std::max(rfsimulator->tx_num_channels, rfsimulator->rx_num_channels), rfsimulator->rx_num_channels, rfsimulator->sample_rate, rfsimulator->rx_freq, @@ -649,7 +651,7 @@ static int rfsimu_setchanmod_cmd(char *buff, int debug, telnet_printfunc_t prnt, if (b->channel_model->model_name == NULL) continue; if (b->conn_sock >= 0 && (strcmp(b->channel_model->model_name, modelname) == 0)) { - channel_desc_t *newmodel = new_channel_desc_scm(t->tx_num_channels, + channel_desc_t *newmodel = new_channel_desc_scm(std::max(t->tx_num_channels, t->rx_num_channels), t->rx_num_channels, static_cast(channelmod), t->sample_rate,