Fix error when stop detail returned code 81 but the stop can be acces…

…ible with an alternative endpoint

emt_madrid/emt_api.py

@@ -181,6 +181,27 @@ def token(self) -> str:
         """Return API token."""
         return self._token
 
+    async def _update_stop_info_around_stop(self) -> Optional[Dict[str, Any]]:
+        """Update information about a bus stop using the stops around stop endpoint."""
+        endpoint = f"v2/transport/busemtmad/stops/arroundstop/{self._stop_id}/0/"
+        headers = {"accessToken": self._token}
+        url = self._base_url + endpoint
+
+        try:
+            async with async_timeout.timeout(DEFAULT_TIMEOUT):
+                response = await self._get_data(url, headers, "GET")
+
+            if response is None:
+                return None
+
+            parsed_stop_info = parse_stop_info(response, self._stop_info)
+
+            return parsed_stop_info
+
+        except asyncio.TimeoutError:
+            _LOGGER.warning("Timeout error fetching data from %s", url)
+            return None
+
     async def update_stop_info(self) -> None:
         """Update information about a bus stop."""
         endpoint = f"v1/transport/busemtmad/stops/{self._stop_id}/detail/"
@@ -193,13 +214,17 @@ async def update_stop_info(self) -> None:
 
             if response is None:
                 return None
+            print(response)
 
-            parsed_stop_info = parse_stop_info(response, self._stop_info)
+            if response.get("code", {}) == "81":
+                parsed_stop_info = await self._update_stop_info_around_stop()
+            else:
+                parsed_stop_info = parse_stop_info(response, self._stop_info)
 
             if parsed_stop_info is None:
                 return None
 
-            if parsed_stop_info.get("error") == "Invalid token":
+            if parsed_stop_info.get("error", None) is not None:
                 return None
 
             async with self._update_semaphore:

emt_madrid/parser.py

