Use robust serial fallback for GPS AT commands and expose diagnostic sensors (#78)

Author: FaserF

custom_components/openwrt/api/base.py

@@ -468,6 +468,9 @@ class QModemInfo:
     nr5g_rsrp: int | None = None
     nr5g_rsrq: int | None = None
     nr5g_sinr: int | None = None
+    gps_latitude: float | None = None
+    gps_longitude: float | None = None
+    gps_last_update: str | None = None
 
 
 @dataclass

custom_components/openwrt/coordinator.py

@@ -455,6 +455,7 @@ async def _async_update_data(self) -> OpenWrtData:
             except Exception as err:
                 _LOGGER.debug("nlbwmon top hosts fetch failed: %s", err)
         if self.config_entry.options.get(CONF_GPS_MODEM_ENABLED, False):
+            data.qmodem_info.enabled = True
             gps_port = self.config_entry.options.get(
                 CONF_GPS_MODEM_PORT, DEFAULT_GPS_MODEM_PORT
             )
@@ -467,10 +468,28 @@ async def _async_update_data(self) -> OpenWrtData:
                 from .helpers.gps import async_update_gps_location
 
                 try:
-                    await async_update_gps_location(self.hass, self.client, gps_port)
+                    res = await async_update_gps_location(
+                        self.hass, self.client, gps_port
+                    )
+                    if res:
+                        lat, lon, last_update = res
+                        data.qmodem_info.gps_latitude = lat
+                        data.qmodem_info.gps_longitude = lon
+                        data.qmodem_info.gps_last_update = last_update
                 except Exception as gps_err:
                     _LOGGER.debug("GPS location update failed: %s", gps_err)
 
+            # Preserve previous GPS data if we are not currently polling it
+            if self.data and self.data.qmodem_info:
+                if data.qmodem_info.gps_latitude is None:
+                    data.qmodem_info.gps_latitude = self.data.qmodem_info.gps_latitude
+                if data.qmodem_info.gps_longitude is None:
+                    data.qmodem_info.gps_longitude = self.data.qmodem_info.gps_longitude
+                if data.qmodem_info.gps_last_update is None:
+                    data.qmodem_info.gps_last_update = (
+                        self.data.qmodem_info.gps_last_update
+                    )
+
         return data
 
     async def _async_fetch_mqtt_presence_data(self, data: OpenWrtData) -> None:

custom_components/openwrt/helpers/gps.py

@@ -6,10 +6,33 @@
 from typing import Any
 
 from homeassistant.core import HomeAssistant
+from homeassistant.util import dt as dt_util
 
 _LOGGER = logging.getLogger(__name__)
 
 
+async def async_execute_at_command(
+    client: Any, port: str, at_command: str, timeout: int = 2
+) -> str:
+    """Send an AT command to the serial port using microcom or a universal fallback."""
+    # Construct a shell command that tries:
+    # 1. microcom
+    # 2. stty + echo + timeout/cat
+    cmd = (
+        f"if command -v microcom >/dev/null 2>&1; then "
+        f'echo -e "{at_command}\\r" | microcom -t {timeout}000 {port} 2>/dev/null; '
+        f"elif command -v stty >/dev/null 2>&1; then "
+        f"stty -F {port} 9600 -echo igncr icanon onlcr 2>/dev/null; "
+        f'(sleep 0.2; echo -e "{at_command}\\r" > {port}) & '
+        f"timeout {timeout} cat {port} 2>/dev/null; "
+        f"else "
+        f'(sleep 0.2; echo -e "{at_command}\\r" > {port}) & '
+        f"timeout {timeout} cat {port} 2>/dev/null; "
+        f"fi"
+    )
+    return await client.execute_command(cmd)
+
+
 def parse_nmea_coordinate(coord_str: str) -> float | None:
     """Parse NMEA coordinate string (ddmm.mmmmN/S or dddmm.mmmmE/W) into decimal degrees."""
     if not coord_str or len(coord_str) < 3:
@@ -39,7 +62,7 @@ def parse_qgpsloc_response(response: str) -> tuple[float, float] | None:
     for line in response.splitlines():
         line = line.strip()
         if "+QGPSLOC:" in line:
-            # Strip anything before "+QGPSLOC: " to handle echoes or prefixes
+            # Strip anything before "+QGPSLOC:" to handle echoes or prefixes
             start = line.find("+QGPSLOC:")
             content = line[start + len("+QGPSLOC:") :].strip()
             parts = content.split(",")
@@ -57,22 +80,19 @@ async def async_update_gps_location(
     hass: HomeAssistant,
     client: Any,
     port: str,
-) -> bool:
-    """Query GPS coordinates from modem and update HA location."""
+) -> tuple[float, float, str] | None:
+    """Query GPS coordinates from modem, update HA location and return coordinates."""
     try:
         # Check if GPS is enabled
-        check_cmd = f'echo -e "AT+QGPS?\\r" | microcom -t 1000 {port}'
-        check_res = await client.execute_command(check_cmd)
+        check_res = await async_execute_at_command(client, port, "AT+QGPS?", timeout=1)
 
         # If QGPS is disabled or query returned error, try enabling it
         if "+QGPS: 0" in check_res or "ERROR" in check_res:
             _LOGGER.debug("GPS is disabled, enabling via AT+QGPS=1")
