30 documentation

README.md

@@ -47,20 +47,7 @@ This library doesn't have knowledge of those datasets, it forms the basis for pr
 
 ## Limitations
 
-There are some major limitations to this as a general purpose tool, these will hopefully be dealt with in time! I'm raising issues as I go
-
-**This is currently very heavy development functionality, but the following commands already work**:
-
-* preprocess_missing_spatial - poorly at present due to missing mask backref implementation
-* preprocess_missing_time
-* preprocess_regrid
-* preprocess_rotate
-* preprocess_dataset
-* preprocess_loader_init
-* preprocess_add_mask
-* preprocess_add_channel
-
-Other stubs probably don't work, unless I forgot to update these docs!
+**This is currently active development, but the commands work to most degrees.**
 
 ## Contributing
 

docs/acknowledgements.md

@@ -0,0 +1,4 @@
+# Acknowledgements
+These are automatically generated from the git repository.
+
+{{ git_site_authors }}

docs/change_log.md

@@ -0,0 +1 @@
+--8<-- "CHANGELOG.md"

docs/commands.md

@@ -0,0 +1,19 @@
+# Commands
+
+## Datasets
+### preprocess_missing_spatial
+### preprocess_missing_time
+### preprocess_regrid
+### preprocess_reproject
+### preprocess_rotate
+
+## Loaders
+### preprocess_loader_init
+### preprocess_loader_copy
+
+### preprocess_add_mask
+### preprocess_add_channel
+
+## AI-ready datasets
+### preprocess_dataset
+### preprocess_add_processed

docs/index.md

@@ -4,8 +4,4 @@ Check out the [usage](usage.md) section for further information, including how t
 
 Check out the [API](preprocess_toolbox/base.md) if you're keen to go deep.
 
-!!! note
-
-    This project is under active development.
-
 {!README.md!}

docs/license.md

@@ -0,0 +1 @@
+--8<-- "LICENSE"

mkdocs.yml

@@ -4,35 +4,45 @@ theme:
   highlightjs: true
 nav:
   - Home: 'index.md'
+  - 'commands.md'
   - 'usage.md'
-  - API:
-    - preprocess.dataset:
-        - "preprocess_toolbox/dataset/cli.md"
-        - "preprocess_toolbox/dataset/process.md"
-        - "preprocess_toolbox/dataset/spatial.md"
-        - "preprocess_toolbox/dataset/time.md"
-    - preprocess.loader:
-        - "preprocess_toolbox/loader/cli.md"
-        - "preprocess_toolbox/loader/utils.md"
-    - "preprocess_toolbox/base.md"
-    - "preprocess_toolbox/interface.md"
-    - "preprocess_toolbox/models.md"
-    - "preprocess_toolbox/processor.md"
-    - "preprocess_toolbox/utils.md"
-    - "preprocess_toolbox/cli.md"
   - GitHub: 'https://github.com/environmental-forecasting/preprocess-toolbox'
 plugins:
   - search
+  # Enables automatically generating authors from git repo, see acknowledgements.md
+  - git-authors
+  # Automatically generates API section from docstrings
+  - mkdocs-autoapi:
+      autoapi_add_nav_entry: API
+      autoapi_ignore:
+        - setup.py
+  - autorefs:
+      resolve_closest: true
   - mkdocstrings:
       handlers:
         python:
+          paths:
+            - .
           options:
-            docstring_style: sphinx
+            show_submodules: true
+            docstring_style: google
             heading_level: 3
 markdown_extensions:
   - markdown_include.include:
       base_path: .
   - admonition
+  # Enables code block highlighting
+  - pymdownx.highlight:
+      anchor_linenums: true
+      line_spans: __span
+      pygments_lang_class: true
+  - pymdownx.inlinehilite
+  - pymdownx.snippets
+  - pymdownx.superfences
+  - pymdownx.blocks.caption
+  - pymdownx.tabbed:
+      alternate_style: true
+copyright: Copyright © 2025 British Antarctic Survey / Alan Turing Institute
 
 # Include extra CSS to make some style adjustments
 extra_css:

preprocess_toolbox/base.py

@@ -144,7 +144,7 @@ def get_dataset(self,
                      for var_filepaths in self.processed_files[vn].values()]
 
         logging.info("Got {} filenames to open dataset with!".format(len(var_files)))
