add test for brightness observation

api/api/service/brightness_servicer.py

@@ -31,11 +31,10 @@ def GetBrightnessObservation(self, request, context):
 
         location = EarthLocation.from_geodetic(lon * u.degree, lat * u.degree)
         site = ObserverSite(location=location)
-        meteo_client = OpenMeteoClient(site=site)
 
         try:
-            cloud_cover, elevation = meteo_client.get_hourly_values_at_site()
-            logging.debug(f"meteo_client response at {lat},{lon} is {cloud_cover}o, {elevation}m")
+            meteo_client = OpenMeteoClient(site=site)
+            cloud_cover, elevation = meteo_client.get_forecast()
         except Exception as e:
             import traceback
 
@@ -59,10 +58,11 @@ def GetBrightnessObservation(self, request, context):
                 ],
                 dtype=torch.float32,
             ).unsqueeze(0)
-            logging.debug(f"vector for site is {x}")
+
 
             with torch.no_grad():
                 predicted_y = model(x)
+                logging.debug(f"predicted {x} to be {predicted_y}")
 
             time_utc = datetime.now(timezone.utc)
             observation = brightness_service_pb2.BrightnessObservation(

api/api/service/open_meteo/open_meteo_client.py

@@ -7,6 +7,7 @@
 from ..observer_site import ObserverSite
 from ..utils import get_astro_time_hour
 
+model = config["meteo"]["model"]
 
 class OpenMeteoClient:
     def __init__(self, site: ObserverSite) -> None:
@@ -17,21 +18,20 @@ def __init__(self, site: ObserverSite) -> None:
         port = config["meteo"]["port"]
         self.url_base = f"{protocol}://{host}:{port}"
 
-    def get_hourly_values_at_site(self) -> t.Tuple[int, float]:
-        """ask open meteo for cloud cover and elevation for the observer site"""
+    def get_forecast(self) -> t.Tuple[int, float]:
         lat, lon = self.site.latitude.value, self.site.longitude.value
-        model = config["meteo"]["model"]
+
+        hourly_params = {"temperature_2m", "cloud_cover"}
         params = {
             "latitude": lat,
             "longitude": lon,
             "models": model,
-            "hourly": "temperature_2m,cloud_cover"
+            "hourly": ",".join(hourly_params)
         }
         r = requests.get(f"{self.url_base}/v1/forecast", params=params)
         r.raise_for_status()
 
         res_json = r.json()
-
         elevation = float(res_json.get("elevation", 0.))
 
         idx = self.get_hourly_index_of_site_time()

api/tests/test_api.py

@@ -1,4 +1,5 @@
 from concurrent import futures
+from unittest.mock import patch
 
 import pytest
 import grpc
@@ -11,7 +12,6 @@
 @pytest.fixture(scope='module')
 def grpc_server():
     server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
-    # example_pb2_grpc.add_GreeterServicer_to_server(Greeter(), server)
     brightness_service_pb2_grpc.add_BrightnessServiceServicer_to_server(BrightnessServicer(), server)
     server.add_insecure_port('[::]:50051')
     server.start()
@@ -26,17 +26,29 @@ def brightness_service_stub(grpc_server):
         yield stub
 
 def test_apis(brightness_service_stub):
-    request= brightness_service_pb2.Coordinates(lat=42.,lon=42.)
+    request = brightness_service_pb2.Coordinates(lat=42.,lon=42.)
 
-    response=brightness_service_stub.GetPollution(request)
+    response = brightness_service_stub.GetPollution(request)
     assert isinstance(response.r,int)
     assert isinstance(response.g,int)
     assert isinstance(response.b,int)
     assert isinstance(response.a,int)
 
-@pytest.mark.skip
-def test_brightness_observation(brightness_service_stub):
-    request= brightness_service_pb2.Coordinates(lat=42.,lon=42.)
+@patch('api.service.open_meteo.open_meteo_client.requests.get')
+def test_brightness_observation(mock_get, brightness_service_stub):
+    from datetime import datetime,UTC
 
-    response=brightness_service_stub.GetBrightnessObservation(request)
-    assert response is not None
+    mock_response = mock_get.return_value
+    mock_response.status_code = 200
+    mock_response.json.return_value={"elevation":0.,"hourly":{"cloud_cover":[0]*24}}
+
+    lat,lon = (42.,42.)
+    request = brightness_service_pb2.Coordinates(lat=lat,lon=lon)
+
+    response = brightness_service_stub.GetBrightnessObservation(request)
+
+    assert response.lat == lat
+    assert response.lon == lon
+    parsed_utc_iso = datetime.fromisoformat(response.utc_iso)
+    assert parsed_utc_iso.date() == datetime.now(UTC).date()
+    assert response.mpsas > 0 and response.mpsas < 22

pp/tests/test_publisher.py

@@ -28,13 +28,9 @@ def mock_grpc_client(mocker):
 @pytest.fixture
 def mock_pika_channel(mocker):
     channel_mock = MagicMock()
-    # create a mock connection that returns the mock channel
     connection_mock = MagicMock()
     connection_mock.channel.return_value = channel_mock
-
-    # patch the BlockingConnection to return the mock connection
     mocker.patch("pika.BlockingConnection", return_value=connection_mock)
-
     return channel_mock