Feature/split by month

cads_adaptors/adaptors/mars.py

@@ -51,6 +51,7 @@ def execute_mars(
     request: dict[str, Any] | list[dict[str, Any]],
     context: Context,
     config: dict[str, Any] = dict(),
+    mapping: dict[str, Any] = dict(),
     target: str = "data.grib",
 ) -> str:
     from cads_mars_server import client as mars_client
@@ -63,7 +64,7 @@ def execute_mars(
         requests, _cacheable = implement_embargo(requests, config["embargo"])
 
     split_on_keys = ALWAYS_SPLIT_ON + ensure_list(config.get("split_on", []))
-    requests = split_requests_on_keys(requests, split_on_keys)
+    requests = split_requests_on_keys(requests, split_on_keys, context, mapping)
 
     mars_servers = get_mars_server_list(config)
 
@@ -178,7 +179,10 @@ def retrieve_list_of_results(self, request: dict[str, Any]) -> list[str]:
         request = self.normalise_request(request)
 
         result: Any = execute_mars(
-            self.mapped_requests, context=self.context, config=self.config
+            self.mapped_requests,
+            context=self.context,
+            config=self.config,
+            mapping=self.mapping,
         )
 
         with dask.config.set(scheduler="threads"):

cads_adaptors/adaptors/multi.py

@@ -195,7 +195,12 @@ def retrieve_list_of_results(self, request: Request) -> list[str]:
         self.context.add_stdout(
             f"MultiMarsCdsAdaptor, mapped_requests: {mapped_requests}"
         )
-        result = execute_mars(mapped_requests, context=self.context, config=self.config)
+        result = execute_mars(
+            mapped_requests,
+            context=self.context,
+            config=self.config,
+            mapping=self.mapping,
+        )
 
         with dask.config.set(scheduler="threads"):
             paths = self.convert_format(

cads_adaptors/tools/general.py

@@ -1,3 +1,5 @@
+from collections import defaultdict
+from datetime import datetime
 from typing import Any
 
 
@@ -10,13 +12,47 @@ def ensure_list(input_item: Any) -> list:
     return [input_item]
 
 
+SPLIT_BY_MONTH_KEY = "__split_by_month"
+
+
+def group_by_month(values, date_format):
+    dates = [datetime.strptime(val, date_format) for val in values]
+    months = defaultdict(list)
+    for date in dates:
+        month = date.replace(day=1)
+        months[month].append(date.strftime(date_format))
+    return list(months.values())
+
+
 def split_requests_on_keys(
-    requests: list[dict[str, Any]], split_on_keys: list[str]
+    requests: list[dict[str, Any]],
+    split_on_keys: list[str],
+    context=None,
+    mapping: dict[str, Any] = dict(),
 ) -> list[dict]:
     """Split a request on keys, returning a list of requests."""
+    import cads_adaptors.mapping as mapping_module
+
     if len(split_on_keys) == 0:
         return requests
 
+    split_by_month = SPLIT_BY_MONTH_KEY in split_on_keys
+    if split_by_month:
+        mapping_options = mapping.get("options", {})
+        if not mapping_options.get("wants_dates", False):
+            split_by_month = False
+            if context:
+                context.add_stderr(
+                    "For the time being, split-by-month is only supported for wants_dates=True!"
+                )
+
+    if split_by_month:
+        date_keyword_configs = mapping_options.get(
+            "date_keyword_config", mapping_module.DATE_KEYWORD_CONFIGS
+        )
+        if isinstance(date_keyword_configs, dict):
+            date_keyword_configs = [date_keyword_configs]
+
     for key in split_on_keys:
         out_requests = []
         for request in requests:
@@ -25,11 +61,21 @@ def split_requests_on_keys(
                 if len(values) == 1:
                     out_requests.append(request)
                 else:
+                    if split_by_month:
+                        for date_keyword_config in date_keyword_configs:
+                            date_key = date_keyword_config.get("date_keyword", "date")
+                            format_key = date_keyword_config.get(
+                                "format_keyword", "date_format"
+                            )
+                            date_format = mapping_options.get(format_key, "%Y-%m-%d")
+                            if key == date_key:
+                                values = group_by_month(values, date_format)
                     for value in values:
                         new_request = request.copy()
                         new_request[key] = value
                         out_requests.append(new_request)
             else:
                 out_requests.append(request)
         requests = out_requests
+
     return out_requests