Improve NCI observation handling

src/CSET/cset_workflow/app/metplus_ascii2nc/file/nci/ascii2nc.conf

@@ -43,16 +43,8 @@ LEAD_SEQ =
 # https://metplus.readthedocs.io/en/latest/Users_Guide/systemconfiguration.html#directory-and-filename-template-info
 ###
 ASCII2NC_INPUT_DIR = {INPUT_BASE}
-ASCII2NC_INPUT_TEMPLATE = *.ascii
+ASCII2NC_INPUT_TEMPLATE = obs.{valid?fmt=%Y%m%dT%H%MZ}.ascii
 ASCII2NC_INPUT_FORMAT = met_point
 
 ASCII2NC_OUTPUT_DIR = {ENV[CYLC_WORKFLOW_SHARE_DIR]}/obs_nc
 ASCII2NC_OUTPUT_TEMPLATE = {valid?fmt=%Y%m%dT%H}.nc
-
-###
-# ASCII2NC Settings
-# https://metplus.readthedocs.io/en/latest/Users_Guide/wrappers.html#ascii2nc
-###
-
-ASCII2NC_WINDOW_BEGIN = 0
-ASCII2NC_WINDOW_END = 0

src/CSET/cset_workflow/app/metplus_point_stat/file/nci/pointstat.conf

@@ -36,10 +36,7 @@ INIT_TIME_FMT = %Y%m%dT%H
 INIT_BEG = {ENV[TASK_START_TIME]}
 INIT_END = {ENV[TASK_START_TIME]}
 INIT_INCREMENT = 1H
-LEAD_SEQ = begin_end_incr(0,{ENV[FORECAST_LENGTH]},1)
-
-# Number of seconds to shift times in the fcst file (try half the time increment)
-FCST_SHIFT = 1800
+LEAD_SEQ = begin_end_incr(4,{ENV[FORECAST_LENGTH]},1)
 
 ###
 # File I/O

src/CSET/cset_workflow/app/metplus_point_stat/file/nci/surface.conf

@@ -17,15 +17,15 @@ POINT_STAT_FCST_FILE_TYPE = NETCDF_NCCF
 POINT_STAT_MESSAGE_TYPE = "ADPSFC"
 
 FCST_POINT_STAT_VAR1_NAME = air_temperature
-FCST_POINT_STAT_VAR1_LEVELS = "({valid?fmt=%Y%m%d_%H%M%S?shift={FCST_SHIFT}},*,*)"
+FCST_POINT_STAT_VAR1_LEVELS = "({valid?fmt=%Y%m%d_%H%M%S},*,*)"
 FCST_POINT_STAT_VAR1_THRESH = <=273, >273
 
 OBS_POINT_STAT_VAR1_NAME = t2m
 OBS_POINT_STAT_VAR1_LEVELS = Z0
 OBS_POINT_STAT_VAR1_THRESH = <=273, >273
 
 FCST_POINT_STAT_VAR2_NAME = relative_humidity
-FCST_POINT_STAT_VAR2_LEVELS = "({valid?fmt=%Y%m%d_%H%M%S?shift={FCST_SHIFT}},*,*)"
+FCST_POINT_STAT_VAR2_LEVELS = "({valid?fmt=%Y%m%d_%H%M%S},*,*)"
 FCST_POINT_STAT_VAR2_THRESH =  <60, >95
 
 OBS_POINT_STAT_VAR2_NAME = rh2m

src/CSET/cset_workflow/app/metplus_prep_obs/bin/odb2/bom.py

@@ -5,12 +5,13 @@
 patterns.
 """
 
+from abc import abstractmethod
 from typing import Iterable
 
 from metomi.isodatetime.data import TimePoint
 from pandas import DataFrame
 
-from .odb2 import PrepODB2, read_file, read_tarfile
+from .odb2 import PrepODB2
 
 # Valid bureau forecast systems
 BOM_SYSTEMS = ["access_g3", "access_g4"]
@@ -32,20 +33,30 @@ def __init__(self, system: str):
         Args:
             system: BOM system to source observations from (e.g. 'access_c3_dn')
         """
+        super().__init__()
         self.system = system
 