-            enable_cmd = f'echo -e "AT+QGPS=1\\r" | microcom -t 1000 {port}'
-            await client.execute_command(enable_cmd)
+            await async_execute_at_command(client, port, "AT+QGPS=1", timeout=1)
 
         # Poll the GPS location
-        loc_cmd = f'echo -e "AT+QGPSLOC?\\r" | microcom -t 2000 {port}'
-        loc_res = await client.execute_command(loc_cmd)
+        loc_res = await async_execute_at_command(client, port, "AT+QGPSLOC?", timeout=2)
 
         coords = parse_qgpsloc_response(loc_res)
         if coords:
@@ -86,10 +106,11 @@ async def async_update_gps_location(
                     "longitude": lon,
                 },
             )
-            return True
+            last_update = dt_util.now().isoformat()
+            return lat, lon, last_update
         _LOGGER.debug(
             "Failed to get valid GPS fix or coordinates from %s: %s", port, loc_res
         )
     except Exception as err:
         _LOGGER.warning("Failed to update GPS location from Quectel modem: %s", err)
-    return False
+    return None

custom_components/openwrt/sensor.py

@@ -840,6 +840,27 @@ def _get_qmodem_sensors() -> tuple[OpenWrtSensorDescription, ...]:
             entity_category=EntityCategory.DIAGNOSTIC,
             value_fn=lambda data: data.qmodem_info.nr5g_sinr,
         ),
+        OpenWrtSensorDescription(
+            key="qmodem_gps_latitude",
+            name="Modem GPS Latitude",
+            translation_key="qmodem_gps_latitude",
+            entity_category=EntityCategory.DIAGNOSTIC,
+            value_fn=lambda data: data.qmodem_info.gps_latitude,
+        ),
+        OpenWrtSensorDescription(
+            key="qmodem_gps_longitude",
+            name="Modem GPS Longitude",
+            translation_key="qmodem_gps_longitude",
+            entity_category=EntityCategory.DIAGNOSTIC,
+            value_fn=lambda data: data.qmodem_info.gps_longitude,
+        ),
+        OpenWrtSensorDescription(
+            key="qmodem_gps_last_update",
+            name="Modem GPS Last Update",
+            translation_key="qmodem_gps_last_update",
+            entity_category=EntityCategory.DIAGNOSTIC,
+            value_fn=lambda data: data.qmodem_info.gps_last_update,
+        ),
     )
 
 

custom_components/openwrt/strings.json

@@ -593,6 +593,15 @@
       "qmodem_nr5g_sinr": {
         "name": "5G NR SINR"
       },
+      "qmodem_gps_latitude": {
+        "name": "Modem GPS Latitude"
+      },
+      "qmodem_gps_longitude": {
+        "name": "Modem GPS Longitude"
+      },
+      "qmodem_gps_last_update": {
+        "name": "Modem GPS Last Update"
+      },
       "dhcp_lease_count": {
         "name": "DHCP Leases"
       },

custom_components/openwrt/translations/de.json

@@ -593,6 +593,15 @@
       "qmodem_nr5g_sinr": {
         "name": "5G NR SINR"
       },
+      "qmodem_gps_latitude": {
+        "name": "Modem GPS Breitengrad"
+      },
+      "qmodem_gps_longitude": {
+        "name": "Modem GPS Längengrad"
+      },
+      "qmodem_gps_last_update": {
+        "name": "Modem GPS Letzte Aktualisierung"
+      },
       "dhcp_lease_count": {
         "name": "DHCP Leases"
       },

custom_components/openwrt/translations/en.json

@@ -593,6 +593,15 @@
       "qmodem_nr5g_sinr": {
         "name": "5G NR SINR"
       },
+      "qmodem_gps_latitude": {
+        "name": "Modem GPS Latitude"
+      },
+      "qmodem_gps_longitude": {
+        "name": "Modem GPS Longitude"
+      },
+      "qmodem_gps_last_update": {
+        "name": "Modem GPS Last Update"
+      },
       "dhcp_lease_count": {
         "name": "DHCP Leases"
       },

tests/test_gps_location.py

@@ -72,7 +72,9 @@ async def test_async_update_gps_location(hass):
         hass.services, "async_call", new_callable=AsyncMock
     ) as mock_service_call:
         res = await async_update_gps_location(hass, mock_client, "/dev/ttyUSB3")
-        assert res is True
+        assert res is not None
+        assert res[0] == pytest.approx(52.01681667)
+        assert res[1] == pytest.approx(-0.71821833)
         mock_service_call.assert_called_once_with(
             "homeassistant",
             "set_location",
@@ -94,6 +96,8 @@ async def test_async_update_gps_location(hass):
         hass.services, "async_call", new_callable=AsyncMock
     ) as mock_service_call:
         res = await async_update_gps_location(hass, mock_client, "/dev/ttyUSB3")
-        assert res is True
+        assert res is not None
+        assert res[0] == pytest.approx(52.01681667)
+        assert res[1] == pytest.approx(-0.71821833)
         assert mock_client.execute_command.call_count == 3
         mock_service_call.assert_called_once()