Merge pull request #214 from EURODEO/issue_178-levels_multiple_range

Issue 178 levels filtering with combination

api/openapi/custom_dimension_examples.py

@@ -36,6 +36,10 @@
         "summary": "Open start range",
         "value": "../10.0",
     },
+    "Combination": {
+        "summary": "Combination",
+        "value": "1.0, 1.5/1.8, R5/2.0/2.0",
+    },
 }
 
 periods = {
@@ -54,6 +58,10 @@
         "summary": "Open end range",
         "value": "PT0S/..",
     },
+    "Combination": {
+        "summary": "Combination",
+        "value": "PT0S, PT1M/PT10M",
+    },
 }
 
 methods = {

api/test/test_api_utilities.py

@@ -0,0 +1,19 @@
+import pytest
+
+from utilities import get_levels_values
+
+levels_in_out = [
+    ("1", ["100"]),
+    ("1,2", ["100", "200"]),
+    ("1,2, 3", ["100", "200", "300"]),
+    ("1/3", ["100/300"]),
+    ("../3", ["../300"]),
+    ("1/..", ["100/.."]),
+    ("R3/1.2/0.3", ["120", "150", "180"]),
+    ("1, 3/5, R3/5/0.1,11", ["100", "300/500", "500", "510", "520", "1100"]),
+]
+
+
+@pytest.mark.parametrize("levels_in, levels_out", levels_in_out)
+def test_get_levels_values(levels_in, levels_out):
+    assert get_levels_values(levels_in) == levels_out

api/test/test_edr.py

@@ -410,3 +410,33 @@ def test_get_data_with_lowercase_period_range_without_existing_data():
 
         assert response.status_code == 404
         assert response.json() == {"detail": "Requested data not found."}
+
+
+def test_get_data_with_combination_levels_filtering():
+    with patch("routers.edr.get_obs_request") as mock_get_obs_request:
+
+        # Load with random test data for making a mock_obs_request
+        test_data = load_json("test/test_data/test_coverages_proto.json")
+        mock_get_obs_request.return_value = create_mock_obs_response(test_data)
+
+        response = client.get("/collections/observations/locations/0-20000-0-06260?levels=R6/0.0/0.1, 1.5/1.8, 10")
+
+        m_args = mock_get_obs_request.call_args[0][0]
+
+        assert response.status_code == 200
+        assert m_args.filter["level"].values == ["0", "10", "20", "30", "40", "50", "150/180", "1000"]
+
+
+def test_get_data_with_combination_periods_filtering():
+    with patch("routers.edr.get_obs_request") as mock_get_obs_request:
+
+        # Load with random test data for making a mock_obs_request
+        test_data = load_json("test/test_data/test_coverages_proto.json")
+        mock_get_obs_request.return_value = create_mock_obs_response(test_data)
+
+        response = client.get("/collections/observations/locations/0-20000-0-06260?periods=PT0S, PT1M/PT10M")
+
+        m_args = mock_get_obs_request.call_args[0][0]
+
+        assert response.status_code == 200
+        assert m_args.filter["period"].values == ["0", "60/600"]

api/utilities.py

@@ -5,6 +5,7 @@
 from datetime import datetime
 from datetime import timedelta
 from typing import Tuple
+from itertools import chain
 
 from isodate import ISO8601Error
 import datastore_pb2 as dstore
@@ -155,7 +156,7 @@ async def add_request_parameters(
         request.filter["standard_name"].values.extend(split_and_strip(standard_names))
 
     if levels:
-        request.filter["level"].values.extend(get_z_levels_or_range(levels))
+        request.filter["level"].values.extend(get_levels_values(levels))
 
     if methods:
         request.filter["function"].values.extend(split_and_strip(methods))
@@ -179,25 +180,40 @@ def get_periods_or_range(periods: str) -> list[str]:
         raise HTTPException(status_code=400, detail=f"{err}")
 
 
-def get_z_levels_or_range(z: str) -> list[str]:
+def get_levels_values(levels: str) -> list[str]:
     """
-    Function for getting the levels as a list or a range
+    Function for getting the levels filters as a list of levels, ranges,
+    range intervals or combination of the previous
     """
     # it can be z=value1,value2,value3: z=2,10,80
     # or z=minimum value/maximum value: z=10/100
     # or z=Rn/min height/height interval: z=R20/100/50
-    try:
-        split_on_slash = z.split("/")
-        if len(split_on_slash) == 2:
-            z_min = convert_m_to_cm(split_on_slash[0]) if split_on_slash[0] != ".." else -sys.maxsize - 1
-            z_max = convert_m_to_cm(split_on_slash[1]) if split_on_slash[1] != ".." else sys.maxsize
-            return [f"{z_min}/{z_max}"]
-        elif len(split_on_slash) > 2:
-            return get_z_values_from_interval(split_on_slash)
-        else:
-            return [convert_m_to_cm(level) for level in z.split(",")]
-    except ValueError:
-        raise HTTPException(status_code=400, detail=f"Invalid levels value: {z}")
+    # or a combination of the above: z=10,30/100,200,300,R20/100/50
+
+    def get_level_or_range(z: str) -> list[str]:
+        try:
+            split_on_slash = z.split("/")
+            if len(split_on_slash) == 2:
+                return get_z_values_from_range(split_on_slash)
+            elif len(split_on_slash) > 2:
+                return get_z_values_from_interval(split_on_slash)
+            else:
+                return [convert_m_to_cm(z)]
+        except ValueError:
+            raise HTTPException(status_code=400, detail=f"Invalid levels value: {z}")
+
+    return list(chain(*[get_level_or_range(z) for z in split_and_strip(levels)]))
+
+
+def get_z_values_from_range(range: list[str]) -> list[str]:
+    """
+    Function for getting the z values from a range.
+    """
+    start, end = range[0], range[1]
+    z_min = convert_m_to_cm(start) if start != ".." else start
+    z_max = convert_m_to_cm(end) if end != ".." else end
+
+    return [f"{z_min}/{z_max}"]
 
 
 def get_z_values_from_interval(interval: list[str]) -> list[str]: