From facc0663896456f1734e26b722f4e79e391ae39d Mon Sep 17 00:00:00 2001 From: mikecarper Date: Sat, 3 Oct 2026 21:42:54 -0700 Subject: [PATCH] Separate GPS UART ownership and collapse shared Companion ports --- .github/workflows/run-unit-tests.yml | 1 + examples/companion_radio/main.cpp | 26 +- test/test_companion_hardware_serial.py | 2 +- test/test_companion_hardware_serial_gps.py | 494 +++++++++++++++++++++ 4 files changed, 517 insertions(+), 6 deletions(-) create mode 100644 test/test_companion_hardware_serial_gps.py diff --git a/.github/workflows/run-unit-tests.yml b/.github/workflows/run-unit-tests.yml index 891f94bdc..42ad32742 100644 --- a/.github/workflows/run-unit-tests.yml +++ b/.github/workflows/run-unit-tests.yml @@ -133,6 +133,7 @@ jobs: python3 -B test/test_companion_response_bounds.py -v python3 -B test/test_companion_ack_delivery.py -v python3 -B test/test_companion_hardware_serial.py -v + python3 -B test/test_companion_hardware_serial_gps.py -v python3 -B test/test_companion_contact_lifetime.py -v python3 -B test/test_companion_identity_cli.py -v python3 -B test/test_companion_uncached_storage.py -v diff --git a/examples/companion_radio/main.cpp b/examples/companion_radio/main.cpp index df5b985d8..c0ed9d69d 100644 --- a/examples/companion_radio/main.cpp +++ b/examples/companion_radio/main.cpp @@ -192,7 +192,21 @@ static bool isNetworkTerminalActive(); #if defined(SERIAL_RX) #include ArduinoSerialInterface hardware_serial_interface; + #if defined(ENABLE_USB_INTERFACE) && defined(CONFIG_IDF_TARGET_ESP32S3) \ + && (!defined(ARDUINO_USB_CDC_ON_BOOT) || !ARDUINO_USB_CDC_ON_BOOT) + // USB-UART boards can name the same pins as an additional serial transport. + // Give that physical port only one parser and TX owner; native USB is separate. + static constexpr bool companion_serial_shares_usb_port = + SERIAL_RX == SOC_RX0 && SERIAL_TX == SOC_TX0; + #else + static constexpr bool companion_serial_shares_usb_port = false; + #endif + #if defined(CONFIG_IDF_TARGET_ESP32S3) && ENV_INCLUDE_GPS == 1 + // GPS owns UART1. The S3's spare UART2 can use the same Companion pins. + HardwareSerial companion_serial(2); + #else HardwareSerial companion_serial(1); + #endif #endif // platform file system @@ -3490,11 +3504,13 @@ void setup() { // add hardware serial interface #if defined(SERIAL_RX) - companion_serial.setPins(SERIAL_RX, SERIAL_TX); - companion_serial.begin(115200); - hardware_serial_interface.begin(companion_serial); - hardware_serial_interface.enableFlowControl(true); - interface_manager.addInterface(InterfaceType::HardwareSerial, &hardware_serial_interface); + if (!companion_serial_shares_usb_port) { + companion_serial.setPins(SERIAL_RX, SERIAL_TX); + companion_serial.begin(115200); + hardware_serial_interface.begin(companion_serial); + hardware_serial_interface.enableFlowControl(true); + interface_manager.addInterface(InterfaceType::HardwareSerial, &hardware_serial_interface); + } #endif the_mesh.startInterface(interface_manager); diff --git a/test/test_companion_hardware_serial.py b/test/test_companion_hardware_serial.py index 98ca16d31..4c7bea1b7 100644 --- a/test/test_companion_hardware_serial.py +++ b/test/test_companion_hardware_serial.py @@ -175,7 +175,7 @@ class CompanionHardwareSerialTest(unittest.TestCase): binary = work / f"{name}.exe" built = subprocess.run([ compiler, "-std=c++17", "-Werror", *SANITIZERS, - "-DD6=6", "-DD7=7", *[f"-D{define}" for define in defines], + "-DD6=43", "-DD7=44", *[f"-D{define}" for define in defines], f"-I{ROOT / 'test/mocks'}", f"-I{ROOT / 'src'}", str(cpp), str(ROOT / "src/helpers/ArduinoSerialInterface.cpp"), "-o", str(binary), diff --git a/test/test_companion_hardware_serial_gps.py b/test/test_companion_hardware_serial_gps.py new file mode 100644 index 000000000..186e3ddaa --- /dev/null +++ b/test/test_companion_hardware_serial_gps.py @@ -0,0 +1,494 @@ +#!/usr/bin/env python3 +"""Exercise actual Companion and GPS startup with shared hardware UARTs.""" + +from pathlib import Path +import configparser +import json +import os +import re +import shutil +import subprocess +import sys +import tempfile +import unittest + +from test_companion_hardware_serial import conditional_block +from test_replay_reset_integration import extract_braced + + +ROOT = Path(__file__).resolve().parents[1] +SANITIZERS = (["-fsanitize=address,undefined", "-fno-sanitize-recover=all", + "-fno-pie", "-no-pie"] if sys.platform.startswith("linux") else []) + + +HARNESS = r''' +#include +#include +#include +#include +#include +#include +#include +#include +#include "helpers/ArduinoSerialInterface.h" +#include "helpers/MultiSerialInterface.h" +#include "helpers/UsbAsciiBinarySwitch.h" +#ifndef ARDUINO_USB_MODE +#define ARDUINO_USB_MODE 0 +#endif +#define MESH_DEBUG_PRINTLN(...) ((void)0) +@STREAM@ +// HardwareSerial objects selecting the same port share pins, baud, and FIFO, +// as the ESP32 backend indexes _uart_bus_array by UART number. +struct Uart { + int rx = -1, tx = -1; + unsigned baud = 0; + bool installed = false; + BufferStream fifo; +}; +static std::array uarts; +class HardwareSerial : public Stream { +public: + const unsigned port; + explicit HardwareSerial(unsigned selected) : port(selected) { + assert(port < uarts.size()); + } + Uart& state() const { return uarts[port]; } + void setPins(int receive, int transmit) { state().rx = receive; state().tx = transmit; } + void begin(unsigned speed) { state().baud = speed; state().installed = true; } + void end() { state().installed = false; } + int available() override { return state().fifo.available(); } + int availableForWrite() override { + return state().installed ? state().fifo.availableForWrite() : 0; + } + int read() override { return state().fifo.read(); } + int peek() override { return state().fifo.peek(); } + size_t write(uint8_t value) override { return write(&value, 1); } + size_t write(const uint8_t* bytes, size_t len) override { + assert(state().installed); + return state().fifo.write(bytes, len); + } +}; +static HardwareSerial Serial1(1); +@DECLARATION@ +static MultiSerialInterface interface_manager; +static void setupHardwareSerial() { +@SETUP@ +} + +struct Location { + unsigned starts = 0, resets = 0, stops = 0; + void begin() { ++starts; } + void reset() { ++resets; } + void stop() { ++stops; } + bool getGPSPowerSaving() const { return false; } + void stopTimeSync() {} + void setNextGPSOff(unsigned) {} + void setNextWake() {} +}; +struct EnvironmentSensorManager { + bool gps_serial_transport_blocked = false, gps_serial_transport = false; + bool gps_detected = false, gps_active = false, powersaving_enabled = false; + bool telemetry_user_enabled = false; + Location location; + Location* _location = &location; + void resetGpsTelemetryTransportState() { telemetry_user_enabled = false; } + void setGpsTelemetryUserEnabled(bool value) { telemetry_user_enabled = value; } + void armGpsPowerSavingCycle() {} + void initBasicGPS(); + void start_gps(); + void stop_gps(); +}; +#if ENV_INCLUDE_GPS == 1 +@GPS_FUNCTIONS@ +#endif + +static void assertCompanionConfiguration() { +#if defined(SERIAL_RX) + assert(companion_serial.port == EXPECTED_COMPANION_UART); + assert(companion_serial.state().installed); + assert(companion_serial.state().rx == SERIAL_RX); + assert(companion_serial.state().tx == SERIAL_TX); + assert(companion_serial.state().baud == 115200); +#endif +} + +int main() { + setupHardwareSerial(); + interface_manager.enable(); + assertCompanionConfiguration(); +#if ENV_INCLUDE_GPS == 1 + EnvironmentSensorManager sensors; + sensors.initBasicGPS(); + assert(sensors.gps_active && sensors.gps_detected); + assert(sensors.gps_serial_transport && sensors.telemetry_user_enabled); + assert(Serial1.state().installed && Serial1.state().baud == 9600); + assert(Serial1.state().rx == PIN_GPS_TX); + assert(Serial1.state().tx == PIN_GPS_RX); + assert(sensors.location.starts == 1 && sensors.location.resets == 1); +#if defined(SERIAL_RX) && defined(CONFIG_IDF_TARGET_ESP32S3) + assertCompanionConfiguration(); + assert(companion_serial.port != Serial1.port); + + // Route a command and response over Companion without consuming GPS bytes. + const uint8_t command[] = {'<', 1, 0, 0x01}; + companion_serial.state().fifo.push(command, sizeof(command)); + Serial1.state().fifo.push("$GPGGA,untouched\r\n"); + const int gps_bytes = Serial1.available(); + uint8_t received[MAX_FRAME_SIZE] = {}; + assert(interface_manager.checkRecvFrame(received) == 1); + assert(interface_manager.captureReplyRoute() == &hardware_serial_interface); + assert(Serial1.available() == gps_bytes); + companion_serial.state().fifo.write_capacity = 0; + const uint8_t response[] = {0x00}; + assert(interface_manager.writeFrameToRoute(&hardware_serial_interface, + response, sizeof(response)) == sizeof(response)); + assert(hardware_serial_interface.hasPendingIO()); + + // Actual GPS stop/start and rediscovery must leave the queued UART response, + // original pins and 115200 baud intact, and must not reset the GPS choice. + sensors.stop_gps(); + assert(!sensors.gps_active && sensors.location.stops == 1); + sensors.start_gps(); + assert(sensors.gps_active); + sensors.initBasicGPS(); + assertCompanionConfiguration(); + assert(Serial1.state().baud == 9600); + assert(Serial1.available() == gps_bytes); + companion_serial.state().fifo.write_capacity = 128; + interface_manager.loop(); + const std::vector expected = {'>', 1, 0, 0x00}; + assert(companion_serial.state().fifo.output == expected); + assert(Serial1.state().fifo.output.empty()); + assert(!hardware_serial_interface.hasPendingIO()); + + // Reinitializing Companion must not change the GPS's existing UART1 state. + setupHardwareSerial(); + assertCompanionConfiguration(); + assert(Serial1.state().installed && Serial1.state().baud == 9600); + assert(Serial1.state().rx == PIN_GPS_TX && Serial1.state().tx == PIN_GPS_RX); +#endif +#else + assertCompanionConfiguration(); +#endif + std::puts("PASS: actual Companion/GPS UART ownership"); +} +''' + + +GPIO_UART = r''' +static std::array tx_owner = [] { + std::array owners; + owners.fill(-1); + return owners; +}(); +static std::array, 64> physical_output; +class HardwareSerial : public Stream { +public: + const unsigned port; + explicit HardwareSerial(unsigned selected) : port(selected) { + assert(port < uarts.size()); + if (port == 0) setPins(SOC_RX0, SOC_TX0); + } + Uart& state() const { return uarts[port]; } + void setPins(int receive, int transmit) { + state().rx = receive; + state().tx = transmit; + tx_owner.at(transmit) = int(port); + } + void begin(unsigned speed) { state().baud = speed; state().installed = true; } + int available() override { return state().fifo.available(); } + int availableForWrite() override { + return state().installed ? state().fifo.availableForWrite() : 0; + } + int read() override { return state().fifo.read(); } + int peek() override { return state().fifo.peek(); } + size_t write(uint8_t value) override { return write(&value, 1); } + size_t write(const uint8_t* bytes, size_t len) override { + assert(state().installed); + const size_t accepted = state().fifo.write(bytes, len); + // A GPIO has only one selected output signal, even if two UARTs have + // independently installed drivers and accept data into their TX FIFOs. + if (tx_owner.at(state().tx) == int(port)) { + auto& wire = physical_output.at(state().tx); + wire.insert(wire.end(), bytes, bytes + accepted); + } + return accepted; + } +}; +static void physicalInput(int gpio, const uint8_t* bytes, size_t len) { + // One input pad can feed multiple UART RX signals. Do not model independent + // UART FIFOs as independent physical ports when their RX pins are identical. + for (auto& uart : uarts) { + if (uart.installed && uart.rx == gpio) uart.fifo.push(bytes, len); + } +} +static const uint8_t SYMBOL_RX = 44, SYMBOL_TX = 43; +// The XIAO ESP32-S3 core names GPIO43/44 as D6/D7, not GPIO6/7. +static const uint8_t D6 = 43, D7 = 44; +''' + + +GPIO_USB = r''' +#if ARDUINO_USB_CDC_ON_BOOT +struct NativeUsb : BufferStream { + unsigned baud = 0; + void begin(unsigned speed) { baud = speed; } + explicit operator bool() const { return true; } +} Serial; +#else +static HardwareSerial Serial(0); +#endif +#if defined(ENABLE_USB_INTERFACE) +ArduinoSerialInterface usb_serial_interface; +static const char USB_TERMINAL_START_TOKEN[] = "+++MESHCORE-TERM-START"; +static bool usb_protocol_initialized = false; +static unsigned default_sessions = 0; +static mesh::UsbHostPresenceDebouncer usb_hwcdc_host_presence; +static constexpr unsigned USB_FRAME_REPLY_GRACE_MS = 2000; +static constexpr unsigned USB_CLIENT_IDLE_TIMEOUT = 600000; +static constexpr unsigned USB_HOST_LOSS_EDGE_MS = 100; +static constexpr unsigned USB_HOST_LOSS_GRACE_MS = 2000; +struct Board { bool isUsbHostConnected() const { return true; } } board; +struct Mesh { + bool hasFiniteDelayedReplyForRoute(BaseSerialInterface*) const { return false; } +} the_mesh; +namespace mesh { +Stream& usbCompanionPort() { return Serial; } +} +static bool acceptUsbBinaryStartupFrame(uint32_t) { return true; } +static void beginUsbDefaultSession() { ++default_sessions; } +#endif +''' + + +GPIO_MAIN = r''' +static void setupPrimaryUsb() { +@USB_SETUP@ +} + +int main() { + // setup() always begins the primary Serial before the interface blocks. + Serial.begin(115200); + setupPrimaryUsb(); + setupHardwareSerial(); + interface_manager.enable(); + assert(companion_serial_shares_usb_port == bool(EXPECTED_ALIAS)); + assert(interface_manager.isInterfaceConnected(InterfaceType::HardwareSerial) + == !bool(EXPECTED_ALIAS)); +#if defined(ENABLE_USB_INTERFACE) + assert(usb_protocol_initialized && default_sessions == 1); +#if ARDUINO_USB_CDC_ON_BOOT && ARDUINO_USB_MODE == 1 + // HWCDC has no DTR: a physical host alone is not yet a Binary client. + assert(!interface_manager.isInterfaceConnected(InterfaceType::USB)); +#else + assert(interface_manager.isInterfaceConnected(InterfaceType::USB)); +#endif + usb_serial_interface.setPassthroughMode(false); // Deliberate Binary owner. +#endif +#if ENV_INCLUDE_GPS == 1 + EnvironmentSensorManager sensors; + sensors.initBasicGPS(); + assert(sensors.gps_active && Serial1.state().baud == 9600); +#endif + if (!EXPECTED_ALIAS) assertCompanionConfiguration(); + else assert(!companion_serial.state().installed); + + const uint8_t command[] = {'<', 1, 0, 0x01}; + const uint8_t response[] = {0x00}; + uint8_t received[MAX_FRAME_SIZE] = {}; + const std::vector expected = {'>', 1, 0, 0x00}; +#if defined(ENABLE_USB_INTERFACE) +#if ARDUINO_USB_CDC_ON_BOOT + Serial.push(command, sizeof(command)); +#else + physicalInput(SOC_RX0, command, sizeof(command)); +#endif + assert(interface_manager.checkRecvFrame(received) == 1); + assert(interface_manager.captureReplyRoute() == &usb_serial_interface); + assert(interface_manager.isInterfaceConnected(InterfaceType::USB)); + // An alias must not leave a second parser holding the same command. + assert(interface_manager.checkRecvFrame(received) == 0); + assert(interface_manager.writeFrameToRoute(&usb_serial_interface, + response, sizeof(response)) == sizeof(response)); +#if ARDUINO_USB_CDC_ON_BOOT + assert(Serial.output == expected); +#else + assert(physical_output.at(SOC_TX0) == expected); + assert(tx_owner.at(SOC_TX0) == 0); + assert(Serial.state().rx == SOC_RX0 && Serial.state().tx == SOC_TX0); + assert(Serial.state().baud == 115200); +#endif +#endif + + if (!EXPECTED_ALIAS) { + physicalInput(SERIAL_RX, command, sizeof(command)); + assert(interface_manager.checkRecvFrame(received) == 1); + assert(interface_manager.captureReplyRoute() == &hardware_serial_interface); + assert(interface_manager.checkRecvFrame(received) == 0); + companion_serial.state().fifo.write_capacity = 0; + assert(interface_manager.writeFrameToRoute(&hardware_serial_interface, + response, sizeof(response)) == sizeof(response)); + assert(hardware_serial_interface.hasPendingIO()); + companion_serial.state().fifo.write_capacity = 128; + interface_manager.loop(); + assert(physical_output.at(SERIAL_TX) == expected); + assert(tx_owner.at(SERIAL_TX) == int(companion_serial.port)); + } +#if ENV_INCLUDE_GPS == 1 + sensors.stop_gps(); + sensors.start_gps(); + sensors.initBasicGPS(); + assert(Serial1.state().baud == 9600); + assert(Serial1.state().rx == PIN_GPS_TX && Serial1.state().tx == PIN_GPS_RX); + if (!EXPECTED_ALIAS) assertCompanionConfiguration(); +#if defined(ENABLE_USB_INTERFACE) && !ARDUINO_USB_CDC_ON_BOOT + assert(Serial.state().baud == 115200); + assert(Serial.state().rx == SOC_RX0 && Serial.state().tx == SOC_TX0); +#endif +#endif + std::puts("PASS: actual GPIO-matrix Companion UART alias ownership"); +} +''' + + +class CompanionHardwareSerialGpsTest(unittest.TestCase): + def test_gpio_alias_has_exactly_one_command_and_reply_owner(self): + compiler = os.environ.get("CXX") or shutil.which("g++") or shutil.which("clang++") + if compiler is None: + self.skipTest("a host C++17 compiler is required") + main = (ROOT / "examples/companion_radio/main.cpp").read_text() + gps = (ROOT / "src/helpers/sensors/EnvironmentSensorManager.cpp").read_text() + native = (ROOT / "test/test_serial_mode_switch/test_serial_mode_switch.cpp").read_text() + source = HARNESS[:HARNESS.index("int main()")] + uart = extract_braced(source, "class HardwareSerial") + ";" + source = source.replace(uart, GPIO_UART) + source = source.replace("@DECLARATION@", GPIO_USB + "\n@DECLARATION@") + source += GPIO_MAIN + source = source.replace("@STREAM@", extract_braced(native, "class BufferStream") + ";") + source = source.replace("@DECLARATION@", conditional_block(main, "// include hardware serial interface")) + source = source.replace("@SETUP@", conditional_block(main, "// add hardware serial interface")) + usb_marker = "#if defined(ENABLE_USB_INTERFACE)\n#if COMPANION_FEATURE_USB_MOTA_SOURCE\n usb_serial_interface.begin" + source = source.replace("@USB_SETUP@", conditional_block(main, usb_marker)) + source = source.replace("@GPS_FUNCTIONS@", "\n".join( + extract_braced(gps, signature) for signature in ( + "void EnvironmentSensorManager::initBasicGPS()", + "void EnvironmentSensorManager::start_gps()", + "void EnvironmentSensorManager::stop_gps()"))) + scenarios = ( + ("m2-full", ["ENABLE_USB_INTERFACE=1", "ARDUINO_USB_CDC_ON_BOOT=0", + "ENV_INCLUDE_GPS=0", "SERIAL_RX=44", "SERIAL_TX=43", + "EXPECTED_ALIAS=1", "EXPECTED_COMPANION_UART=1"]), + ("m5-full", ["ENABLE_USB_INTERFACE=1", "ARDUINO_USB_CDC_ON_BOOT=0", + "ENV_INCLUDE_GPS=1", "SERIAL_RX=44", "SERIAL_TX=43", + "EXPECTED_ALIAS=1", "EXPECTED_COMPANION_UART=2"]), + ("symbolic-pins-full", ["ENABLE_USB_INTERFACE=1", "ARDUINO_USB_CDC_ON_BOOT=0", + "ENV_INCLUDE_GPS=1", "SERIAL_RX=SYMBOL_RX", "SERIAL_TX=SYMBOL_TX", + "EXPECTED_ALIAS=1", "EXPECTED_COMPANION_UART=2"]), + ("native-usb-full", ["ENABLE_USB_INTERFACE=1", "ARDUINO_USB_CDC_ON_BOOT=1", + "ENV_INCLUDE_GPS=1", "SERIAL_RX=44", "SERIAL_TX=43", + "EXPECTED_ALIAS=0", "EXPECTED_COMPANION_UART=2"]), + ("xiao-native-usb-full", ["ENABLE_USB_INTERFACE=1", "ARDUINO_USB_CDC_ON_BOOT=1", + "ARDUINO_USB_MODE=1", "ENV_INCLUDE_GPS=0", "SERIAL_RX=D7", "SERIAL_TX=D6", + "EXPECTED_ALIAS=0", "EXPECTED_COMPANION_UART=1"]), + ("distinct-pins-full", ["ENABLE_USB_INTERFACE=1", "ARDUINO_USB_CDC_ON_BOOT=0", + "ENV_INCLUDE_GPS=1", "SERIAL_RX=10", "SERIAL_TX=11", + "EXPECTED_ALIAS=0", "EXPECTED_COMPANION_UART=2"]), + ("m5-serial", ["ARDUINO_USB_CDC_ON_BOOT=0", "ENV_INCLUDE_GPS=1", + "SERIAL_RX=44", "SERIAL_TX=43", "EXPECTED_ALIAS=0", + "EXPECTED_COMPANION_UART=2"]), + ) + with tempfile.TemporaryDirectory(prefix="meshcore-companion-uart-alias-") as directory: + work = Path(directory) + cpp = work / "test.cpp" + cpp.write_text(source, encoding="ascii") + for name, defines in scenarios: + with self.subTest(profile=name): + binary = work / f"{name}.exe" + built = subprocess.run([ + compiler, "-std=c++17", "-Werror", *SANITIZERS, + "-DSOC_RX0=44", "-DSOC_TX0=43", "-DCONFIG_IDF_TARGET_ESP32S3=1", + "-DCOMPANION_FEATURE_USB_MOTA_SOURCE=0", + "-DESP32=1", "-DPIN_GPS_TX=19", "-DPIN_GPS_RX=20", "-DPIN_GPS_EN=11", + "-DPERSISTANT_GPS=1", "-DENV_SKIP_GPS_DETECT=1", + *[f"-D{define}" for define in defines], + f"-I{ROOT / 'test/mocks'}", f"-I{ROOT / 'src'}", str(cpp), + str(ROOT / "src/helpers/ArduinoSerialInterface.cpp"), "-o", str(binary), + ], capture_output=True, text=True, timeout=60) + self.assertEqual(built.returncode, 0, built.stdout + built.stderr) + checked = subprocess.run([str(binary)], capture_output=True, + text=True, timeout=10) + self.assertEqual(checked.returncode, 0, checked.stdout + checked.stderr) + self.assertIn("PASS: actual GPIO-matrix Companion UART alias ownership", checked.stdout) + + def test_actual_gps_startup_and_restart_preserve_companion(self): + compiler = os.environ.get("CXX") or shutil.which("g++") or shutil.which("clang++") + if compiler is None: + self.skipTest("a host C++17 compiler is required") + main = (ROOT / "examples/companion_radio/main.cpp").read_text() + gps = (ROOT / "src/helpers/sensors/EnvironmentSensorManager.cpp").read_text() + native = (ROOT / "test/test_serial_mode_switch/test_serial_mode_switch.cpp").read_text() + variant = (ROOT / "variants/thinknode_m5/variant.h").read_text() + config = configparser.ConfigParser(interpolation=None) + config.read(ROOT / "variants/thinknode_m5/platformio.ini") + self.assertEqual(config["ThinkNode_M5"]["extends"], "esp32_base") + board = json.loads((ROOT / "boards" / (config["ThinkNode_M5"]["board"] + ".json")).read_text()) + self.assertEqual(board["build"]["mcu"], "esp32s3") + base_flags = config["ThinkNode_M5"]["build_flags"] + for flag in ("ENV_INCLUDE_GPS=1", "PERSISTANT_GPS=1", "ENV_SKIP_GPS_DETECT=1"): + self.assertRegex(base_flags, r"-D\s+" + re.escape(flag) + r"\b") + target = (ROOT / "variants/thinknode_m5/target.cpp").read_text() + self.assertRegex(target, r"MicroNMEALocationProvider\(Serial1,\s*&rtc_clock\)") + definitions = [] + for name in ("PIN_GPS_RX", "PIN_GPS_TX", "PIN_GPS_EN"): + value = re.search(r"^#define\s+" + name + r"\s+\((\d+)\)", variant, re.MULTILINE) + self.assertIsNotNone(value, name) + definitions.append(f"{name}={value.group(1)}") + source = HARNESS.replace("@STREAM@", extract_braced(native, "class BufferStream") + ";") + source = source.replace("@DECLARATION@", conditional_block(main, "// include hardware serial interface")) + source = source.replace("@SETUP@", conditional_block(main, "// add hardware serial interface")) + source = source.replace("@GPS_FUNCTIONS@", "\n".join( + extract_braced(gps, signature) for signature in ( + "void EnvironmentSensorManager::initBasicGPS()", + "void EnvironmentSensorManager::start_gps()", + "void EnvironmentSensorManager::stop_gps()"))) + scenarios = [] + for profile in ("serial", "full"): + flags = config[f"env:ThinkNode_M5_companion_radio_{profile}"]["build_flags"] + rx = re.findall(r"-D\s+SERIAL_RX=(\d+)\b", flags) + tx = re.findall(r"-D\s+SERIAL_TX=(\d+)\b", flags) + self.assertEqual(rx, ["44"]) + self.assertEqual(tx, ["43"]) + scenarios.append((f"m5-{profile}", ["CONFIG_IDF_TARGET_ESP32S3=1", "ENV_INCLUDE_GPS=1", + "EXPECTED_COMPANION_UART=2", f"SERIAL_RX={rx[0]}", f"SERIAL_TX={tx[0]}"])) + scenarios.extend(( + ("s3-no-gps", ["CONFIG_IDF_TARGET_ESP32S3=1", "ENV_INCLUDE_GPS=0", + "EXPECTED_COMPANION_UART=1", "SERIAL_RX=44", "SERIAL_TX=43"]), + ("other-platform", ["ENV_INCLUDE_GPS=1", "EXPECTED_COMPANION_UART=1", + "SERIAL_RX=44", "SERIAL_TX=43"]), + ("gps-without-companion-uart", ["CONFIG_IDF_TARGET_ESP32S3=1", "ENV_INCLUDE_GPS=1", + "EXPECTED_COMPANION_UART=0"]), + )) + with tempfile.TemporaryDirectory(prefix="meshcore-companion-gps-uart-") as directory: + work = Path(directory) + cpp = work / "test.cpp" + cpp.write_text(source, encoding="ascii") + for name, defines in scenarios: + with self.subTest(profile=name): + binary = work / f"{name}.exe" + built = subprocess.run([ + compiler, "-std=c++17", "-Werror", *SANITIZERS, + "-DPERSISTANT_GPS=1", "-DENV_SKIP_GPS_DETECT=1", + *[f"-D{define}" for define in definitions + defines], + f"-I{ROOT / 'test/mocks'}", f"-I{ROOT / 'src'}", str(cpp), + str(ROOT / "src/helpers/ArduinoSerialInterface.cpp"), "-o", str(binary), + ], capture_output=True, text=True, timeout=60) + self.assertEqual(built.returncode, 0, built.stdout + built.stderr) + checked = subprocess.run([str(binary)], capture_output=True, + text=True, timeout=10) + self.assertEqual(checked.returncode, 0, checked.stdout + checked.stderr) + self.assertIn("PASS: actual Companion/GPS UART ownership", checked.stdout) + + +if __name__ == "__main__": + unittest.main()