+        # Filter out observations rejected by OPS
+        self.odb_where += (
+            " AND ops_report_flags@hdr = 0 AND ops_datum_flags.b0@body = 0"
+        )
+
+    @abstractmethod
     def read_c3_type(self, type: str, valid_time: TimePoint) -> Iterable[DataFrame]:
         """Read C3 data pattern."""
         raise NotImplementedError
 
+    @abstractmethod
     def read_c4_type(self, type: str, valid_time: TimePoint) -> Iterable[DataFrame]:
         """Read C4 data pattern."""
         raise NotImplementedError
 
+    @abstractmethod
     def read_g3_type(self, type: str, valid_time: TimePoint) -> Iterable[DataFrame]:
         """Read G3 data pattern."""
         raise NotImplementedError
 
+    @abstractmethod
     def read_g4_type(self, type: str, valid_time: TimePoint) -> Iterable[DataFrame]:
         """Read G4 data pattern."""
         raise NotImplementedError
@@ -58,16 +69,16 @@ def read_type(self, type: str, valid_time: TimePoint) -> Iterable[DataFrame]:
             type: observation type (ODB2 file name without the suffix)
             valid_time: time to load
 
-        Returns an iterable of ODB2 DataFrames
+        Returns an iterable of ODB2 DataFrames containing all the observations for the given type and time.
         """
         if self.system.startswith("access_c3"):
-            yield from self.read_c3_type(type, valid_time)
+            return self.read_c3_type(type, valid_time)
         elif self.system.startswith("access_c4"):
-            yield from self.read_c4_type(type, valid_time)
+            return self.read_c4_type(type, valid_time)
         elif self.system.startswith("access_g3"):
-            yield from self.read_g3_type(type, valid_time)
+            return self.read_g3_type(type, valid_time)
         elif self.system.startswith("access_g4"):
-            yield from self.read_g4_type(type, valid_time)
+            return self.read_g4_type(type, valid_time)
 
     def read_odb(self, valid_time: TimePoint) -> Iterable[DataFrame]:
         """
@@ -76,7 +87,7 @@ def read_odb(self, valid_time: TimePoint) -> Iterable[DataFrame]:
         Args:
             valid_time: time to load
 
-        Returns an iterable of ODB2 DataFrames
+        Returns an iterable of ODB2 DataFrames containing all the observations for the given time.
         """
         if self.system.startswith("access_c"):
             for type in access_c_types:
@@ -107,25 +118,25 @@ def read_c3_type(self, type: str, valid_time: TimePoint) -> Iterable[DataFrame]:
         domain = self.system.split("_")[-1]
         tarfile = f"/g/data/ig2/odb2/access_c3/%Y/%m/%Y%m%dT%H%MZ/%Y%m%dT%H%MZ_{domain}_odb2.tar.zst"
         obsfile = f"ukv_odb2/{type}.odb"
-        return read_tarfile(tarfile, obsfile, valid_time)
+        return self.read_tarfile(tarfile, obsfile, valid_time)
 
     def read_c4_type(self, type: str, valid_time: TimePoint) -> Iterable[DataFrame]:
         """Read C4 data pattern."""
         domain = self.system.split("_")[-1]
         tarfile = f"/g/data/ig2/odb2/access_c4/%Y/%m/%Y%m%dT%H%MZ/%Y%m%dT%H%MZ_{domain}_odb2.tgz"
         obsfile = f"ukv_odb2/{type}.odb"
-        return read_tarfile(tarfile, obsfile, valid_time)
+        return self.read_tarfile(tarfile, obsfile, valid_time)
 
     def read_g3_type(self, type: str, valid_time: TimePoint) -> Iterable[DataFrame]:
         """Read G3 data pattern."""
         pattern = (
             "/g/data/ig2/odb2/access_g3/%Y/%m/%Y%m%dT%H%MZ/glm_odb2/{type}.odb.zst"
         )
-        return read_file(pattern, valid_time)
+        return self.read_file(pattern, valid_time)
 
     def read_g4_type(self, type: str, valid_time: TimePoint) -> Iterable[DataFrame]:
         """Read G4 data pattern."""
         pattern = (
             "/g/data/ig2/odb2/access_g4/%Y/%m/%Y%m%dT%H%MZ/glm_odb2/{type}.odb.zst"
         )
-        return read_file(pattern, valid_time)
+        return self.read_file(pattern, valid_time)