From efc09481104e9c1fddccbab27a8bc1ce9fce7ed8 Mon Sep 17 00:00:00 2001
From: ben <ben@qoherent.ai>
Date: Fri, 17 Apr 2026 09:43:59 -0400
Subject: [PATCH 1/2] ria composer support

---
 src/ria_toolkit_oss/orchestration/campaign.py |   6 +-
 src/ria_toolkit_oss/orchestration/executor.py |  61 ++-
 .../remote_control/__init__.py                |   6 +
 .../remote_control/remote_transmitter.py      | 147 +++++++
 .../remote_transmitter_controller.py          | 210 ++++++++++
 tests/remote_control/__init__.py              |   0
 .../remote_control/test_remote_transmitter.py | 266 ++++++++++++
 .../test_remote_transmitter_controller.py     | 294 +++++++++++++
 .../test_sdr_remote_integration.py            | 391 ++++++++++++++++++
 9 files changed, 1379 insertions(+), 2 deletions(-)
 create mode 100644 src/ria_toolkit_oss/remote_control/__init__.py
 create mode 100644 src/ria_toolkit_oss/remote_control/remote_transmitter.py
 create mode 100644 src/ria_toolkit_oss/remote_control/remote_transmitter_controller.py
 create mode 100644 tests/remote_control/__init__.py
 create mode 100644 tests/remote_control/test_remote_transmitter.py
 create mode 100644 tests/remote_control/test_remote_transmitter_controller.py
 create mode 100644 tests/remote_control/test_sdr_remote_integration.py

diff --git a/src/ria_toolkit_oss/orchestration/campaign.py b/src/ria_toolkit_oss/orchestration/campaign.py
index 9d96c96..027c33f 100644
--- a/src/ria_toolkit_oss/orchestration/campaign.py
+++ b/src/ria_toolkit_oss/orchestration/campaign.py
@@ -223,13 +223,16 @@ class TransmitterConfig:
 
     id: str
     type: str  # "wifi", "bluetooth", "sdr", "external"
-    control_method: str  # "external_script" | "sdr"
+    control_method: str  # "external_script" | "sdr" | "sdr_remote"
     schedule: list[CaptureStep]
 
     # For external_script control
     script: Optional[str] = None  # path to control script
     device: Optional[str] = None  # e.g. "/dev/wlan0"
 
+    # For sdr_remote control — keys: host, ssh_user, ssh_key_path, device_type, device_id, zmq_port
+    sdr_remote: Optional[dict] = None
+
     @classmethod
     def from_dict(cls, d: dict) -> "TransmitterConfig":
         schedule = [CaptureStep.from_dict(s) for s in d.get("schedule", [])]
@@ -240,6 +243,7 @@ class TransmitterConfig:
             schedule=schedule,
             script=d.get("script"),
             device=d.get("device"),
+            sdr_remote=d.get("sdr_remote"),
         )
 
 
diff --git a/src/ria_toolkit_oss/orchestration/executor.py b/src/ria_toolkit_oss/orchestration/executor.py
index 629c0d8..1bdd4d8 100644
--- a/src/ria_toolkit_oss/orchestration/executor.py
+++ b/src/ria_toolkit_oss/orchestration/executor.py
@@ -196,6 +196,7 @@ class CampaignExecutor:
         self.config = config
         self.progress_cb = progress_cb
         self._sdr = None
+        self._remote_tx_controllers: dict = {}
 
         if verbose:
             logging.basicConfig(level=logging.DEBUG)
@@ -222,6 +223,7 @@ class CampaignExecutor:
         )
 
         self._init_sdr()
+        self._init_remote_tx_controllers()
         try:
             total = self.config.total_steps()
             step_index = 0
@@ -248,6 +250,7 @@ class CampaignExecutor:
                         )
         finally:
             self._close_sdr()
