Updated StationConfiguration IO to handle unknown attributes from input

src/pypromice/postprocess/create_bufr_files.py

@@ -28,11 +28,13 @@ def create_bufr_files(
     Generate hourly bufr files from the for all input files
 
     :param input_files: Paths to csv l3 hourly data files
+    :param station_configuration_root: Root directory containing station configuration toml files
     :param period_start: Datetime string for period start. Eg '2024-01-01T00:00' or '20240101
     :param period_end: Datetime string for period end
     :param output_root: Output dir for both bufr files for individual stations and compiled. Organized in two sub directories.
     :param override: If False: Skip a period if the compiled output file exists.
     :param break_on_error: If True: Stop processing if an error occurs
+    :param output_filename_suffix: Suffix for the compiled output file
     :return:
     """
     periods = pd.date_range(period_start, period_end, freq="H")
@@ -41,7 +43,10 @@ def create_bufr_files(
     output_individual_root.mkdir(parents=True, exist_ok=True)
     output_compiled_root.mkdir(parents=True, exist_ok=True)
 
-    station_configuration_mapping = load_station_configuration_mapping(station_configuration_root)
+    station_configuration_mapping = load_station_configuration_mapping(
+        station_configuration_root,
+        skip_unexpected_fields=True,
+    )
 
     for period in periods:
         period: pd.Timestamp

src/pypromice/postprocess/get_bufr.py

@@ -462,7 +462,8 @@ def main():
             input_files += map(Path, glob.glob(path.as_posix()))
 
     station_configuration_mapping = load_station_configuration_mapping(
-        args.station_configurations_root
+        args.station_configurations_root,
+        skip_unexpected_fields=True,
     )
 
     get_bufr(

src/pypromice/station_configuration.py

@@ -1,3 +1,4 @@
+import logging
 from pathlib import Path
 from typing import Optional, Dict, Mapping, Sequence
 
@@ -30,6 +31,7 @@ class StationConfiguration:
     sonic_ranger_from_gps: Optional[float] = None
     static_height_of_gps_from_mean_sea_level: Optional[float] = None
     station_relocation: Sequence[str] = attrs.field(factory=list)
+    location_type: Optional[str] = None
 
     # The station data will be exported to BUFR if True. Otherwise, it will only export latest position
     export_bufr: bool = False
@@ -45,8 +47,22 @@ class StationConfiguration:
     positions_update_timestamp_only: bool = False
 
     @classmethod
-    def load_toml(cls, path):
-        return cls(**toml.load(path))
+    def load_toml(cls, path, skip_unexpected_fields=False):
+        config_fields = {field.name for field in attrs.fields(cls)}
+        input_dict = toml.load(path)
+        unexpected_fields = set(input_dict.keys()) - config_fields
+        if unexpected_fields:
+            if skip_unexpected_fields:
+                logging.info(
+                    f"Skipping unexpected fields in toml file {path}: "
+                    + ", ".join(unexpected_fields)
+                )
+                for field in unexpected_fields:
+                    input_dict.pop(field)
+            else:
+                raise ValueError(f"Unexpected fields: {unexpected_fields}")
+
+        return cls(**input_dict)
 
     def dump_toml(self, path: Path):
         with path.open("w") as fp:
@@ -58,6 +74,7 @@ def as_dict(self) -> Dict:
 
 def load_station_configuration_mapping(
     configurations_root_dir: Path,
+    **kwargs,
 ) -> Mapping[str, StationConfiguration]:
     """
     Load station configurations from toml files in configurations_root_dir
@@ -66,14 +83,16 @@ def load_station_configuration_mapping(
     ----------
     configurations_root_dir
         Root directory containing toml files
+    kwargs
+        Additional arguments to pass to StationConfiguration.load_toml
 
     Returns
     -------
     Mapping from stid to StationConfiguration
 
     """
     return {
-        config_file.stem: StationConfiguration(**toml.load(config_file))
+        config_file.stem: StationConfiguration.load_toml(config_file, **kwargs)
         for config_file in configurations_root_dir.glob("*.toml")
     }
 

tests/unit/test_station_config.py

@@ -55,6 +55,57 @@ def test_read_toml(self):
                 station_configuration,
             )
 
+    def test_read_toml_with_unexpected_field(self):
+        with TemporaryDirectory() as temp_dir:
+            source_path = Path(temp_dir) / "UPE_L.toml"
+            source_str = """
+                stid = "UPE_L"
+                station_site = "UPE_L"
+                project = "Promice"
+                station_type = "mobile"
+                wmo_id = "04423"
+                barometer_from_gps = -0.25
+                anemometer_from_sonic_ranger = 0.4
+                temperature_from_sonic_ranger = 0.0
+                height_of_gps_from_station_ground = 0.9
+                sonic_ranger_from_gps = 1.3
+                export_bufr = true
+                skipped_variables = []
+                positions_update_timestamp_only = false
+                an_unexpected_field = 42
+            """
+            with source_path.open("w") as source_io:
+                source_io.writelines(source_str)
+
+            expected_configuration = StationConfiguration(
+                stid="UPE_L",
+                station_site="UPE_L",
+                project="Promice",
+                station_type="mobile",
+                wmo_id="04423",
+                barometer_from_gps=-0.25,
+                anemometer_from_sonic_ranger=0.4,
+                temperature_from_sonic_ranger=0.0,
+                height_of_gps_from_station_ground=0.9,
+                sonic_ranger_from_gps=1.3,
+                export_bufr=True,
+                comment=None,
+                skipped_variables=[],
+                positions_update_timestamp_only=False,
+            )
+
+            with self.assertRaises(ValueError):
+                StationConfiguration.load_toml(source_path)
+
+            station_configuration = StationConfiguration.load_toml(source_path, skip_unexpected_fields=True)
+
+            self.assertEqual(
+                expected_configuration,
+                station_configuration,
+            )
+
+
+
     def test_write_read(self):
         with TemporaryDirectory() as temp_dir:
             output_path = Path(temp_dir) / "UPE_L.toml"