-        logging.debug(pformat(var_files))
+        # logging.debug(pformat(var_files))
 
         # TODO: where's my parallel mfdataset please!?
         with (dask.config.set(**{'array.slicing.split_large_chunks': True})):

preprocess_toolbox/dataset/process.py

@@ -150,6 +150,9 @@ def rotate_dataset(ref_file: os.PathLike,
                     wind_cubes[vars_to_rotate[1]],
                     angles,
                 )
+            if len(wind_cubes_r[vars_to_rotate[0]].shape) == 2 and len(wind_cubes_r[vars_to_rotate[1]].shape) == 2:
+                wind_cubes_r[vars_to_rotate[0]] = iris.util.new_axis(wind_cubes_r[vars_to_rotate[0]], "time")
+                wind_cubes_r[vars_to_rotate[1]] = iris.util.new_axis(wind_cubes_r[vars_to_rotate[1]], "time")
         except iris.exceptions.CoordinateNotFoundError:
             logging.exception("Failure to rotate due to coordinate issues. "
                               "moving onto next file")

preprocess_toolbox/interface.py

@@ -34,7 +34,6 @@ def get_processor_implementation(config: os.PathLike) -> object:
 
     create_kwargs = dict(**remaining)
     logging.info("Attempting to instantiate {} with loaded configuration".format(implementation))
-    logging.debug("Converted kwargs from the retrieved configuration: {}".format(create_kwargs))
 
     return implementation(**create_kwargs)
 
@@ -57,7 +56,6 @@ def get_processor_from_source(identifier: str, source_cfg: dict) -> object:
 
     create_kwargs = {k: v for k, v in source_cfg.items() if k not in ["dataset_config", "implementation"]}
     logging.info("Attempting to instantiate {} with loaded configuration".format(source_cfg["implementation"]))
