Add Location

This will be used to retrieve the microgrid information.

Signed-off-by: Leandro Lucarella <[email protected]>

Author: llucax

src/frequenz/client/microgrid/__init__.py

@@ -32,6 +32,7 @@
     UnrecognizedGrpcStatus,
 )
 from ._lifetime import Lifetime
+from ._location import Location
 
 __all__ = [
     "ApiClientError",
@@ -45,6 +46,7 @@
     "InternalError",
     "InvalidArgument",
     "Lifetime",
+    "Location",
     "MicrogridApiClient",
     "OperationAborted",
     "OperationCancelled",

src/frequenz/client/microgrid/_location.py

@@ -0,0 +1,50 @@
+# License: MIT
+# Copyright © 2025 Frequenz Energy-as-a-Service GmbH
+
+"""Location information for a microgrid."""
+
+
+import logging
+from dataclasses import dataclass
+from functools import cached_property
+from zoneinfo import ZoneInfo
+
+import timezonefinder
+
+_timezone_finder = timezonefinder.TimezoneFinder()
+_logger = logging.getLogger(__name__)
+
+
+@dataclass(frozen=True, kw_only=True)
+class Location:
+    """A location of a microgrid."""
+
+    latitude: float | None
+    """The latitude of the microgrid in degree."""
+
+    longitude: float | None
+    """The longitude of the microgrid in degree."""
+
+    country_code: str | None
+    """The country code of the microgrid in ISO 3166-1 Alpha 2 format."""
+
+    @cached_property
+    def timezone(self) -> ZoneInfo | None:
+        """The timezone of the microgrid, or `None` if it could not be determined."""
+        if self.latitude is None or self.longitude is None:
+            _logger.warning(
+                "Latitude (%s) or longitude (%s) missing, cannot determine timezone"
+            )
+            return None
+        timezone = _timezone_finder.timezone_at(lat=self.latitude, lng=self.longitude)
+        return ZoneInfo(key=timezone) if timezone else None
+
+    def __str__(self) -> str:
+        """Return the short string representation of this instance."""
+        country = self.country_code or "<NO COUNTRY CODE>"
+        lat = f"{self.latitude:.2f}" if self.latitude is not None else "?"
+        lon = f"{self.longitude:.2f}" if self.longitude is not None else "?"
+        coordinates = ""
+        if self.latitude is not None or self.longitude is not None:
+            coordinates = f":({lat}, {lon})"
+        return f"{country}{coordinates}"

src/frequenz/client/microgrid/_location_proto.py

@@ -0,0 +1,47 @@
+# License: MIT
+# Copyright © 2025 Frequenz Energy-as-a-Service GmbH
+
+"""Loading of Location objects from protobuf messages."""
+
+import logging
+
+from frequenz.api.common.v1 import location_pb2
+
+from ._location import Location
+
+_logger = logging.getLogger(__name__)
+
+
+def location_from_proto(message: location_pb2.Location) -> Location:
+    """Convert a protobuf location message to a location object.
+
+    Args:
+        message: The protobuf message to convert.
+
+    Returns:
+        The resulting location object.
+    """
+    issues: list[str] = []
+
+    latitude: float | None = message.latitude if -90 <= message.latitude <= 90 else None
+    if latitude is None:
+        issues.append("latitude out of range [-90, 90]")
+
+    longitude: float | None = (
+        message.longitude if -180 <= message.longitude <= 180 else None
+    )
+    if longitude is None:
+        issues.append("longitude out of range [-180, 180]")
+
+    country_code = message.country_code or None
+    if country_code is None:
+        issues.append("country code is empty")
+
+    if issues:
+        _logger.warning(
+            "Found issues in location: %s | Protobuf message:\n%s",
+            ", ".join(issues),
+            message,
+        )
+
+    return Location(latitude=latitude, longitude=longitude, country_code=country_code)

tests/test_location.py

@@ -0,0 +1,175 @@
+# License: MIT
+# Copyright © 2025 Frequenz Energy-as-a-Service GmbH
+
+"""Tests for the microgrid metadata types."""
+
+from collections.abc import Iterator
+from dataclasses import dataclass
+from unittest.mock import MagicMock, patch
+from zoneinfo import ZoneInfo
+
+import pytest
+from frequenz.api.common.v1 import location_pb2
+
+from frequenz.client.microgrid import Location
+from frequenz.client.microgrid._location_proto import location_from_proto
+
+
+@dataclass(frozen=True, kw_only=True)
+class _ProtoConversionTestCase:  # pylint: disable=too-many-instance-attributes
+    """Test case for protobuf conversion."""
+
+    name: str
+    """The description of the test case."""
+
+    latitude: float
+    """The latitude to set in the protobuf message."""
+
+    longitude: float
+    """The longitude to set in the protobuf message."""
+
+    country_code: str
+    """The country code to set in the protobuf message."""
+
+    expected_none_latitude: bool = False
+    """The latitude is expected to be None."""
+
+    expected_none_longitude: bool = False
+    """The longitude is expected to be None."""
+
+    expected_none_country_code: bool = False
+    """The country code is expected to be None."""
+
+    expect_warning: bool = False
+    """Whether to expect a warning during conversion."""
+
+
+@pytest.fixture
+def timezone_finder() -> Iterator[MagicMock]:
+    """Return a mock timezone finder."""
+    with patch(
+        "frequenz.client.microgrid._location._timezone_finder", autospec=True
+    ) as mock_timezone_finder:
+        yield mock_timezone_finder
+
+
+def test_timezone_not_looked_up_if_unused(timezone_finder: MagicMock) -> None:
+    """Test the location timezone is not looked up if it is not used."""
+    location = Location(latitude=52.52, longitude=13.405, country_code="DE")
+
+    assert location.latitude == 52.52
+    assert location.longitude == 13.405
+    assert location.country_code == "DE"
+    timezone_finder.timezone_at.assert_not_called()
+
+
+def test_timezone_looked_up_but_not_found(timezone_finder: MagicMock) -> None:
+    """Test the location timezone is not looked up if it is not used."""
+    timezone_finder.timezone_at.return_value = None
+
+    location = Location(latitude=52.52, longitude=13.405, country_code="DE")
+
+    assert location.timezone is None
+    timezone_finder.timezone_at.assert_called_once_with(lat=52.52, lng=13.405)
+
+
+def test_timezone_looked_up_and_found(timezone_finder: MagicMock) -> None:
+    """Test the location timezone is not looked up if it is not used."""
+    timezone_finder.timezone_at.return_value = "Europe/Berlin"
+
+    location = Location(latitude=52.52, longitude=13.405, country_code="DE")
+
+    assert location.timezone == ZoneInfo(key="Europe/Berlin")
+    timezone_finder.timezone_at.assert_called_once_with(lat=52.52, lng=13.405)
+
+
+@pytest.mark.parametrize(
+    "case",
+    [
+        _ProtoConversionTestCase(
+            name="valid",
+            latitude=52.52,
+            longitude=13.405,
+            country_code="DE",
+        ),
+        _ProtoConversionTestCase(
+            name="boundary_latitude",
+            latitude=90.0,
+            longitude=13.405,
+            country_code="DE",
+        ),
+        _ProtoConversionTestCase(
+            name="boundary_longitude",
+            latitude=52.52,
+            longitude=180.0,
+            country_code="DE",
+        ),
+        _ProtoConversionTestCase(
+            name="invalid_latitude",
+            latitude=91.0,
+            longitude=13.405,
+            country_code="DE",
+            expected_none_latitude=True,
+            expect_warning=True,
+        ),
+        _ProtoConversionTestCase(
+            name="invalid_longitude",
+            latitude=52.52,
+            longitude=181.0,
+            country_code="DE",
+            expected_none_longitude=True,
+            expect_warning=True,
+        ),
+        _ProtoConversionTestCase(
+            name="empty_country_code",
+            latitude=52.52,
+            longitude=13.405,
+            country_code="",
+            expected_none_country_code=True,
+            expect_warning=True,
+        ),
+        _ProtoConversionTestCase(
+            name="all_invalid",
+            latitude=-91.0,
+            longitude=181.0,
+            country_code="",
+            expected_none_latitude=True,
+            expected_none_longitude=True,
+            expected_none_country_code=True,
+            expect_warning=True,
+        ),
+    ],
+    ids=lambda case: case.name,
+)
+def test_from_proto(
+    caplog: pytest.LogCaptureFixture, case: _ProtoConversionTestCase
+) -> None:
+    """Test conversion from protobuf message to Location."""
+    proto = location_pb2.Location(
+        latitude=case.latitude,
+        longitude=case.longitude,
+        country_code=case.country_code,
+    )
+    with caplog.at_level("WARNING"):
+        location = location_from_proto(proto)
+
+    if case.expected_none_latitude:
+        assert location.latitude is None
+    else:
+        assert location.latitude == pytest.approx(case.latitude)
+
+    if case.expected_none_longitude:
+        assert location.longitude is None
+    else:
+        assert location.longitude == pytest.approx(case.longitude)
+
+    if case.expected_none_country_code:
+        assert location.country_code is None
+    else:
+        assert location.country_code == case.country_code
+
+    if case.expect_warning:
+        assert len(caplog.records) > 0
+        assert "Found issues in location:" in caplog.records[0].message
+    else:
+        assert len(caplog.records) == 0