Support BUFR-to-IODA ingest

src/swell/configuration/jedi/interfaces/geos_atmosphere/ingest_observations/ncep_1bamua_bufr.yaml

@@ -0,0 +1,2 @@
+acquisition_method: local
+source: ioda/1bamua/*.nc4

src/swell/configuration/jedi/interfaces/geos_atmosphere/ingest_observations/ncep_mtiasi_bufr.yaml

@@ -0,0 +1,2 @@
+acquisition_method: local
+source: ioda/mtiasi/*.nc4

src/swell/suites/ingest_obs/flow.cylc

@@ -40,7 +40,9 @@
         {% for cycle_time in cycle_times %}
             {{cycle_time.cycle_time}} = """
                 {% for model_component in model_components %}
-                {% if download_convert_pipeline %}
+                {% if bufr_pipeline %}
+                BufrToIoda-{{model_component}} => IngestObs-{{model_component}}
+                {% elif download_convert_pipeline %}
                 DownloadObs-{{model_component}} => ConvertObsToIoda-{{model_component}}
                 BuildJediByLinking[^]? | BuildJedi[^]  => ConvertObsToIoda-{{model_component}}
                 ConvertObsToIoda-{{model_component}} => IngestObs-{{model_component}}
@@ -83,6 +85,12 @@
 
     {% for model_component in model_components %}
 
+    {% if bufr_pipeline %}
+    [[BufrToIoda-{{model_component}}]]
+        script = "swell task BufrToIoda $config -d $datetime -m {{model_component}}"
+        execution time limit = PT30M
+    {% endif %}
+
     {% if download_convert_pipeline %}
     [[DownloadObs-{{model_component}}]]
         script = "swell task DownloadObs $config -d $datetime -m {{model_component}}"

src/swell/suites/ingest_obs/suite_config.py

@@ -22,6 +22,28 @@ class SuiteConfig(QuestionContainer, Enum):
     )
     # This name should be unique and not conflict with other suites
     # (otherwise it might get overwritten)
+    ingest_obs_atmosphere = QuestionList(
+        list_name="ingest_obs_atmosphere",
+        questions=[
+            ingest_obs,
+            qd.start_cycle_point("2023-10-10T00:00:00Z"),
+            qd.final_cycle_point("2023-10-10T00:00:00Z"),
+            qd.model_components(['geos_atmosphere']),
+            qd.runahead_limit("P5"),
+            qd.bufr_pipeline(True),
+        ],
+        geos_atmosphere=[
+            qd.window_length("PT6H"),
+            qd.cycle_times(['T00']),
+            qd.bufr_dir(
+                "/discover/nobackup/fgoktas/SwellExperiments/"
+                "swell-convert_bufr/run/%Y%m%dT000000Z/geos_atmosphere/bufr/"
+            ),
+            qd.obs_to_ingest(['ncep_1bamua_bufr', 'ncep_mtiasi_bufr']),
+            qd.dry_run(True),
+        ]
+    )
+
     ingest_obs_marine = QuestionList(
         list_name="ingest_obs_marine",
         questions=[

src/swell/tasks/bufr_to_ioda.py

@@ -185,7 +185,9 @@ def get_bufr_mapping_yaml(self,
     def execute(self) -> None:
 
         # Set Bufr File Directory (Input)
-        bufr_dir = os.path.join(self.cycle_dir(), 'bufr')
+        bufr_dir_config = self.config.bufr_dir(None)
+        bufr_dir = (self.cycle_time_dto().strftime(bufr_dir_config)
+                    if bufr_dir_config else os.path.join(self.cycle_dir(), 'bufr'))
 
         # Set Ioda File Directory (Output) and create if needed
         ioda_dir = os.path.join(self.cycle_dir(), 'ioda')

src/swell/tasks/ingest_obs.py

@@ -158,18 +158,47 @@ def process_obs_config(
         dt = datetime.strptime(cycle_time, "%Y-%m-%dT%H:%M:%SZ")
 
         if acquisition_method == 'local':
-            # File was produced locally for this cycle (e.g. by ConvertObsToIoda).
-            # 'source' is a path relative to cycle_dir using strftime placeholders.
+            # File was produced locally for this cycle (e.g. by ConvertObsToIoda or BufrToIoda).
+            # 'source' is a path relative to cycle_dir, supports strftime and glob patterns.
             source = config.get('source')
             if not source:
                 msg = f"No 'source' key in {obs_name}.yaml for acquisition_method 'local'."
                 self.logger.error(msg)
                 raise ValueError(msg)
-            target_file = os.path.join(self.cycle_dir(), dt.strftime(source))
-            if not os.path.exists(target_file):
-                self.logger.warning(f"Local file not found: {target_file}")
+            expanded = os.path.join(self.cycle_dir(), dt.strftime(source))
+            matched_files = glob.glob(expanded)
+            if not matched_files:
+                self.logger.warning(f"Local file(s) not found: {expanded}")
                 return ingested, [(obs_name, "File not found")]
 
+            for target_file in matched_files:
+                if dry_run:
+                    self.logger.info(f"  [DRY RUN] Would ingest:")
+                    self.logger.info(f"    Obs Name: {obs_name}")
+                    self.logger.info(f"    Provider: {provider}")
+                    self.logger.info(f"    Method: {acquisition_method}")
+                    self.logger.info(f"    Source: {target_file}")
+                    ingested.append(target_file)
+                else:
+                    try:
+                        r2d2.store(
+                            item='observation',
+                            provider=provider,
+                            observation_type=obs_name,
+                            file_extension=os.path.splitext(target_file)[1][1:],
+                            window_start=window_start,
+                            window_length=window_length,
+                            source_file=target_file,
+                        )
+                    except (ValueError, KeyError, FileNotFoundError,
+                            OSError, requests.RequestException) as e:
+                        self.logger.error(f"Failed to ingest {obs_name}: {e}")
+                        failed.append((obs_name, str(e)))
+                    else:
+                        ingested.append(target_file)
+                        self.logger.info(f"Successfully ingested {obs_name} from {target_file}")
+            return ingested, failed
+
         else:
             # Remote/static path: 'cp_source' or 's3_source' with Skylab-style placeholders.
             source_pattern = config.get(f'{acquisition_method}_source')

src/swell/utilities/question_defaults.py

@@ -326,6 +326,21 @@ class background_time_offset(TaskQuestion):
         widget_type: WType = WType.ISO_DURATION
 
     # --------------------------------------------------------------------------------------------------
+    @dataclass
+    class bufr_dir(TaskQuestion):
+        default_value: str = "/path/to/bufr/%Y%m%d/"
+        question_name: str = "bufr_dir"
+        ask_question: bool = True
+        options: str = "defer_to_model"
+        models: List[str] = mutable_field([
+            "geos_atmosphere"
+        ])
+        prompt: str = ("Path to the directory containing raw BUFR files. "
+                       "Supports strftime placeholders (e.g. /path/to/bufr/%Y%m%d/).")
+        widget_type: WType = WType.STRING
+
+    # --------------------------------------------------------------------------------------------------
+
     @dataclass
     class bufr_obs_classes(TaskQuestion):
         default_value: str = "defer_to_model"
@@ -1581,6 +1596,17 @@ class window_type(TaskQuestion):
         widget_type: WType = WType.STRING_DROP_LIST
 
 # --------------------------------------------------------------------------------------------------
+    @dataclass
+    class bufr_pipeline(SuiteQuestion):
+        default_value: bool = False
+        question_name: str = "bufr_pipeline"
+        ask_question: bool = False
+        prompt: str = ("Run the BufrToIoda and IngestObs tasks? "
+                       "(BufrToIoda -> IngestObs) to R2D2")
+        widget_type: WType = WType.BOOLEAN
+
+    # --------------------------------------------------------------------------------------------------
+
     @dataclass
     class download_convert_pipeline(SuiteQuestion):
         default_value: bool = False