@@ -42,56 +42,83 @@ def parse_stop_info(
     assert stop_info is not None
     try:
         response_code = response.get("code")
-        if response_code in ("90", "81"):
+        if response_code == "90":
             raise BusStopDisabled
         if response_code == "80":
             raise InvalidToken
         if response_code == "98":
             raise APILimitReached
 
-        response_stop = response["data"][0]["stops"][0]
-        stop_info.update(
-            {
-                "stop_id": response_stop["stop"],
-                "stop_name": response_stop["name"],
-                "stop_coordinates": response_stop["geometry"]["coordinates"],
-                "stop_address": response_stop["postalAddress"],
-                "lines": parse_lines(response_stop["dataLine"]),
-            }
-        )
+        if "stopName" in response["data"][0]:
+            response_stop = response["data"][0]
+            stop_info.update(
+                {
+                    "stop_id": str(response_stop["stopId"]),
+                    "stop_name": response_stop["stopName"].rstrip(),
+                    "stop_coordinates": response_stop["geometry"]["coordinates"],
+                    "stop_address": response_stop["address"].rstrip(),
+                    "lines": parse_lines(response_stop["lines"], "basic"),
+                }
+            )
+
+        else:
+            response_stop = response["data"][0]["stops"][0]
+            stop_info.update(
+                {
+                    "stop_id": response_stop["stop"],
+                    "stop_name": response_stop["name"].rstrip(),
+                    "stop_coordinates": response_stop["geometry"]["coordinates"],
+                    "stop_address": response_stop["postalAddress"].rstrip(),
+                    "lines": parse_lines(response_stop["dataLine"], "full"),
+                }
+            )
         return stop_info
 
     except BusStopDisabled:
         _LOGGER.warning("Bus Stop disabled or does not exist")
-        return None
+        return {"error": "Bus Stop disabled", "error_code": response.get("code")}
     except InvalidToken:
         _LOGGER.warning("Invalid or expired token")
-        return {"error": "Invalid token"}
+        return {"error": "Invalid token", "error_code": response.get("code")}
     except APILimitReached:
         _LOGGER.warning("Maximum daily API usage has been exceeded.")
-        return None
+        return {
+            "error": "Maximum daily API usage reached",
+            "error_code": response.get("code"),
+        }
 
 
-def parse_lines(lines: List[Dict[str, Any]]) -> Dict[str, Any]:
+def parse_lines(lines: List[Dict[str, Any]], mode: str) -> Dict[str, Any]:
     """Parse the line info from the API response."""
     line_info: Dict[str, Any] = {}
-    for line in lines:
-        line_number = str(line.get("label"))
-        line_info[line_number] = {
-            "destination": line.get("headerA")
-            if line.get("direction") == "A"
-            else line.get("headerB"),
-            "origin": line.get("headerA")
-            if line.get("direction") == "B"
-            else line.get("headerB"),
-            "max_freq": int(line.get("maxFreq") or 0),
-            "min_freq": int(line.get("minFreq") or 0),
-            "start_time": line.get("startTime"),
-            "end_time": line.get("stopTime"),
-            "day_type": line.get("dayType"),
-            "distance": [],
-            "arrivals": [],
-        }
+    if mode == "full":
+        for line in lines:
+            line_number = str(line.get("label"))
+            line_info[line_number] = {
+                "destination": line.get("headerA")
+                if line.get("direction") == "A"
+                else line.get("headerB"),
+                "origin": line.get("headerA")
+                if line.get("direction") == "B"
+                else line.get("headerB"),
+                "max_freq": int(line.get("maxFreq") or 0),
+                "min_freq": int(line.get("minFreq") or 0),
+                "start_time": line.get("startTime"),
+                "end_time": line.get("stopTime"),
+                "day_type": line.get("dayType"),
+                "distance": [],
+                "arrivals": [],
+            }
+    elif mode == "basic":
+        line_info = {}
+        for line in lines:
+            line_number = line["label"]
+            line_info[line_number] = {
+                "destination": line["nameA"] if line["to"] == "A" else line["nameB"],
+                "origin": line["nameA"] if line["to"] == "B" else line["nameB"],
+                "distance": [],
+                "arrivals": [],
+            }
     return line_info
 
 

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "emt_madrid"
-version = "0.0.1"
+version = "0.0.2"
 description = "Python wrapper for the Madrid EMT (Empresa Municipal de Transportes) API"
 readme = "README.md"
 requires-python = ">=3.7"

tests/conftest.py

@@ -5,6 +5,7 @@
 from typing import Any, Dict, Optional
 
 TEST_EXAMPLES_PATH = pathlib.Path(__file__).parent / "response_examples"
+_FIXTURE_STOP_INFO = "STOP_INFO.json"
 
 _FIXTURE_LOGIN_OK = "LOGIN_OK.json"
 _FIXTURE_LOGIN_INVALID_USER = "LOGIN_INVALID_USER.json"
@@ -14,9 +15,10 @@
 _FIXTURE_STOP_DETAIL_INVALID_TOKEN = "STOP_DETAIL_INVALID_TOKEN.json"
 _FIXTURE_STOP_ARRIVAL_OK = "STOP_ARRIVAL_OK.json"
 _FIXTURE_STOP_ARRIVAL_INVALID_STOP = "STOP_ARRIVAL_INVALID_STOP.json"
+_FIXTURE_STOP_DETAIL_NOT_FOUND = "STOP_DETAIL_NOT_FOUND.json"
 _FIXTURE_STOP_ARRIVAL_INVALID_TOKEN = "STOP_ARRIVAL_INVALID_TOKEN.json"
 _FIXTURE_API_LIMIT = "API_LIMIT.json"
-_FIXTURE_STOP_INFO = "STOP_INFO.json"
+_FIXTURE_STOPS_AROUND_STOP_OK = "STOPS_AROUND_STOP_OK.json"
 
 
 class MockAsyncSession:
@@ -51,33 +53,49 @@ async def json(self, *_args, **_kwargs):
     async def get(self, url: str, headers: Dict[str, Any], *_args, **_kwargs):
         """Dumb get method."""
         self._counter += 1
-        if self.exc:
+        if self.exc and not ("detail" in url and "stop_not_found" in url):
             raise self.exc
+
         if "user/login/" in url:
-            if headers["email"] == "invalid_email":
-                self._raw_response = load_fixture(_FIXTURE_LOGIN_INVALID_USER)
-            elif headers["password"] == "invalid_password":
-                self._raw_response = load_fixture(_FIXTURE_LOGIN_INVALID_PASSWORD)
-            elif headers["password"] == "api_limit":
-                self._raw_response = load_fixture(_FIXTURE_API_LIMIT)
-            else:
-                self._raw_response = load_fixture(_FIXTURE_LOGIN_OK)
-
-        elif "stops" in url and "detail" in url:
-            stop_id = url.split("/")[-3]
-            if (
-                headers["accessToken"] == "invalid_token"
-                or headers["accessToken"] is None
-            ):
-                self._raw_response = load_fixture(_FIXTURE_STOP_DETAIL_INVALID_TOKEN)
-            elif headers["accessToken"] == "api_limit":
-                self._raw_response = load_fixture(_FIXTURE_API_LIMIT)
-            elif stop_id in ("invalid_stop_id", "None"):
-                self._raw_response = load_fixture(_FIXTURE_STOP_DETAIL_INVALID_STOP)
-            else:
-                self._raw_response = load_fixture(_FIXTURE_STOP_DETAIL_OK)
+            self._raw_response = self._handle_login_request(headers)
+        elif "detail" in url:
+            self._raw_response = self._handle_detail_request(url, headers)
+        elif "arroundstop" in url:
+            self._raw_response = self._handle_arroundstop_request(headers)
+
         return self
 
+    def _handle_login_request(self, headers: Dict[str, Any]):
+        """Handle the login request based on the provided headers."""
+        if headers["email"] == "invalid_email":
+            return load_fixture(_FIXTURE_LOGIN_INVALID_USER)
+        if headers["password"] == "invalid_password":
+            return load_fixture(_FIXTURE_LOGIN_INVALID_PASSWORD)
+        if headers["password"] == "api_limit":
+            return load_fixture(_FIXTURE_API_LIMIT)
+
+        return load_fixture(_FIXTURE_LOGIN_OK)
+
+    def _handle_detail_request(self, url: str, headers: Dict[str, Any]):
+        """Handle the detail request based on the provided URL and headers."""
+        stop_id = url.split("/")[-3]
+        if headers["accessToken"] == "invalid_token" or headers["accessToken"] is None:
+            return load_fixture(_FIXTURE_STOP_DETAIL_INVALID_TOKEN)
+        if headers["accessToken"] == "api_limit" and stop_id != "stop_not_found":
+            return load_fixture(_FIXTURE_API_LIMIT)
+        if stop_id in ("invalid_stop_id", "None"):
+            return load_fixture(_FIXTURE_STOP_DETAIL_INVALID_STOP)
+        if stop_id in ("stop_not_found"):
+            return load_fixture(_FIXTURE_STOP_DETAIL_NOT_FOUND)
+
+        return load_fixture(_FIXTURE_STOP_DETAIL_OK)
+
+    def _handle_arroundstop_request(self, headers: Dict[str, Any]):
+        """Handle the 'arroundstop' request."""
+        if headers["accessToken"] == "api_limit":
+            return load_fixture(_FIXTURE_API_LIMIT)
+        return load_fixture(_FIXTURE_STOPS_AROUND_STOP_OK)
+
     async def post(self, url: str, headers: Dict[str, Any], *_args, **_kwargs):
         """Dumb post method."""
         self._counter += 1
@@ -106,17 +124,32 @@ def load_fixture(filename: str):
     return json.loads((TEST_EXAMPLES_PATH / filename).read_text())
 
 
-def check_stop_info(stop_info, distance=0, arrivals=0):
+def check_stop_info(stop_info, distance=0, arrivals=0, code="00"):
     """Verify that the stop_info is correct."""
-    lines = stop_info.get("lines")
-    assert len(lines) == 12
-    line = lines.get("27")
-    assert line.get("destination") == "PLAZA CASTILLA"
-    assert line.get("origin") == "EMBAJADORES"
-    assert line.get("max_freq") == 11
-    assert line.get("min_freq") == 3
-    assert line.get("start_time") == "05:35"
-    assert line.get("end_time") == "00:01"
-    assert line.get("day_type") == "LA"
-    assert len(line.get("distance")) == distance
-    assert len(line.get("arrivals")) == arrivals
+    if code == "00":
+        lines = stop_info.get("lines")
+        assert len(lines) == 12
+        line = lines.get("27")
+        assert line.get("destination") == "PLAZA CASTILLA"
+        assert line.get("origin") == "EMBAJADORES"
+        assert line.get("max_freq") == 11
+        assert line.get("min_freq") == 3
+        assert line.get("start_time") == "05:35"
+        assert line.get("end_time") == "00:01"
+        assert line.get("day_type") == "LA"
+        assert len(line.get("distance")) == distance
+        assert len(line.get("arrivals")) == arrivals
+
+    if code == "81":
+        lines = stop_info.get("lines")
+        assert len(lines) == 2
+        line = lines.get("172")
+        assert line.get("destination") == "LAS TABLAS"
+        assert line.get("origin") == "TELEFONICA"
+        assert line.get("max_freq") is None
+        assert line.get("min_freq") is None
+        assert line.get("start_time") is None
+        assert line.get("end_time") is None
+        assert line.get("day_type") is None
+        assert len(line.get("distance")) == distance
+        assert len(line.get("arrivals")) == arrivals

tests/response_examples/STOPS_AROUND_STOP_OK.json

@@ -0,0 +1,34 @@
+{
+  "code": "00",
+  "description": "Data recovered OK (lapsed: 82 millsecs)",
+  "data": [
+    {
+      "stopId": 4490,
+      "geometry": {
+        "type": "Point",
+        "coordinates": [-3.67357040366554, 40.4977004070578]
+      },
+      "stopName": "Camino de Santiago-Valcarlos",
+      "address": "Valcarlos con Av. Camino de Santiago                   ",
+      "metersToPoint": 0,
+      "lines": [
+        {
+          "line": "172",
+          "label": "172",
+          "nameA": "TELEFONICA",
+          "nameB": "LAS TABLAS",
+          "metersFromHeader": 8082,
+          "to": "B"
+        },
+        {
+          "line": "372",
+          "label": "172SF",
+          "nameA": "TELEFONICA",
+          "nameB": "LAS TABLAS",
+          "metersFromHeader": 8082,
+          "to": "B"
+        }
+      ]
+    }
+  ]
+}

tests/response_examples/STOP_DETAIL_NOT_FOUND.json

@@ -0,0 +1,6 @@
+{
+  "code": "81",
+  "description": "No records found or error, (lapsed: 444 millsecs)",
+  "datetime": "2023-07-18T16:49:03.435480",
+  "data": [{}]
+}

tests/test_emt_api.py

@@ -61,6 +61,29 @@ async def test_update_stop_info(
         assert mock_session.call_count == call_count
 
 
+@pytest.mark.parametrize(
+    "token, stop_id, status, exception, num_log_msgs, call_count",
+    (
+        ("token", "stop_not_found", 200, None, 0, 2),
+        ("api_limit", "stop_not_found", 200, None, 1, 2),
+        ("token", "stop_not_found", 200, asyncio.TimeoutError, 1, 2),
+        ("token", "stop_not_found", 200, TimeoutError, 1, 2),
+        ("token", "stop_not_found", 200, ClientError, 1, 2),
+    ),
+)
+@pytest.mark.asyncio
+async def test_update_stop_info_alternative_method(
+    token, stop_id, status, exception, num_log_msgs, call_count, caplog
+):  # pylint: disable=too-many-arguments
+    """Test update_stop_info method."""
+    mock_session = MockAsyncSession(status=status, exc=exception)
+    with caplog.at_level(logging.WARNING):
+        emt_api = EMTAPIBusStop(session=mock_session, token=token, stop_id=stop_id)
+        await emt_api.update_stop_info()
+        assert len(caplog.messages) == num_log_msgs
+        assert mock_session.call_count == call_count
+
+
 @pytest.mark.parametrize(
     "token, stop_info, status, exception, num_log_msgs, call_count",
     (