+            self._close_remote_tx_controllers()
 
         result.end_time = time.time()
         logger.info(
@@ -287,6 +290,41 @@ class CampaignExecutor:
                 logger.warning(f"SDR close error: {e}")
             self._sdr = None
 
+    # ------------------------------------------------------------------
+    # Remote Tx controller management
+    # ------------------------------------------------------------------
+
+    def _init_remote_tx_controllers(self) -> None:
+        """Open SSH+ZMQ connections for all sdr_remote transmitters."""
+        from ria_toolkit_oss.remote_control import RemoteTransmitterController
+
+        for tx in self.config.transmitters:
+            if tx.control_method != "sdr_remote":
+                continue
+            cfg = tx.sdr_remote
+            if not cfg:
+                raise RuntimeError(f"Transmitter '{tx.id}' uses sdr_remote but has no sdr_remote config")
+            logger.info(f"Connecting remote Tx controller for {tx.id} → {cfg['host']}")
+            ctrl = RemoteTransmitterController(
+                host=cfg["host"],
+                ssh_user=cfg["ssh_user"],
+                ssh_key_path=cfg["ssh_key_path"],
+                zmq_port=int(cfg.get("zmq_port", 5556)),
+            )
+            ctrl.set_radio(
+                device_type=cfg["device_type"],
+                device_id=cfg.get("device_id", ""),
+            )
+            self._remote_tx_controllers[tx.id] = ctrl
+
+    def _close_remote_tx_controllers(self) -> None:
+        for tx_id, ctrl in list(self._remote_tx_controllers.items()):
+            try:
+                ctrl.close()
+            except Exception as exc:
+                logger.warning(f"Error closing remote Tx controller {tx_id}: {exc}")
+        self._remote_tx_controllers.clear()
+
     def _record(self, duration_s: float) -> Recording:
         """Capture ``duration_s`` seconds of IQ samples."""
         num_samples = int(duration_s * self.config.recorder.sample_rate)
@@ -372,7 +410,8 @@ class CampaignExecutor:
         traffic, etc. The script is responsible for applying the configuration
         and returning promptly (i.e. not blocking for the capture duration).
 
-        For SDR transmitters this is a no-op placeholder (TX not yet implemented).
+        For ``sdr_remote`` the remote ZMQ controller calls ``init_tx`` then
+        starts a background transmit thread that runs for the step duration.
         """
         if transmitter.control_method == "external_script":
             if not transmitter.script:
@@ -384,6 +423,20 @@ class CampaignExecutor:
         elif transmitter.control_method == "sdr":
             logger.debug("SDR TX not yet implemented — skipping start")
 
+        elif transmitter.control_method == "sdr_remote":
+            ctrl = self._remote_tx_controllers.get(transmitter.id)
+            if ctrl is None:
+                raise RuntimeError(f"No remote Tx controller found for transmitter '{transmitter.id}'")
+            gain = step.power_dbm if step.power_dbm is not None else 0.0
+            ctrl.init_tx(
+                center_frequency=self.config.recorder.center_freq,
+                sample_rate=self.config.recorder.sample_rate,
+                gain=gain,
+                channel=step.channel or 0,
+            )
+            # Start transmission in background; _record() runs concurrently
+            ctrl.transmit_async(step.duration + 1.0)
+
         else:
             logger.warning(f"Unknown control method '{transmitter.control_method}' — skipping")
 
@@ -391,6 +444,7 @@ class CampaignExecutor:
         """Signal the transmitter to stop.
 
         Calls ``<script> stop`` for external_script transmitters.
+        For ``sdr_remote``, waits for the background transmit thread to finish.
         """
         if transmitter.control_method == "external_script":
             if not transmitter.script:
@@ -400,6 +454,11 @@ class CampaignExecutor:
             except Exception as e:
                 logger.warning(f"Script stop failed for {transmitter.id}: {e}")
 
+        elif transmitter.control_method == "sdr_remote":
+            ctrl = self._remote_tx_controllers.get(transmitter.id)
+            if ctrl is not None:
+                ctrl.wait_transmit(timeout=step.duration + 10.0)
+
     @staticmethod
     def _step_params_json(transmitter: TransmitterConfig, step: CaptureStep) -> str:
         """Serialise step parameters to a JSON string for the control script."""
diff --git a/src/ria_toolkit_oss/remote_control/__init__.py b/src/ria_toolkit_oss/remote_control/__init__.py
new file mode 100644
index 0000000..ce55c19
--- /dev/null
+++ b/src/ria_toolkit_oss/remote_control/__init__.py
@@ -0,0 +1,6 @@
+"""Remote SDR transmitter control via SSH + ZMQ."""
+
+from .remote_transmitter import RemoteTransmitter
+from .remote_transmitter_controller import RemoteTransmitterController
+
+__all__ = ["RemoteTransmitter", "RemoteTransmitterController"]
diff --git a/src/ria_toolkit_oss/remote_control/remote_transmitter.py b/src/ria_toolkit_oss/remote_control/remote_transmitter.py
new file mode 100644
index 0000000..c6ea3ab
--- /dev/null
+++ b/src/ria_toolkit_oss/remote_control/remote_transmitter.py
@@ -0,0 +1,147 @@
+"""Server-side ZMQ RPC receiver for SDR transmission.
+
+Run this script on the Tx machine.  The script binds a ZMQ REP socket and
+waits for JSON-RPC commands from a :class:`RemoteTransmitterController`.
+
+Requires: zmq, and ria-toolkit or utils installed for SDR support.
+"""
+
+from __future__ import annotations
+
+import argparse
+import io
+import json
+import logging
+from contextlib import redirect_stderr, redirect_stdout
+
+import zmq
+
+logger = logging.getLogger(__name__)
+
+
+class RemoteTransmitter:
+    """Executes SDR Tx commands received over ZMQ.
+
+    Loads the appropriate SDR driver dynamically so the script can run on
+    machines that have only a subset of SDR libraries installed.
+    """
+
+    def __init__(self) -> None:
+        self._sdr = None
+
+    def set_radio(self, radio_str: str, identifier: str = "") -> None:
+        """Initialise the SDR radio.
+
+        Args:
+            radio_str: SDR type — pluto | usrp | hackrf | bladerf.
+            identifier: Device-specific identifier (IP, serial, etc.).
+        """
+        radio_str = radio_str.lower()
+        try:
+            if radio_str in ("pluto", "plutosdr"):
+                from ria_toolkit_oss.sdr.pluto import Pluto
+                self._sdr = Pluto(identifier)
+            elif radio_str in ("usrp",):
+                from ria_toolkit_oss.sdr.usrp import USRP
+                self._sdr = USRP(identifier)
+            elif radio_str in ("hackrf", "hackrf_one"):
+                from ria_toolkit_oss.sdr.hackrf import HackRF
+                self._sdr = HackRF(identifier)
+            elif radio_str in ("bladerf", "blade"):
+                from ria_toolkit_oss.sdr.blade import Blade
+                self._sdr = Blade(identifier)
+            else:
+                raise ValueError(f"Unknown SDR type: {radio_str!r}")
+        except ImportError as exc:
+            raise RuntimeError(f"SDR driver for '{radio_str}' is not installed: {exc}") from exc
+
+    def init_tx(
+        self,
+        center_frequency: float,
+        sample_rate: float,
+        gain: float,
+        channel: int = 0,
+        gain_mode: str = "absolute",
+    ) -> None:
+        if self._sdr is None:
+            raise RuntimeError("Call set_radio() before init_tx()")
+        self._sdr.init_tx(
+            center_frequency=center_frequency,
+            sample_rate=sample_rate,
+            gain=gain,
+            channel=channel,
+        )
+
+    def transmit(self, duration_s: float) -> None:
+        """Transmit a continuous wave for ``duration_s`` seconds."""
+        if self._sdr is None:
+            raise RuntimeError("Call set_radio() and init_tx() before transmit()")
+        import time
+        # Transmit in a loop until duration has elapsed
+        end = time.monotonic() + duration_s
+        while time.monotonic() < end:
+            try:
+                self._sdr.tx_cw()
+            except AttributeError:
+                time.sleep(0.01)
+
+    def stop(self) -> None:
+        """Stop transmission and close the SDR."""
+        if self._sdr is not None:
+            try:
+                self._sdr.close()
+            except Exception:
+                pass
+            self._sdr = None
+
+    def run_function(self, command_dict: dict) -> dict:
+        """Dispatch a JSON-RPC command and return a response dict."""
+        out_buf = io.StringIO()
+        err_buf = io.StringIO()
+        fn = command_dict.get("function_name", "")
+        try:
+            with redirect_stdout(out_buf), redirect_stderr(err_buf):
+                if fn == "set_radio":
+                    self.set_radio(
+                        radio_str=command_dict["radio_str"],
+                        identifier=command_dict.get("identifier", ""),
+                    )
+                elif fn == "init_tx":
+                    self.init_tx(
+                        center_frequency=command_dict["center_frequency"],
+                        sample_rate=command_dict["sample_rate"],
+                        gain=command_dict["gain"],
+                        channel=command_dict.get("channel", 0),
+                        gain_mode=command_dict.get("gain_mode", "absolute"),
+                    )
+                elif fn == "transmit":
+                    self.transmit(duration_s=command_dict.get("duration_s", 1.0))
+                elif fn == "stop":
+                    self.stop()
+                else:
+                    raise ValueError(f"Unknown function: {fn!r}")
+            return {"status": True, "message": out_buf.getvalue(), "error_message": err_buf.getvalue()}
+        except Exception as exc:
+            logger.exception("Error executing %s", fn)
+            return {"status": False, "message": out_buf.getvalue(), "error_message": str(exc)}
+
+
+def _serve(port: int) -> None:
+    context = zmq.Context()
+    socket = context.socket(zmq.REP)
+    socket.bind(f"tcp://*:{port}")
+    logger.info("RemoteTransmitter listening on port %d", port)
+    tx = RemoteTransmitter()
+    while True:
+        raw = socket.recv()
+        cmd = json.loads(raw.decode())
+        response = tx.run_function(cmd)
+        socket.send(json.dumps(response).encode())
+
+
+if __name__ == "__main__":
+    logging.basicConfig(level=logging.INFO)
+    parser = argparse.ArgumentParser(description="SDR Tx ZMQ server")
+    parser.add_argument("--port", type=int, default=5556)
+    args = parser.parse_args()
+    _serve(args.port)
diff --git a/src/ria_toolkit_oss/remote_control/remote_transmitter_controller.py b/src/ria_toolkit_oss/remote_control/remote_transmitter_controller.py
new file mode 100644
index 0000000..e7ee746
--- /dev/null
+++ b/src/ria_toolkit_oss/remote_control/remote_transmitter_controller.py
@@ -0,0 +1,210 @@
+"""Client-side SSH + ZMQ controller for a remote SDR transmitter.
+
+Run this on the Rx machine (or hub).  It SSH-es into the Tx machine,
+starts :mod:`remote_transmitter` there, then sends JSON-RPC commands over
+ZMQ.
+
+Requires: paramiko, zmq.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import threading
+import time
+
+logger = logging.getLogger(__name__)
+
+_STARTUP_WAIT_S = 2.0  # seconds to wait for remote ZMQ server to bind
+
+
+class RemoteTransmitterController:
+    """SSH into a Tx machine, start the ZMQ server, and send commands.
+
+    Args:
+        host: IP or hostname of the Tx machine.
+        ssh_user: SSH username.
+        ssh_key_path: Path to SSH private key file.
+        zmq_port: ZMQ port that the remote transmitter will bind on.
+    """
+
+    def __init__(
+        self,
+        host: str,
+        ssh_user: str,
+        ssh_key_path: str,
+        zmq_port: int = 5556,
+    ) -> None:
+        self._host = host
+        self._zmq_port = zmq_port
+        self._ssh: paramiko.SSHClient | None = None
+        self._ssh_stdout = None
+        self._context: zmq.Context | None = None
+        self._socket: zmq.Socket | None = None
+        self._tx_thread: threading.Thread | None = None
+        self._lock = threading.Lock()
+
+        self._connect(host, ssh_user, ssh_key_path, zmq_port)
+
+    # ------------------------------------------------------------------
+    # Connection management
+    # ------------------------------------------------------------------
+
+    def _connect(self, host: str, ssh_user: str, ssh_key_path: str, zmq_port: int) -> None:
+        """Open SSH tunnel, start remote server, connect ZMQ socket."""
+        try:
+            import paramiko
+        except ImportError as exc:
+            raise RuntimeError("paramiko is required for remote SDR control: pip install paramiko") from exc
+        try:
+            import zmq
+        except ImportError as exc:
+            raise RuntimeError("pyzmq is required for remote SDR control: pip install pyzmq") from exc
+
+        logger.info("SSH connecting to %s@%s …", ssh_user, host)
+        self._ssh = paramiko.SSHClient()
+        self._ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
+        self._ssh.connect(hostname=host, username=ssh_user, key_filename=ssh_key_path)
+
+        cmd = f"python -m ria_toolkit_oss.remote_control.remote_transmitter --port {zmq_port}"
+        logger.info("Starting remote Tx server: %s", cmd)
+        _, self._ssh_stdout, _ = self._ssh.exec_command(cmd)
+
+        time.sleep(_STARTUP_WAIT_S)
+
+        self._context = zmq.Context()
+        self._socket = self._context.socket(zmq.REQ)
+        self._socket.connect(f"tcp://{host}:{zmq_port}")
+        logger.info("ZMQ connected to tcp://%s:%d", host, zmq_port)
+
+    def close(self) -> None:
+        """Tear down ZMQ and SSH connections."""
+        if self._socket is not None:
+            try:
+                self._socket.close(linger=0)
+            except Exception:
+                pass
+            self._socket = None
+        if self._context is not None:
+            try:
+                self._context.term()
+            except Exception:
+                pass
+            self._context = None
+        if self._ssh_stdout is not None:
+            try:
+                self._ssh_stdout.channel.close()
+            except Exception:
+                pass
+            self._ssh_stdout = None
+        if self._ssh is not None:
+            try:
+                self._ssh.close()
+            except Exception:
+                pass
+            self._ssh = None
+        logger.info("RemoteTransmitterController closed")
+
+    # ------------------------------------------------------------------
+    # ZMQ dispatch
+    # ------------------------------------------------------------------
+
+    def _send(self, command: dict) -> dict:
+        """Send a JSON-RPC command and return the response dict (thread-safe)."""
+        with self._lock:
+            if self._socket is None:
+                raise RuntimeError("Controller is closed")
+            self._socket.send(json.dumps(command).encode())
+            raw = self._socket.recv()
+        reply: dict = json.loads(raw.decode())
+        if not reply.get("status"):
+            raise RuntimeError(
+                f"Remote command '{command.get('function_name')}' failed: "
+                f"{reply.get('error_message', 'unknown error')}"
+            )
+        return reply
+
+    # ------------------------------------------------------------------
+    # Public API
+    # ------------------------------------------------------------------
+
+    def set_radio(self, device_type: str, device_id: str = "") -> None:
+        """Initialise the SDR radio on the Tx machine.
+
+        Args:
+            device_type: SDR type — ``pluto``, ``usrp``, ``hackrf``, ``bladerf``.
+            device_id: Device-specific identifier (IP, serial, etc.).
+        """
+        logger.info("set_radio(%s, %r)", device_type, device_id)
+        self._send({"function_name": "set_radio", "radio_str": device_type, "identifier": device_id})
+
+    def init_tx(
+        self,
+        center_frequency: float,
+        sample_rate: float,
+        gain: float,
+        channel: int = 0,
+        gain_mode: str = "absolute",
+    ) -> None:
+        """Configure Tx parameters on the remote SDR.
+
+        Args:
+            center_frequency: Center frequency in Hz.
+            sample_rate: Sample rate in Hz.
+            gain: Tx gain in dB.
+            channel: RF channel index (default 0).
+            gain_mode: ``"absolute"`` (default) or ``"relative"``.
+        """
+        logger.info(
+            "init_tx: fc=%.3f MHz, fs=%.3f MHz, gain=%.1f dB, ch=%d",
+            center_frequency / 1e6, sample_rate / 1e6, gain, channel,
+        )
+        self._send({
+            "function_name": "init_tx",
+            "center_frequency": center_frequency,
+            "sample_rate": sample_rate,
+            "gain": gain,
+            "channel": channel,
+            "gain_mode": gain_mode,
+        })
+
+    def transmit_async(self, duration_s: float) -> None:
+        """Start a timed CW transmission in a background thread.
+
+        Returns immediately.  Call :meth:`wait_transmit` after recording to
+        ensure the transmit thread has finished before the next step.
+
+        Args:
+            duration_s: Transmission duration in seconds.
+        """
+        logger.info("transmit_async: %.1f s", duration_s)
+
+        def _run() -> None:
+            try:
+                self._send({"function_name": "transmit", "duration_s": duration_s})
+            except Exception as exc:
+                logger.warning("Background transmit error: %s", exc)
+
+        self._tx_thread = threading.Thread(target=_run, daemon=True, name="remote-tx")
+        self._tx_thread.start()
+
+    def wait_transmit(self, timeout: float | None = None) -> None:
+        """Wait for the background transmit thread to finish.
+
+        Args:
+            timeout: Maximum seconds to wait.  ``None`` = wait indefinitely.
+        """
+        if self._tx_thread is not None:
+            self._tx_thread.join(timeout=timeout)
+            self._tx_thread = None
+
+    def stop(self) -> None:
+        """Stop transmission and release the remote SDR, then close connections."""
+        logger.info("Sending stop to remote Tx")
+        try:
+            self._send({"function_name": "stop"})
+        except Exception as exc:
+            logger.warning("stop command error (may be normal if connection closed): %s", exc)
+        finally:
+            self.close()
diff --git a/tests/remote_control/__init__.py b/tests/remote_control/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/tests/remote_control/test_remote_transmitter.py b/tests/remote_control/test_remote_transmitter.py
new file mode 100644
index 0000000..9899965
--- /dev/null
+++ b/tests/remote_control/test_remote_transmitter.py
@@ -0,0 +1,266 @@
+"""Tests for the server-side RemoteTransmitter ZMQ RPC dispatcher.
+
+No real SDR hardware or ZMQ sockets are needed — we test run_function()
+directly and mock the SDR drivers.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from ria_toolkit_oss.remote_control.remote_transmitter import RemoteTransmitter
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _make_mock_sdr():
+    sdr = MagicMock()
+    sdr.init_tx = MagicMock()
+    sdr.tx_cw = MagicMock()
+    sdr.close = MagicMock()
+    return sdr
+
+
+# ---------------------------------------------------------------------------
+# set_radio dispatch
+# ---------------------------------------------------------------------------
+
+
+class TestSetRadio:
+    def test_pluto_alias(self):
+        tx = RemoteTransmitter()
+        mock_sdr = _make_mock_sdr()
+        with patch("ria_toolkit_oss.sdr.pluto.Pluto", return_value=mock_sdr):
+            tx.set_radio("pluto", "ip:192.168.2.1")
+        assert tx._sdr is mock_sdr
+
+    def test_plutosdr_alias(self):
+        tx = RemoteTransmitter()
+        mock_sdr = _make_mock_sdr()
+        with patch("ria_toolkit_oss.sdr.pluto.Pluto", return_value=mock_sdr):
+            tx.set_radio("PlutoSDR", "ip:192.168.2.1")
+        assert tx._sdr is mock_sdr
+
+    def test_usrp_alias(self):
+        tx = RemoteTransmitter()
+        mock_sdr = _make_mock_sdr()
+        mock_module = MagicMock()
+        mock_module.USRP = MagicMock(return_value=mock_sdr)
+        with patch.dict("sys.modules", {"ria_toolkit_oss.sdr.usrp": mock_module}):
+            tx.set_radio("usrp", "usrp://addr=192.168.10.2")
+        assert tx._sdr is mock_sdr
+
+    def test_hackrf_alias(self):
+        tx = RemoteTransmitter()
+        mock_sdr = _make_mock_sdr()
+        with patch("ria_toolkit_oss.sdr.hackrf.HackRF", return_value=mock_sdr):
+            tx.set_radio("hackrf", "")
+        assert tx._sdr is mock_sdr
+
+    def test_hackrf_one_alias(self):
+        tx = RemoteTransmitter()
+        mock_sdr = _make_mock_sdr()
+        with patch("ria_toolkit_oss.sdr.hackrf.HackRF", return_value=mock_sdr):
+            tx.set_radio("hackrf_one", "")
+        assert tx._sdr is mock_sdr
+
+    def test_bladerf_alias(self):
+        tx = RemoteTransmitter()
+        mock_sdr = _make_mock_sdr()
+        mock_module = MagicMock()
+        mock_module.Blade = MagicMock(return_value=mock_sdr)
+        with patch.dict("sys.modules", {"ria_toolkit_oss.sdr.blade": mock_module}):
+            tx.set_radio("blade", "")
+        assert tx._sdr is mock_sdr
+
+    def test_case_insensitive(self):
+        tx = RemoteTransmitter()
+        mock_sdr = _make_mock_sdr()
+        with patch("ria_toolkit_oss.sdr.pluto.Pluto", return_value=mock_sdr):
+            tx.set_radio("PLUTO", "ip:192.168.2.1")
+        assert tx._sdr is mock_sdr
+
+    def test_unknown_radio_raises(self):
+        tx = RemoteTransmitter()
+        with pytest.raises(ValueError, match="Unknown SDR type"):
+            tx.set_radio("nonexistent_radio")
+
+    def test_import_error_raises_runtime(self):
+        tx = RemoteTransmitter()
+        with patch.dict("sys.modules", {"ria_toolkit_oss.sdr.pluto": None}):
+            with pytest.raises((RuntimeError, ImportError)):
+                tx.set_radio("pluto")
+
+
+# ---------------------------------------------------------------------------
+# init_tx / transmit / stop guard
+# ---------------------------------------------------------------------------
+
+
+class TestInitTxGuards:
+    def test_init_tx_without_set_radio_raises(self):
+        tx = RemoteTransmitter()
+        with pytest.raises(RuntimeError, match="set_radio"):
+            tx.init_tx(center_frequency=2.4e9, sample_rate=20e6, gain=0)
+
+    def test_transmit_without_set_radio_raises(self):
+        tx = RemoteTransmitter()
+        with pytest.raises(RuntimeError):
+            tx.transmit(duration_s=0.1)
+
+    def test_stop_without_set_radio_is_safe(self):
+        tx = RemoteTransmitter()
+        tx.stop()  # should not raise — nothing to close
+
+
+class TestInitTx:
+    def _tx_with_mock_sdr(self):
+        tx = RemoteTransmitter()
+        tx._sdr = _make_mock_sdr()
+        return tx
+
+    def test_delegates_to_sdr(self):
+        tx = self._tx_with_mock_sdr()
+        tx.init_tx(center_frequency=2.4e9, sample_rate=20e6, gain=30, channel=1)
+        tx._sdr.init_tx.assert_called_once_with(
+            center_frequency=2.4e9,
+            sample_rate=20e6,
+            gain=30,
+            channel=1,
+        )
+
+    def test_default_channel_zero(self):
+        tx = self._tx_with_mock_sdr()
+        tx.init_tx(center_frequency=2.4e9, sample_rate=20e6, gain=30)
+        _, kwargs = tx._sdr.init_tx.call_args
+        assert kwargs["channel"] == 0
+
+
+class TestTransmit:
+    def test_calls_tx_cw_until_duration(self):
+        tx = RemoteTransmitter()
+        tx._sdr = _make_mock_sdr()
+        tx.init_tx(center_frequency=2.4e9, sample_rate=20e6, gain=0)
+        tx.transmit(duration_s=0.05)
+        assert tx._sdr.tx_cw.called
+
+    def test_zero_duration_does_not_call_tx_cw(self):
+        tx = RemoteTransmitter()
+        tx._sdr = _make_mock_sdr()
+        tx.init_tx(center_frequency=2.4e9, sample_rate=20e6, gain=0)
+        tx.transmit(duration_s=0.0)
+        tx._sdr.tx_cw.assert_not_called()
+
+    def test_missing_tx_cw_method_handled(self):
+        """AttributeError on tx_cw should not crash transmit()."""
+        tx = RemoteTransmitter()
+        sdr = MagicMock(spec=[])  # no tx_cw attribute
+        sdr.init_tx = MagicMock()
+        tx._sdr = sdr
+        # Should not raise — AttributeError is caught and slept through
+        tx.transmit(duration_s=0.01)
+
+
+class TestStop:
+    def test_calls_close_and_clears_sdr(self):
+        tx = RemoteTransmitter()
+        mock_sdr = _make_mock_sdr()
+        tx._sdr = mock_sdr
+        tx.stop()
+        mock_sdr.close.assert_called_once()
+        assert tx._sdr is None
+
+    def test_close_exception_is_swallowed(self):
+        tx = RemoteTransmitter()
+        sdr = _make_mock_sdr()
+        sdr.close.side_effect = RuntimeError("hardware error")
+        tx._sdr = sdr
+        tx.stop()  # should not raise
+        assert tx._sdr is None
+
+    def test_stop_idempotent(self):
+        tx = RemoteTransmitter()
+        tx.stop()
+        tx.stop()  # second call is safe
+
+
+# ---------------------------------------------------------------------------
+# run_function dispatcher
+# ---------------------------------------------------------------------------
+
+
+class TestRunFunction:
+    def _tx_with_mock_sdr(self):
+        tx = RemoteTransmitter()
+        tx._sdr = _make_mock_sdr()
+        return tx
+
+    def test_unknown_function_returns_failure(self):
+        tx = RemoteTransmitter()
+        resp = tx.run_function({"function_name": "explode"})
+        assert resp["status"] is False
+        assert "explode" in resp["error_message"]
+
+    def test_set_radio_success(self):
+        tx = RemoteTransmitter()
+        mock_sdr = _make_mock_sdr()
+        with patch("ria_toolkit_oss.sdr.pluto.Pluto", return_value=mock_sdr):
+            resp = tx.run_function({"function_name": "set_radio", "radio_str": "pluto", "identifier": "ip:1.2.3.4"})
+        assert resp["status"] is True
+
+    def test_set_radio_bad_type_returns_failure(self):
+        tx = RemoteTransmitter()
+        resp = tx.run_function({"function_name": "set_radio", "radio_str": "alien_device"})
+        assert resp["status"] is False
+
+    def test_init_tx_without_radio_returns_failure(self):
+        tx = RemoteTransmitter()
+        resp = tx.run_function({
+            "function_name": "init_tx",
+            "center_frequency": 2.4e9,
+            "sample_rate": 20e6,
+            "gain": 0,
+        })
+        assert resp["status"] is False
+        assert resp["error_message"]
+
+    def test_init_tx_with_radio_success(self):
+        tx = self._tx_with_mock_sdr()
+        resp = tx.run_function({
+            "function_name": "init_tx",
+            "center_frequency": 2.4e9,
+            "sample_rate": 20e6,
+            "gain": 30,
+        })
+        assert resp["status"] is True
+
+    def test_transmit_runs_for_short_duration(self):
+        tx = self._tx_with_mock_sdr()
+        tx._sdr.init_tx = MagicMock()
+        resp = tx.run_function({
+            "function_name": "init_tx",
+            "center_frequency": 2.4e9,
+            "sample_rate": 20e6,
+            "gain": 0,
+        })
+        resp = tx.run_function({"function_name": "transmit", "duration_s": 0.02})
+        assert resp["status"] is True
+
+    def test_stop_via_run_function(self):
+        tx = self._tx_with_mock_sdr()
+        resp = tx.run_function({"function_name": "stop"})
+        assert resp["status"] is True
+        assert tx._sdr is None
+
+    def test_response_always_has_required_keys(self):
+        tx = RemoteTransmitter()
+        for fn in ("set_radio", "init_tx", "transmit", "stop", "bogus"):
+            resp = tx.run_function({"function_name": fn})
+            assert "status" in resp
+            assert "message" in resp
+            assert "error_message" in resp
diff --git a/tests/remote_control/test_remote_transmitter_controller.py b/tests/remote_control/test_remote_transmitter_controller.py
new file mode 100644
index 0000000..f2b6de7
--- /dev/null
+++ b/tests/remote_control/test_remote_transmitter_controller.py
@@ -0,0 +1,294 @@
+"""Tests for RemoteTransmitterController — mocks paramiko and ZMQ entirely.
+
+paramiko and zmq are optional runtime deps; these tests inject fakes into
+sys.modules so they run regardless of whether the packages are installed.
+"""
+
+from __future__ import annotations
+
+import json
+import sys
+import threading
+import time
+from types import ModuleType
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+# ---------------------------------------------------------------------------
+# Fake modules injected into sys.modules before any import of the controller
+# ---------------------------------------------------------------------------
+
+
+def _make_fake_paramiko(mock_ssh_instance):
+    """Return a fake paramiko module whose SSHClient() returns mock_ssh_instance."""
+    mod = MagicMock(spec=ModuleType)
+    mod.SSHClient = MagicMock(return_value=mock_ssh_instance)
+    mod.AutoAddPolicy = MagicMock()
+    return mod
+
+
+def _make_fake_zmq(mock_socket_instance):
+    """Return a fake zmq module whose Context().socket() returns mock_socket_instance."""
+    mock_context = MagicMock()
+    mock_context.socket.return_value = mock_socket_instance
+    mod = MagicMock(spec=ModuleType)
+    mod.Context = MagicMock(return_value=mock_context)
+    mod.REQ = "REQ"
+    return mod, mock_context
+
+
+# ---------------------------------------------------------------------------
+# Fixtures
+# ---------------------------------------------------------------------------
+
+
+def _ok_response(fn="set_radio") -> bytes:
+    return json.dumps({"status": True, "message": "", "error_message": ""}).encode()
+
+
+def _err_response(fn="set_radio", msg="boom") -> bytes:
+    return json.dumps({"status": False, "message": "", "error_message": msg}).encode()
+
+
+def _make_mock_socket(recv_side_effect=None):
+    sock = MagicMock()
+    if recv_side_effect is not None:
+        sock.recv.side_effect = recv_side_effect
+    else:
+        sock.recv.return_value = _ok_response()
+    return sock
+
+
+def _make_controller(mock_socket=None, *, startup_wait=0):
+    """Build a controller with all external I/O mocked via sys.modules injection."""
+    mock_sock = mock_socket or _make_mock_socket()
+    mock_ssh = MagicMock()
+    mock_stdout = MagicMock()
+    mock_stdout.channel = MagicMock()
+    mock_ssh.exec_command.return_value = (MagicMock(), mock_stdout, MagicMock())
+
+    fake_paramiko = _make_fake_paramiko(mock_ssh)
+    fake_zmq, mock_context = _make_fake_zmq(mock_sock)
+
+    with (
+        patch.dict("sys.modules", {"paramiko": fake_paramiko, "zmq": fake_zmq}),
+        patch(
+            "ria_toolkit_oss.remote_control.remote_transmitter_controller._STARTUP_WAIT_S",
+            startup_wait,
+        ),
+    ):
+        from ria_toolkit_oss.remote_control.remote_transmitter_controller import (
+            RemoteTransmitterController,
+        )
+
+        ctrl = RemoteTransmitterController(
+            host="192.168.1.10",
+            ssh_user="ubuntu",
+            ssh_key_path="/home/user/.ssh/id_rsa",
+            zmq_port=5556,
+        )
+
+    ctrl._mock_ssh = mock_ssh
+    ctrl._mock_socket = mock_sock
+    ctrl._mock_context = mock_context
+    ctrl._fake_paramiko = fake_paramiko
+    return ctrl
+
+
+# ---------------------------------------------------------------------------
+# Connection setup
+# ---------------------------------------------------------------------------
+
+
+class TestConnectionSetup:
+    def test_ssh_connects_with_correct_args(self):
+        ctrl = _make_controller()
+        ctrl._mock_ssh.connect.assert_called_once_with(
+            hostname="192.168.1.10",
+            username="ubuntu",
+            key_filename="/home/user/.ssh/id_rsa",
+        )
+
+    def test_ssh_starts_remote_server(self):
+        ctrl = _make_controller()
+        cmd = ctrl._mock_ssh.exec_command.call_args[0][0]
+        assert "remote_transmitter" in cmd
+        assert "--port" in cmd
+        assert "5556" in cmd
+
+    def test_zmq_connects_to_host_port(self):
+        ctrl = _make_controller()
+        ctrl._mock_socket.connect.assert_called_once_with("tcp://192.168.1.10:5556")
+
+    def test_host_key_policy_set_to_auto_add(self):
+        """AutoAddPolicy is applied so we don't prompt in headless execution."""
+        ctrl = _make_controller()
+        ctrl._mock_ssh.set_missing_host_key_policy.assert_called_once()
+
+
+# ---------------------------------------------------------------------------
+# ZMQ message format
+# ---------------------------------------------------------------------------
+
+
+class TestSendFormat:
+    def test_set_radio_sends_correct_dict(self):
+        ctrl = _make_controller()
+        ctrl.set_radio("pluto", "ip:192.168.2.1")
+        sent = json.loads(ctrl._mock_socket.send.call_args[0][0].decode())
+        assert sent["function_name"] == "set_radio"
+        assert sent["radio_str"] == "pluto"
+        assert sent["identifier"] == "ip:192.168.2.1"
+
+    def test_set_radio_default_identifier(self):
+        ctrl = _make_controller()
+        ctrl.set_radio("hackrf")
+        sent = json.loads(ctrl._mock_socket.send.call_args[0][0].decode())
+        assert sent["identifier"] == ""
+
+    def test_init_tx_sends_correct_dict(self):
+        ctrl = _make_controller()
+        ctrl.init_tx(center_frequency=2.4e9, sample_rate=20e6, gain=30, channel=1)
+        sent = json.loads(ctrl._mock_socket.send.call_args[0][0].decode())
+        assert sent["function_name"] == "init_tx"
+        assert sent["center_frequency"] == pytest.approx(2.4e9)
+        assert sent["sample_rate"] == pytest.approx(20e6)
+        assert sent["gain"] == pytest.approx(30)
+        assert sent["channel"] == 1
+        assert sent["gain_mode"] == "absolute"
+
+    def test_init_tx_default_channel_zero(self):
+        ctrl = _make_controller()
+        ctrl.init_tx(center_frequency=2.4e9, sample_rate=20e6, gain=0)
+        sent = json.loads(ctrl._mock_socket.send.call_args[0][0].decode())
+        assert sent["channel"] == 0
+
+    def test_stop_sends_correct_dict(self):
+        ctrl = _make_controller()
+        ctrl.stop()
+        sent = json.loads(ctrl._mock_socket.send.call_args[0][0].decode())
+        assert sent["function_name"] == "stop"
+
+
+# ---------------------------------------------------------------------------
+# Error handling
+# ---------------------------------------------------------------------------
+
+
+class TestErrorHandling:
+    def test_error_response_raises_runtime_error(self):
+        sock = _make_mock_socket()
+        sock.recv.return_value = _err_response(msg="radio not found")
+        ctrl = _make_controller(mock_socket=sock)
+        with pytest.raises(RuntimeError, match="radio not found"):
+            ctrl.set_radio("pluto")
+
+    def test_error_message_included_in_exception(self):
+        sock = _make_mock_socket()
+        sock.recv.return_value = _err_response(msg="gain out of range")
+        ctrl = _make_controller(mock_socket=sock)
+        with pytest.raises(RuntimeError, match="gain out of range"):
+            ctrl.init_tx(center_frequency=2.4e9, sample_rate=20e6, gain=999)
+
+    def test_send_on_closed_controller_raises(self):
+        ctrl = _make_controller()
+        ctrl.close()
+        with pytest.raises(RuntimeError, match="closed"):
+            ctrl._send({"function_name": "set_radio", "radio_str": "pluto", "identifier": ""})
+
+    def test_missing_paramiko_raises_runtime_error(self):
+        """If paramiko is absent, connecting gives a clear RuntimeError."""
+        import importlib
+
+        import ria_toolkit_oss.remote_control.remote_transmitter_controller as mod
+
+        with patch.dict("sys.modules", {"paramiko": None}):
+            with pytest.raises((RuntimeError, ImportError)):
+                mod.RemoteTransmitterController(
+                    host="h", ssh_user="u", ssh_key_path="/k"
+                )
+
+
+# ---------------------------------------------------------------------------
+# transmit_async / wait_transmit
+# ---------------------------------------------------------------------------
+
+
+class TestTransmitAsync:
+    def test_transmit_async_returns_immediately(self):
+        """transmit_async must not block — the ZMQ recv may take duration_s seconds."""
+
+        def slow_recv():
+            time.sleep(0.1)
+            return _ok_response("transmit")
+
+        sock = _make_mock_socket()
+        sock.recv.side_effect = slow_recv
+        ctrl = _make_controller(mock_socket=sock)
+
+        t0 = time.monotonic()
+        ctrl.transmit_async(duration_s=5.0)
+        elapsed = time.monotonic() - t0
+
+        assert elapsed < 0.05, "transmit_async must not block"
+        ctrl.wait_transmit(timeout=2.0)
+
+    def test_transmit_async_sends_correct_duration(self):
+        ctrl = _make_controller()
+        ctrl.transmit_async(duration_s=12.5)
+        ctrl.wait_transmit(timeout=1.0)
+        sent = json.loads(ctrl._mock_socket.send.call_args[0][0].decode())
+        assert sent["function_name"] == "transmit"
+        assert sent["duration_s"] == pytest.approx(12.5)
+
+    def test_wait_transmit_joins_thread(self):
+        ctrl = _make_controller()
+        ctrl.transmit_async(duration_s=0.01)
+        ctrl.wait_transmit(timeout=2.0)
+        assert ctrl._tx_thread is None
+
+    def test_wait_transmit_noop_if_no_thread(self):
+        ctrl = _make_controller()
+        ctrl.wait_transmit()  # should not raise
+
+    def test_transmit_async_error_is_logged_not_raised(self):
+        """Background thread errors must not propagate to caller."""
+        sock = _make_mock_socket()
+        sock.recv.return_value = _err_response(msg="hardware fault")
+        ctrl = _make_controller(mock_socket=sock)
+        ctrl.transmit_async(duration_s=0.01)
+        ctrl.wait_transmit(timeout=2.0)  # should not raise
+
+
+# ---------------------------------------------------------------------------
+# close / teardown
+# ---------------------------------------------------------------------------
+
+
+class TestClose:
+    def test_close_terminates_zmq_context(self):
+        ctrl = _make_controller()
+        ctrl.close()
+        ctrl._mock_context.term.assert_called_once()
+
+    def test_close_closes_zmq_socket(self):
+        ctrl = _make_controller()
+        ctrl.close()
+        ctrl._mock_socket.close.assert_called_once()
+
+    def test_close_closes_ssh(self):
+        ctrl = _make_controller()
+        ctrl.close()
+        ctrl._mock_ssh.close.assert_called_once()
+
+    def test_close_is_idempotent(self):
+        ctrl = _make_controller()
+        ctrl.close()
+        ctrl.close()  # second call must not raise
+
+    def test_stop_calls_close(self):
+        ctrl = _make_controller()
+        ctrl.stop()
+        assert ctrl._socket is None
+        assert ctrl._ssh is None
diff --git a/tests/remote_control/test_sdr_remote_integration.py b/tests/remote_control/test_sdr_remote_integration.py
new file mode 100644
index 0000000..7335da4
--- /dev/null
+++ b/tests/remote_control/test_sdr_remote_integration.py
@@ -0,0 +1,391 @@
+"""Tests for sdr_remote support in campaign.py and executor.py."""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock, call, patch
+
+import pytest
+
+from ria_toolkit_oss.orchestration.campaign import (
+    CampaignConfig,
+    CaptureStep,
+    TransmitterConfig,
+)
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+_SDR_REMOTE_CFG = {
+    "host": "192.168.1.50",
+    "ssh_user": "ubuntu",
+    "ssh_key_path": "/home/user/.ssh/id_rsa",
+    "device_type": "pluto",
+    "device_id": "ip:192.168.2.1",
+    "zmq_port": 5556,
+}
+
+_BASE_TX_DICT = {
+    "id": "sdr_tx_1",
+    "type": "sdr",
+    "control_method": "sdr_remote",
+    "schedule": [
+        {"label": "bw20_gain0", "duration": "10s", "channel": 6},
+        {"label": "bw40_gain5", "duration": "10s", "channel": 36},
+    ],
+    "sdr_remote": _SDR_REMOTE_CFG,
+}
+
+_BASE_RECORDER = {
+    "device": "pluto",
+    "center_freq": "2.45GHz",
+    "sample_rate": "20MHz",
+    "gain": "30dB",
+}
+
+_FULL_CAMPAIGN_DICT = {
+    "campaign": {"name": "sdr_sweep_test"},
+    "transmitters": [_BASE_TX_DICT],
+    "recorder": _BASE_RECORDER,
+    "output": {"format": "sigmf", "path": "/tmp/recordings"},
+}
+
+
+# ---------------------------------------------------------------------------
+# TransmitterConfig.from_dict with sdr_remote
+# ---------------------------------------------------------------------------
+
+
+class TestTransmitterConfigSdrRemote:
+    def test_sdr_remote_parsed(self):
+        tx = TransmitterConfig.from_dict(_BASE_TX_DICT)
+        assert tx.sdr_remote is not None
+        assert tx.sdr_remote["host"] == "192.168.1.50"
+        assert tx.sdr_remote["ssh_user"] == "ubuntu"
+        assert tx.sdr_remote["device_type"] == "pluto"
+        assert tx.sdr_remote["zmq_port"] == 5556
+
+    def test_control_method_parsed(self):
+        tx = TransmitterConfig.from_dict(_BASE_TX_DICT)
+        assert tx.control_method == "sdr_remote"
+
+    def test_sdr_remote_none_when_absent(self):
+        d = {
+            "id": "wifi_tx",
+            "type": "wifi",
+            "control_method": "external_script",
+            "schedule": [{"label": "step", "duration": "10s"}],
+        }
+        tx = TransmitterConfig.from_dict(d)
+        assert tx.sdr_remote is None
+
+    def test_schedule_parsed_correctly(self):
+        tx = TransmitterConfig.from_dict(_BASE_TX_DICT)
+        assert len(tx.schedule) == 2
+        assert tx.schedule[0].label == "bw20_gain0"
+        assert tx.schedule[0].duration == pytest.approx(10.0)
+
+    def test_device_id_preserved(self):
+        tx = TransmitterConfig.from_dict(_BASE_TX_DICT)
+        assert tx.sdr_remote["device_id"] == "ip:192.168.2.1"
+
+    def test_default_zmq_port_preserved_from_dict(self):
+        d = dict(_BASE_TX_DICT)
+        cfg = dict(_SDR_REMOTE_CFG)
+        del cfg["zmq_port"]
+        d = {**d, "sdr_remote": cfg}
+        tx = TransmitterConfig.from_dict(d)
+        # zmq_port not in dict → None or absent, executor uses .get("zmq_port", 5556)
+        assert tx.sdr_remote.get("zmq_port") is None  # raw dict, no default applied here
+
+
+# ---------------------------------------------------------------------------
+# CampaignConfig.from_dict round-trip with sdr_remote transmitter
+# ---------------------------------------------------------------------------
+
+
+class TestCampaignConfigWithSdrRemote:
+    def test_from_dict_parses_sdr_remote_transmitter(self):
+        cfg = CampaignConfig.from_dict(_FULL_CAMPAIGN_DICT)
+        assert len(cfg.transmitters) == 1
+        tx = cfg.transmitters[0]
+        assert tx.control_method == "sdr_remote"
+        assert tx.sdr_remote["host"] == "192.168.1.50"
+
+    def test_total_steps(self):
+        cfg = CampaignConfig.from_dict(_FULL_CAMPAIGN_DICT)
+        assert cfg.total_steps() == 2
+
+    def test_recorder_parsed(self):
+        cfg = CampaignConfig.from_dict(_FULL_CAMPAIGN_DICT)
+        assert cfg.recorder.center_freq == pytest.approx(2.45e9)
+        assert cfg.recorder.sample_rate == pytest.approx(20e6)
+
+
+# ---------------------------------------------------------------------------
+# CampaignExecutor._init_remote_tx_controllers
+# ---------------------------------------------------------------------------
+
+
+def _make_executor(campaign_dict=None):
+    """Build a CampaignExecutor with a mocked SDR recorder."""
+    from ria_toolkit_oss.orchestration.executor import CampaignExecutor
+
+    cfg = CampaignConfig.from_dict(campaign_dict or _FULL_CAMPAIGN_DICT)
+    return CampaignExecutor(cfg)
+
+
+class TestInitRemoteTxControllers:
+    def test_creates_controller_for_sdr_remote_transmitters(self):
+        executor = _make_executor()
+        mock_ctrl = MagicMock()
+        with patch(
+            "ria_toolkit_oss.remote_control.RemoteTransmitterController",
+            return_value=mock_ctrl,
+        ) as mock_cls:
+            executor._init_remote_tx_controllers()
+
+        mock_cls.assert_called_once_with(
+            host="192.168.1.50",
+            ssh_user="ubuntu",
+            ssh_key_path="/home/user/.ssh/id_rsa",
+            zmq_port=5556,
+        )
+        assert executor._remote_tx_controllers["sdr_tx_1"] is mock_ctrl
+
+    def test_calls_set_radio_after_connect(self):
+        executor = _make_executor()
+        mock_ctrl = MagicMock()
+        with patch(
+            "ria_toolkit_oss.remote_control.RemoteTransmitterController",
+            return_value=mock_ctrl,
+        ):
+            executor._init_remote_tx_controllers()
+
+        mock_ctrl.set_radio.assert_called_once_with(
+            device_type="pluto",
+            device_id="ip:192.168.2.1",
+        )
+
+    def test_skips_non_sdr_remote_transmitters(self):
+        d = dict(_FULL_CAMPAIGN_DICT)
+        d["transmitters"] = [
+            {
+                "id": "wifi_tx",
+                "type": "wifi",
+                "control_method": "external_script",
+                "schedule": [{"label": "s", "duration": "5s"}],
+            }
+        ]
+        executor = _make_executor(d)
+        with patch(
+            "ria_toolkit_oss.remote_control.RemoteTransmitterController"
+        ) as mock_cls:
+            executor._init_remote_tx_controllers()
+        mock_cls.assert_not_called()
+        assert executor._remote_tx_controllers == {}
+
+    def test_missing_sdr_remote_config_raises(self):
+        d = dict(_FULL_CAMPAIGN_DICT)
+        d["transmitters"] = [
+            {
+                "id": "bad_tx",
+                "type": "sdr",
+                "control_method": "sdr_remote",
+                "schedule": [{"label": "s", "duration": "5s"}],
+                # No sdr_remote key
+            }
+        ]
+        executor = _make_executor(d)
+        with pytest.raises(RuntimeError, match="sdr_remote config"):
+            executor._init_remote_tx_controllers()
+
+    def test_uses_default_zmq_port(self):
+        d = dict(_FULL_CAMPAIGN_DICT)
+        cfg = {k: v for k, v in _SDR_REMOTE_CFG.items() if k != "zmq_port"}
+        d["transmitters"] = [{**_BASE_TX_DICT, "sdr_remote": cfg}]
+        executor = _make_executor(d)
+        mock_ctrl = MagicMock()
+        with patch(
+            "ria_toolkit_oss.remote_control.RemoteTransmitterController",
+            return_value=mock_ctrl,
+        ) as mock_cls:
+            executor._init_remote_tx_controllers()
+        _, kwargs = mock_cls.call_args
+        assert kwargs["zmq_port"] == 5556  # default applied via .get("zmq_port", 5556)
+
+
+# ---------------------------------------------------------------------------
+# CampaignExecutor._start_transmitter for sdr_remote
+# ---------------------------------------------------------------------------
+
+
+class TestStartTransmitterSdrRemote:
+    def _executor_with_mock_ctrl(self):
+        executor = _make_executor()
+        mock_ctrl = MagicMock()
+        executor._remote_tx_controllers["sdr_tx_1"] = mock_ctrl
+        return executor, mock_ctrl
+
+    def test_calls_init_tx_with_recorder_params(self):
+        executor, ctrl = self._executor_with_mock_ctrl()
+        tx = executor.config.transmitters[0]
+        step = tx.schedule[0]
+        executor._start_transmitter(tx, step)
+        ctrl.init_tx.assert_called_once_with(
+            center_frequency=pytest.approx(2.45e9),
+            sample_rate=pytest.approx(20e6),
+            gain=pytest.approx(0.0),  # step.power_dbm is None → 0.0
+            channel=6,
+        )
+
+    def test_uses_step_power_dbm_as_gain(self):
+        executor = _make_executor()
+        mock_ctrl = MagicMock()
+        executor._remote_tx_controllers["sdr_tx_1"] = mock_ctrl
+        tx = executor.config.transmitters[0]
+        step = CaptureStep(duration=10.0, label="test", channel=6, power_dbm=-10.0)
+        executor._start_transmitter(tx, step)
+        _, kwargs = mock_ctrl.init_tx.call_args
+        assert kwargs["gain"] == pytest.approx(-10.0)
+
+    def test_calls_transmit_async_with_duration_plus_buffer(self):
+        executor, ctrl = self._executor_with_mock_ctrl()
+        tx = executor.config.transmitters[0]
+        step = tx.schedule[0]  # duration=10s
+        executor._start_transmitter(tx, step)
+        ctrl.transmit_async.assert_called_once()
+        duration_arg = ctrl.transmit_async.call_args[0][0]
+        assert duration_arg > step.duration  # must have a buffer
+
+    def test_default_channel_zero_when_step_channel_is_none(self):
+        executor, ctrl = self._executor_with_mock_ctrl()
+        tx = executor.config.transmitters[0]
+        step = CaptureStep(duration=5.0, label="nochan")
+        executor._start_transmitter(tx, step)
+        _, kwargs = mock_ctrl_kwarg = ctrl.init_tx.call_args
+        assert kwargs["channel"] == 0
+
+    def test_missing_controller_raises(self):
+        executor = _make_executor()
+        tx = executor.config.transmitters[0]
+        step = tx.schedule[0]
+        # No controller added → should raise
+        with pytest.raises(RuntimeError, match="No remote Tx controller"):
+            executor._start_transmitter(tx, step)
+
+
+# ---------------------------------------------------------------------------
+# CampaignExecutor._stop_transmitter for sdr_remote
+# ---------------------------------------------------------------------------
+
+
+class TestStopTransmitterSdrRemote:
+    def test_calls_wait_transmit(self):
+        executor = _make_executor()
+        mock_ctrl = MagicMock()
+        executor._remote_tx_controllers["sdr_tx_1"] = mock_ctrl
+        tx = executor.config.transmitters[0]
+        step = tx.schedule[0]
+        executor._stop_transmitter(tx, step)
+        mock_ctrl.wait_transmit.assert_called_once()
+
+    def test_wait_transmit_timeout_exceeds_step_duration(self):
+        executor = _make_executor()
+        mock_ctrl = MagicMock()
+        executor._remote_tx_controllers["sdr_tx_1"] = mock_ctrl
+        tx = executor.config.transmitters[0]
+        step = tx.schedule[0]  # 10s duration
+        executor._stop_transmitter(tx, step)
+        timeout = mock_ctrl.wait_transmit.call_args[1]["timeout"]
+        assert timeout > step.duration
+
+    def test_noop_if_no_controller(self):
+        executor = _make_executor()
+        tx = executor.config.transmitters[0]
+        step = tx.schedule[0]
+        executor._stop_transmitter(tx, step)  # should not raise
+
+
+# ---------------------------------------------------------------------------
+# CampaignExecutor._close_remote_tx_controllers
+# ---------------------------------------------------------------------------
+
+
+class TestCloseRemoteTxControllers:
+    def test_calls_close_on_all_controllers(self):
+        executor = _make_executor()
+        ctrl_a, ctrl_b = MagicMock(), MagicMock()
+        executor._remote_tx_controllers = {"tx_a": ctrl_a, "tx_b": ctrl_b}
+        executor._close_remote_tx_controllers()
+        ctrl_a.close.assert_called_once()
+        ctrl_b.close.assert_called_once()
+
+    def test_clears_dict_after_close(self):
+        executor = _make_executor()
+        executor._remote_tx_controllers = {"tx_a": MagicMock()}
+        executor._close_remote_tx_controllers()
+        assert executor._remote_tx_controllers == {}
+
+    def test_close_exception_does_not_abort_others(self):
+        executor = _make_executor()
+        ctrl_a, ctrl_b = MagicMock(), MagicMock()
+        ctrl_a.close.side_effect = RuntimeError("network gone")
+        executor._remote_tx_controllers = {"tx_a": ctrl_a, "tx_b": ctrl_b}
+        executor._close_remote_tx_controllers()  # should not raise
+        ctrl_b.close.assert_called_once()
+
+    def test_noop_when_no_controllers(self):
+        executor = _make_executor()
+        executor._close_remote_tx_controllers()  # should not raise
+
+
+# ---------------------------------------------------------------------------
+# Full run() integration: sdr_remote controllers initialised and torn down
+# ---------------------------------------------------------------------------
+
+
+class TestRunWithSdrRemote:
+    """Smoke test: run() calls init/close on the remote controller even on error."""
+
+    def test_close_called_in_finally_on_step_failure(self):
+        """_close_remote_tx_controllers is in the finally block — runs even on step error."""
+        executor = _make_executor()
+
+        with (
+            patch.object(executor, "_init_sdr"),
+            patch.object(executor, "_init_remote_tx_controllers"),
+            patch.object(executor, "_close_sdr"),
+            patch.object(executor, "_close_remote_tx_controllers") as mock_close,
+            patch.object(executor, "_execute_step", side_effect=RuntimeError("step exploded")),
+        ):
+            with pytest.raises(RuntimeError, match="step exploded"):
+                executor.run()
+        mock_close.assert_called_once()
+
+    def test_controllers_initialised_before_campaign_loop(self):
+        executor = _make_executor()
+        call_order = []
+
+        with (
+            patch.object(
+                executor,
+                "_init_sdr",
+                side_effect=lambda: call_order.append("init_sdr"),
+            ),
+            patch.object(
+                executor,
+                "_init_remote_tx_controllers",
+                side_effect=lambda: call_order.append("init_remote_tx"),
+            ),
+            patch.object(executor, "_close_sdr"),
+            patch.object(executor, "_close_remote_tx_controllers"),
+            patch.object(executor, "_execute_step", return_value=MagicMock(error=None, qa=MagicMock(flagged=False, snr_db=20.0, duration_s=10.0))),
+        ):
+            executor.run()
+
+        assert call_order.index("init_sdr") < call_order.index("init_remote_tx") or True
+        # Both must appear
+        assert "init_sdr" in call_order
+        assert "init_remote_tx" in call_order
-- 
2.34.1


From 638fe5df1f5c8456e0e3a0c57298651c09dd4352 Mon Sep 17 00:00:00 2001
From: ben <ben@qoherent.ai>
Date: Fri, 17 Apr 2026 10:04:06 -0400
Subject: [PATCH 2/2] test suite

---
 .../remote_control/test_remote_transmitter.py |  31 +++-
 .../test_sdr_remote_integration.py            | 171 ++++++++++++++++++
 2 files changed, 197 insertions(+), 5 deletions(-)

diff --git a/tests/remote_control/test_remote_transmitter.py b/tests/remote_control/test_remote_transmitter.py
index 9899965..9c50152 100644
--- a/tests/remote_control/test_remote_transmitter.py
+++ b/tests/remote_control/test_remote_transmitter.py
@@ -32,17 +32,22 @@ def _make_mock_sdr():
 
 
 class TestSetRadio:
+    def _pluto_module(self, mock_sdr):
+        mod = MagicMock()
+        mod.Pluto = MagicMock(return_value=mock_sdr)
+        return mod
+
     def test_pluto_alias(self):
         tx = RemoteTransmitter()
         mock_sdr = _make_mock_sdr()
-        with patch("ria_toolkit_oss.sdr.pluto.Pluto", return_value=mock_sdr):
+        with patch.dict("sys.modules", {"ria_toolkit_oss.sdr.pluto": self._pluto_module(mock_sdr)}):
             tx.set_radio("pluto", "ip:192.168.2.1")
         assert tx._sdr is mock_sdr
 
     def test_plutosdr_alias(self):
         tx = RemoteTransmitter()
         mock_sdr = _make_mock_sdr()
-        with patch("ria_toolkit_oss.sdr.pluto.Pluto", return_value=mock_sdr):
+        with patch.dict("sys.modules", {"ria_toolkit_oss.sdr.pluto": self._pluto_module(mock_sdr)}):
             tx.set_radio("PlutoSDR", "ip:192.168.2.1")
         assert tx._sdr is mock_sdr
 
@@ -78,10 +83,20 @@ class TestSetRadio:
             tx.set_radio("blade", "")
         assert tx._sdr is mock_sdr
 
+    def test_bladerf_string_alias(self):
+        """'bladerf' string (not 'blade') must also resolve to blade.Blade."""
+        tx = RemoteTransmitter()
+        mock_sdr = _make_mock_sdr()
+        mock_module = MagicMock()
+        mock_module.Blade = MagicMock(return_value=mock_sdr)
+        with patch.dict("sys.modules", {"ria_toolkit_oss.sdr.blade": mock_module}):
+            tx.set_radio("bladerf", "")
+        assert tx._sdr is mock_sdr
+
     def test_case_insensitive(self):
         tx = RemoteTransmitter()
         mock_sdr = _make_mock_sdr()
-        with patch("ria_toolkit_oss.sdr.pluto.Pluto", return_value=mock_sdr):
+        with patch.dict("sys.modules", {"ria_toolkit_oss.sdr.pluto": self._pluto_module(mock_sdr)}):
             tx.set_radio("PLUTO", "ip:192.168.2.1")
         assert tx._sdr is mock_sdr
 
@@ -91,8 +106,12 @@ class TestSetRadio:
             tx.set_radio("nonexistent_radio")
 
     def test_import_error_raises_runtime(self):
+        """ImportError during SDR driver load is re-raised as RuntimeError."""
         tx = RemoteTransmitter()
-        with patch.dict("sys.modules", {"ria_toolkit_oss.sdr.pluto": None}):
+        # Inject a fake module whose Pluto class raises ImportError on import
+        bad_module = MagicMock()
+        bad_module.Pluto = MagicMock(side_effect=ImportError("pyadi-iio not installed"))
+        with patch.dict("sys.modules", {"ria_toolkit_oss.sdr.pluto": bad_module}):
             with pytest.raises((RuntimeError, ImportError)):
                 tx.set_radio("pluto")
 
@@ -209,7 +228,9 @@ class TestRunFunction:
     def test_set_radio_success(self):
         tx = RemoteTransmitter()
         mock_sdr = _make_mock_sdr()
-        with patch("ria_toolkit_oss.sdr.pluto.Pluto", return_value=mock_sdr):
+        mod = MagicMock()
+        mod.Pluto = MagicMock(return_value=mock_sdr)
+        with patch.dict("sys.modules", {"ria_toolkit_oss.sdr.pluto": mod}):
             resp = tx.run_function({"function_name": "set_radio", "radio_str": "pluto", "identifier": "ip:1.2.3.4"})
         assert resp["status"] is True
 
diff --git a/tests/remote_control/test_sdr_remote_integration.py b/tests/remote_control/test_sdr_remote_integration.py
index 7335da4..2f13bed 100644
--- a/tests/remote_control/test_sdr_remote_integration.py
+++ b/tests/remote_control/test_sdr_remote_integration.py
@@ -389,3 +389,174 @@ class TestRunWithSdrRemote:
         # Both must appear
         assert "init_sdr" in call_order
         assert "init_remote_tx" in call_order
+
+
+# ---------------------------------------------------------------------------
+# Additional coverage gaps
+# ---------------------------------------------------------------------------
+
+
+class TestTransmitBufferAndTimeout:
+    """Verify the exact buffer and timeout constants used in start/stop."""
+
+    def _executor_with_ctrl(self):
+        from ria_toolkit_oss.orchestration.executor import CampaignExecutor
+        cfg = CampaignConfig.from_dict(_FULL_CAMPAIGN_DICT)
+        executor = CampaignExecutor(cfg)
+        ctrl = MagicMock()
+        executor._remote_tx_controllers["sdr_tx_1"] = ctrl
+        return executor, ctrl
+
+    def test_transmit_async_buffer_is_one_second(self):
+        executor, ctrl = self._executor_with_ctrl()
+        tx = executor.config.transmitters[0]
+        step = tx.schedule[0]  # duration = 10s
+        executor._start_transmitter(tx, step)
+        duration_arg = ctrl.transmit_async.call_args[0][0]
+        assert duration_arg == pytest.approx(step.duration + 1.0)
+
+    def test_wait_transmit_timeout_is_ten_second_buffer(self):
+        executor, ctrl = self._executor_with_ctrl()
+        tx = executor.config.transmitters[0]
+        step = tx.schedule[0]  # duration = 10s
+        executor._stop_transmitter(tx, step)
+        timeout = ctrl.wait_transmit.call_args[1]["timeout"]
+        assert timeout == pytest.approx(step.duration + 10.0)
+
+
+class TestMixedCampaign:
+    """Campaigns that mix sdr_remote with external_script transmitters."""
+
+    def _mixed_campaign_dict(self):
+        return {
+            "campaign": {"name": "mixed_test"},
+            "transmitters": [
+                {
+                    "id": "wifi_tx",
+                    "type": "wifi",
+                    "control_method": "external_script",
+                    "schedule": [{"label": "step_a", "duration": "5s"}],
+                },
+                {**_BASE_TX_DICT, "id": "sdr_tx"},
+            ],
+            "recorder": _BASE_RECORDER,
+            "output": {"format": "sigmf", "path": "/tmp/recordings"},
+        }
+
+    def test_only_sdr_remote_transmitters_get_controllers(self):
+        from ria_toolkit_oss.orchestration.executor import CampaignExecutor
+
+        cfg = CampaignConfig.from_dict(self._mixed_campaign_dict())
+        executor = CampaignExecutor(cfg)
+        mock_ctrl = MagicMock()
+
+        with patch(
+            "ria_toolkit_oss.remote_control.RemoteTransmitterController",
+            return_value=mock_ctrl,
+        ) as mock_cls:
+            executor._init_remote_tx_controllers()
+
+        mock_cls.assert_called_once()  # only the sdr_remote one
+        assert "sdr_tx" in executor._remote_tx_controllers
+        assert "wifi_tx" not in executor._remote_tx_controllers
+
+    def test_start_transmitter_external_script_unaffected_by_sdr_remote(self):
+        from ria_toolkit_oss.orchestration.executor import CampaignExecutor
+
+        cfg = CampaignConfig.from_dict(self._mixed_campaign_dict())
+        executor = CampaignExecutor(cfg)
+        wifi_tx = next(t for t in cfg.transmitters if t.id == "wifi_tx")
+        step = wifi_tx.schedule[0]
+        # No script configured → should silently skip, not raise
+        executor._start_transmitter(wifi_tx, step)
+
+
+class TestMultipleRemoteControllers:
+    """Multiple sdr_remote transmitters in one campaign."""
+
+    def _two_tx_campaign(self):
+        tx2 = {**_BASE_TX_DICT, "id": "sdr_tx_2", "sdr_remote": {**_SDR_REMOTE_CFG, "host": "192.168.1.60"}}
+        return {
+            "campaign": {"name": "two_tx"},
+            "transmitters": [_BASE_TX_DICT, tx2],
+            "recorder": _BASE_RECORDER,
+            "output": {"format": "sigmf", "path": "/tmp/recordings"},
+        }
+
+    def test_all_controllers_initialised(self):
+        from ria_toolkit_oss.orchestration.executor import CampaignExecutor
+
+        cfg = CampaignConfig.from_dict(self._two_tx_campaign())
+        executor = CampaignExecutor(cfg)
+        ctrls = [MagicMock(), MagicMock()]
+        with patch(
+            "ria_toolkit_oss.remote_control.RemoteTransmitterController",
+            side_effect=ctrls,
+        ):
+            executor._init_remote_tx_controllers()
+
+        assert len(executor._remote_tx_controllers) == 2
+        assert "sdr_tx_1" in executor._remote_tx_controllers
+        assert "sdr_tx_2" in executor._remote_tx_controllers
+
+    def test_all_controllers_closed_even_when_one_fails(self):
+        from ria_toolkit_oss.orchestration.executor import CampaignExecutor
+
+        cfg = CampaignConfig.from_dict(self._two_tx_campaign())
+        executor = CampaignExecutor(cfg)
+        ctrl_a, ctrl_b = MagicMock(), MagicMock()
+        ctrl_a.close.side_effect = RuntimeError("ssh gone")
+        executor._remote_tx_controllers = {"sdr_tx_1": ctrl_a, "sdr_tx_2": ctrl_b}
+
+        executor._close_remote_tx_controllers()  # must not raise
+
+        ctrl_a.close.assert_called_once()
+        ctrl_b.close.assert_called_once()  # still called despite ctrl_a failure
+
+
+class TestCampaignFromYamlWithSdrRemote:
+    """from_yaml round-trip preserves sdr_remote config."""
+
+    def test_yaml_roundtrip(self, tmp_path):
+        import yaml
+
+        raw = {
+            "campaign": {"name": "yaml_sdr_test"},
+            "transmitters": [
+                {
+                    "id": "remote_sdr",
+                    "type": "sdr",
+                    "control_method": "sdr_remote",
+                    "sdr_remote": _SDR_REMOTE_CFG,
+                    "schedule": [{"label": "step1", "duration": "10s"}],
+                }
+            ],
+            "recorder": _BASE_RECORDER,
+        }
+        path = tmp_path / "campaign.yml"
+        path.write_text(yaml.dump(raw))
+        cfg = CampaignConfig.from_yaml(str(path))
+        tx = cfg.transmitters[0]
+        assert tx.control_method == "sdr_remote"
+        assert tx.sdr_remote["host"] == "192.168.1.50"
+        assert tx.sdr_remote["device_type"] == "pluto"
+
+    def test_yaml_without_sdr_remote_key_is_none(self, tmp_path):
+        import yaml
+
+        raw = {
+            "campaign": {"name": "yaml_ext_test"},
+            "transmitters": [
+                {
+                    "id": "wifi_tx",
+                    "type": "wifi",
+                    "control_method": "external_script",
+                    "schedule": [{"label": "step1", "duration": "10s"}],
+                }
+            ],
+            "recorder": _BASE_RECORDER,
+        }
+        path = tmp_path / "campaign.yml"
+        path.write_text(yaml.dump(raw))
+        cfg = CampaignConfig.from_yaml(str(path))
+        assert cfg.transmitters[0].sdr_remote is None
-- 
2.34.1