-    logging.debug("Converted kwargs from the retrieved configuration: {}".format(create_kwargs))
 
     return get_implementation(source_cfg["implementation"])(
         get_dataset_config_implementation(source_cfg["dataset_config"]),

preprocess_toolbox/processor.py

@@ -95,8 +95,16 @@ def __init__(self,
         self._normalisation_splits = [] if normalisation_splits is None else normalisation_splits
         self._parallel = parallel_opens
         self._refdir = ref_procdir
+
+        ##
+        # Split dates -
+        #
+
         # TODO: splits -> { dates, sources }, but currently sources are separate...
         self._splits = splits
+        self._dropped_split_dates = {}
+        # TODO: add self._dropped_dates based on DATA
+
         self._source_files = dict()
 
         if init_source:
@@ -223,45 +231,49 @@ def _init_source_data(self,
         :return:
         """
 
-        split_dates_required = dict()
         drop_dates = dict()
+        all_dates = dict()
 
         for split in self._splits.keys():
-            dates = sorted(self._splits[split])
+            all_dates[split] = sorted(self._splits[split])
             drop_dates[split] = list()
 
-            if dates:
+            if all_dates[split]:
                 logging.info("Processing {} dates for {} category: {} - {}".
-                             format(len(dates), split, min(dates), max(dates)))
+                             format(len(all_dates[split]), split, min(all_dates[split]), max(all_dates[split])))
             else:
                 logging.info("No {} dates for this processor".format(split))
                 continue
 
             # Calculating lead and lag dates that aren't already accounted for in splits
-            if self._lag_time > 0:
+            if self._lag_time >= 0:
                 logging.info("Including lag of {} {}s".format(self._lag_time, ds_config.frequency.attribute))
-                additional_lag_dates, dropped_lag_dates = get_extension_dates(ds_config, dates, self._lag_time, reverse=True)
-                dates += additional_lag_dates
+                additional_lag_dates, dropped_lag_dates = get_extension_dates(
+                    ds_config, all_dates[split],
+                    # We offset by two, because -1 is channel one, so we need to account for lag == 1 being -2
+                    self._lag_time + 2,
+                    start_step=1, reverse=True)
+                all_dates[split] += additional_lag_dates
                 drop_dates[split] += dropped_lag_dates
                 logging.info("Lag added {} dates for {} category: {} - {}".
-                             format(len(dates), split, min(dates), max(dates)))
+                             format(len(all_dates[split]), split, min(all_dates[split]), max(all_dates[split])))
             if self._lead_time > 0:
                 logging.info("Including lead of {} {}s".format(self._lead_time, ds_config.frequency.attribute))
-                additional_lead_dates, dropped_lead_dates = get_extension_dates(ds_config, dates, self._lead_time)
-                dates += additional_lead_dates
+                additional_lead_dates, dropped_lead_dates = get_extension_dates(ds_config, all_dates[split], self._lead_time)
+                all_dates[split] += additional_lead_dates
                 drop_dates[split] += dropped_lead_dates
                 logging.info("Lead added {} dates for {} category: {} - {}".
-                             format(len(dates), split, min(dates), max(dates)))
+                             format(len(all_dates[split]), split, min(all_dates[split]), max(all_dates[split])))
 
-            split_dates_required[split] = sorted([_ for _ in dates if _ not in drop_dates[split]])
+            self._dropped_split_dates[split] = sorted(drop_dates[split])
+            all_dates[split] = sorted([_ for _ in all_dates[split] if _ not in drop_dates[split]])
 
         for split in self._splits.keys():
-            self._source_files[split] = {var_config.name: ds_config.var_filepaths(var_config, split_dates_required[split])
+            self._source_files[split] = {var_config.name: ds_config.var_filepaths(var_config, all_dates[split])
                                          for var_config in ds_config.variables}
 
             for var_name, var_files in self._source_files[split].items():
                 logging.info("Got {} files for {}:{}".format(len(var_files), split, var_name))
-        logging.debug(pformat(self._source_files))
 
     def _normalise_array_mean(self, var_name: str, da: object, denormalise: bool=False):
         """
@@ -341,7 +353,6 @@ def _normalise_array_scaling(self, var_name: str, da: object, denormalise: bool=
         elif self.norm_split_dates:
             logging.debug("Generating norm-scaling min-max from {} training "
                           "dates".format(len(self.norm_split_dates)))
-
             norm_samples = da.sel(time=self.norm_split_dates).data
             norm_samples = norm_samples.ravel()
 
@@ -375,7 +386,8 @@ def _process_channel(self,
                                                 for split, var_files in self.source_files.items()
                                                 for vn, files in var_files.items()
                                                 for file in files
-                                                if var_name == vn])))
+                                                if var_name == vn
+                                                and os.path.exists(file)])))
 
                 if len(source_files) > 0:
                     logging.info("Opening {} files for {}".format(len(source_files), var_name))
@@ -384,22 +396,15 @@ def _process_channel(self,
                     # data so this was harder. Now we work with whatever we get from download-toolbox
                     ds = xr.open_mfdataset(
                         source_files,
-                        # Solves issue with inheriting files without
-                        # time dimension (only having coordinate)
-                        combine="nested",
-                        concat_dim="time",
-                        coords="minimal",
-                        compat="override",
-                        # TODO: review this, but if lat-lon is in the file, it's signalling bigger issues
-                        # drop_variables=("lat", "lon"),
-                        parallel=self._parallel)
+                        parallel=self._parallel,
+                        lock=False)
                     da = getattr(ds, var_name)
                     da = da.astype(self.dtype)
+                    logging.debug("Files to be opened: {}".format(da.dims))
 
                     # FIXME: we should ideally store train dates against the
                     #  normalisation and climatology, to ensure recalculation on
                     #  reprocess. All this need be is in the path, to be honest
-
                     if var_suffix == "anom":
                         if len(self._anom_clim_splits) < 1 and self._refdir is None:
                             raise ProcessingError("You must provide a list of splits via "
@@ -507,7 +512,9 @@ def get_config(self, **kwargs):
             "path": self.path,
             "processed_files": self._processed_files,
             "source_files": self._source_files,
-            "splits": self.splits,
+            "splits": {split: [
+                date for date in dates if date not in self._dropped_split_dates[split]
+            ] for split, dates in self._splits.items()},
         }
 
     @staticmethod
@@ -588,7 +595,6 @@ def lead_time(self) -> int:
 
     @property
     def norm_split_dates(self):
-        # TODO: functools.cached_property, though slightly odd behaviour re. write-ability
         return [date
                 for clim_split in self._normalisation_splits
                 for date in self._splits[clim_split]]