Add unified jinja obs list to marine DA

ci/cases/gfsv17/marine3dvar.yaml

@@ -22,5 +22,5 @@ prepoceanobs:
 marineanl:
   SOCA_INPUT_FIX_DIR: {{ HOMEgfs }}/fix/gdas/soca/1440x1080x75/soca
   SOCA_ANL_GEOM: {{ HOMEgfs }}/fix/gdas/soca/720x540x75/soca
-  SOCA_OBS_LIST: {{ HOMEgfs }}/sorc/gdas.cd/parm/soca/obs/obs_list.yaml
+  SOCA_OBS_LIST: {{ HOMEgfs }}/sorc/gdas.cd/parm/soca/obs/obs_list.yaml.j2
   SOCA_NINNER: 100

jobs/JGLOBAL_PREP_OCEAN_OBS

@@ -1,5 +1,5 @@
 #!/bin/bash
-source "${HOMEgfs}/ush/jjob_header.sh" -e "prepoceanobs" -c "base prepoceanobs"
+source "${HOMEgfs}/ush/jjob_header.sh" -e "prepoceanobs" -c "base marineanl prepoceanobs"
 
 
 ##############################################

parm/config/gfs/config.marineanl.j2

@@ -20,5 +20,6 @@ export MARINE_UTILITY_YAML_TMPL="${PARMgfs}/gdas/soca/soca_utils_stage.yaml.j2"
 export MARINE_ENSDA_STAGE_BKG_YAML_TMPL="${PARMgfs}/gdas/soca/ensda/stage_ens_mem.yaml.j2"
 export MARINE_DET_STAGE_BKG_YAML_TMPL="${PARMgfs}/gdas/soca/soca_det_bkg_stage.yaml.j2"
 export MARINE_JCB_GDAS_ALGO="${PARMgfs}/gdas/jcb-gdas/algorithm/marine"
+export MARINE_JCB_GDAS_OBS="${PARMgfs}/gdas/jcb-gdas/observations/marine"
 
 echo "END: config.marineanl"

parm/config/gfs/config.prepoceanobs.j2

@@ -8,10 +8,7 @@ export OCNOBS2IODAEXEC="${HOMEgfs}/sorc/gdas.cd/build/bin/gdas_obsprovider2ioda.
 
 export SOCA_INPUT_FIX_DIR="{{ SOCA_INPUT_FIX_DIR }}"
 
-export MARINE_OBS_YAML_DIR="${PARMgfs}/gdas/soca/obs/config"
 export OBSPREP_YAML="{{ OBSPREP_YAML }}"
-export OBS_LIST="{{ SOCA_OBS_LIST }}"
-export OBS_YAML=${OBS_LIST}
 
 # ocean analysis needs own dmpdir until standard dmpdir has full ocean obs
 use_exp_obs="{{ use_exp_obs }}"
@@ -21,10 +18,6 @@ fi
 
 export DMPDIR="${dmpdir_exp:-${DMPDIR}}"
 
-# For BUFR2IODA json and python scripts
-export JSON_TMPL_DIR="${PARMgfs}/gdas/ioda/bufr2ioda"
-export BUFR2IODA_PY_DIR="${USHgfs}"
-
 # Get task specific resources
 . "${EXPDIR}/config.resources" prepoceanobs
 echo "END: config.prepoceanobs"

parm/config/gfs/yaml/defaults.yaml

@@ -57,13 +57,12 @@ snowanl:
 marineanl:
   SOCA_INPUT_FIX_DIR: "${FIXgfs}/gdas/soca/72x35x25/soca"
   SOCA_ANL_GEOM: "${FIXgfs}/gdas/soca/72x35x25/soca"
-  SOCA_OBS_LIST: "${PARMgfs}/gdas/soca/obs/obs_list.yaml"  # TODO: This is also repeated in oceanprepobs
+  SOCA_OBS_LIST: "${PARMgfs}/gdas/soca/obs/obs_list.yaml.j2"  # TODO: This is also repeated in oceanprepobs
   SOCA_NINNER: 100
   JCB_ALGO_YAML_VAR: "${PARMgfs}/gdas/soca/marine-jcb-3dfgat.yaml.j2"
 
 prepoceanobs:
   SOCA_INPUT_FIX_DIR: "${FIXgfs}/gdas/soca/72x35x25/soca"
-  SOCA_OBS_LIST: "${PARMgfs}/gdas/soca/obs/obs_list.yaml"  # TODO: This is also repeated in ocnanal
   OBSPREP_YAML: "${PARMgfs}/gdas/soca/obsprep/obsprep_config.yaml"
   use_exp_obs: "YES"
   dmpdir_exp: "${BASE_DATA}/experimental_obs"

ush/python/pygfs/task/marine_analysis.py

@@ -25,21 +25,6 @@
 logger = getLogger(__name__.split('.')[-1])
 
 
-def parse_obs_list_file(obs_list_yaml_path):
-    # Get the list of observation types from the obs_list.yaml
-    obs_types = []
-    with open(obs_list_yaml_path, 'r') as file:
-        for line in file:
-            # Remove leading/trailing whitespace and check if the line is uncommented
-            line = line.strip()
-            if line.startswith('- !INC') and not line.startswith('#'):
-                # Extract the type using regex
-                match = re.search(r'\$\{MARINE_OBS_YAML_DIR\}/(.+)\.yaml', line)
-                if match:
-                    obs_types.append(str(match.group(1)))
-    return obs_types
-
-
 class MarineAnalysis(Task):
     """
     Class for global marine analysis tasks
@@ -70,7 +55,8 @@ def __init__(self, config):
                 'ENSPERT_RELPATH': _enspert_relpath,
                 'CALC_SCALE_EXEC': _calc_scale_exec,
                 'OPREFIX': f"{self.task_config.RUN}.t{self.task_config.cyc:02d}z.",
-                'APREFIX': f"{self.task_config.RUN}.t{self.task_config.cyc:02d}z."
+                'APREFIX': f"{self.task_config.RUN}.t{self.task_config.cyc:02d}z.",
+                'app_path_observations': self.task_config.MARINE_JCB_GDAS_OBS
             }
         )
 
@@ -134,16 +120,23 @@ def _fetch_observations(self: Task) -> None:
         """
 
         # get the list of observations
-        obs_list_config = YAMLFile(self.task_config.MARINE_OBS_LIST_YAML)
-        obs_list_config = Template.substitute_structure(obs_list_config, TemplateConstants.DOLLAR_PARENTHESES, self.task_config)
-        obs_list_config = {'observations': obs_list_config}
-        logger.info(f"{obs_list_config}")
+
+        # "observations" is expected by later JCB code to populate it with config info,
+        # but the obs_list as such is needed later
+        self.task_config.observations = parse_j2yaml(self.task_config.MARINE_OBS_LIST_YAML, self.task_config)['observations']
+        self.task_config.obs_list = self.task_config.observations
+
+        obsconfigfile = os.path.join(self.task_config['PARMgfs'], 'gdas/soca/obs/obs_list_base_yaml.j2')
+        self.task_config.observations = parse_j2yaml(obsconfigfile, self.task_config)
 
         obs_files = []
-        for ob in obs_list_config['observations']['observers']:
-            logger.info(f"******** {self.task_config.OPREFIX}{ob['obs space']['name'].lower()}.{to_YMD(self.task_config.PDY)}{self.task_config.cyc:02d}.nc4")
-            obs_files.append(f"{self.task_config.OPREFIX}{ob['obs space']['name'].lower()}.{to_YMD(self.task_config.PDY)}{self.task_config.cyc:02d}.nc4")
-        obs_list = []
+
+        for observer in self.task_config['observations']['observers']:
+            filename = f"{self.task_config.OPREFIX}{observer['obs space']['name'].lower()}.{to_YMD(self.task_config.PDY)}{self.task_config.cyc:02d}.nc4"
+            logger.info(f"******** {filename}")
+            obs_files.append(filename)
+
+        obs_files_to_copy = []
 
         # copy obs from COM_OBS to DATA/obs
         for obs_file in obs_files:
@@ -153,11 +146,11 @@ def _fetch_observations(self: Task) -> None:
             logger.info(f"******* {obs_src}")
             if os.path.exists(obs_src):
                 logger.info(f"******* fetching {obs_file}")
-                obs_list.append([obs_src, obs_dst])
+                obs_files_to_copy.append([obs_src, obs_dst])
             else:
                 logger.info(f"******* {obs_file} is not in the database")
 
-        FileHandler({'copy': obs_list}).sync()
+        FileHandler({'copy': obs_files_to_copy}).sync()
 
     @logit(logger)
     def _prep_scratch_dir(self: Task) -> None:
@@ -217,15 +210,15 @@ def _prep_variational_yaml(self: Task) -> None:
         envconfig_jcb['PDY'] = os.getenv('PDY')
         envconfig_jcb['cyc'] = os.getenv('cyc')
         envconfig_jcb['SOCA_NINNER'] = self.task_config.SOCA_NINNER
-        envconfig_jcb['obs_list'] = ['adt_rads_all']
         envconfig_jcb['HOMEgfs'] = self.task_config.HOMEgfs
         envconfig_jcb['DO_TEST_MODE'] = self.task_config.DO_TEST_MODE
         envconfig_jcb['RUN'] = self.task_config.RUN
         envconfig_jcb['current_cycle'] = self.task_config.current_cycle
         envconfig_jcb['MOM6_LEVS'] = mdau.get_mom6_levels(str(self.task_config.OCNRES).zfill(3))
+        envconfig_jcb['observations'] = self.task_config.observations
 
         # Write obs_list_short
-        save_as_yaml(parse_obs_list_file(self.task_config.MARINE_OBS_LIST_YAML), 'obs_list_short.yaml')
+        save_as_yaml(self.task_config['obs_list'], 'obs_list_short.yaml')
         os.environ['OBS_LIST_SHORT'] = 'obs_list_short.yaml'
 
         # Render the JCB configuration files

ush/python/pygfs/task/marine_letkf.py

@@ -54,6 +54,11 @@ def __init__(self, config: Dict) -> None:
         self.task_config.mom_input_nml = os.path.join(self.task_config.DATA, 'mom_input.nml')
         self.task_config.obs_dir = os.path.join(self.task_config.DATA, 'obs')
         self.task_config.ENSPERT_RELPATH = _enspert_relpath
+        self.task_config.PARMsoca = os.path.join(self.task_config.PARMgfs, 'gdas', 'soca')
+        self.task_config.cyc = os.getenv('cyc')
+        self.task_config.PDY = os.getenv('PDY')
+        self.task_config.app_path_observations = self.task_config.MARINE_JCB_GDAS_OBS
+        self.task_config.letkf_app = "true"
 
     @logit(logger)
     def initialize(self):
@@ -71,8 +76,11 @@ def initialize(self):
         # make directories and stage ensemble background files
         soca_fix_stage_list = parse_j2yaml(self.task_config.SOCA_FIX_YAML_TMPL, self.task_config)
         FileHandler(soca_fix_stage_list).sync()
-        stageconf = AttrDict()
-        keys = ['current_cycle',
+        stageconfig = AttrDict()
+        keys = ['app_path_observations',
+                'cyc',
+                'current_cycle',
+                'letkf_app',
                 'previous_cycle',
                 'COM_ICE_LETKF_TMPL',
                 'COM_OCEAN_LETKF_TMPL',
@@ -83,37 +91,48 @@ def initialize(self):
                 'COMOUT_ICE_LETKF',
                 'COMOUT_OCEAN_LETKF',
                 'DATA',
+                'DIST_HALO_SIZE',
                 'ENSPERT_RELPATH',
                 'GDUMP_ENS',
                 'NMEM_ENS',
                 'OPREFIX',
                 'PARMgfs',
+                'PDY',
                 'ROTDIR',
                 'RUN',
                 'WINDOW_BEGIN',
                 'WINDOW_MIDDLE']
         for key in keys:
-            stageconf[key] = self.task_config[key]
+            stageconfig[key] = self.task_config[key]
 
-        # stage ensemble background files
-        soca_ens_bkg_stage_list = parse_j2yaml(self.task_config.MARINE_ENSDA_STAGE_BKG_YAML_TMPL, stageconf)
-        FileHandler(soca_ens_bkg_stage_list).sync()
+        jcb_base_yaml = os.path.join(self.task_config.PARMsoca, 'marine-jcb-base.yaml')
+        jcb_base_config = parse_j2yaml(path=jcb_base_yaml, data=stageconfig)
+
+        jcb_config = {**jcb_base_config, **stageconfig}
 
         # stage letkf-specific files
-        letkf_stage_list = parse_j2yaml(self.task_config.MARINE_LETKF_STAGE_YAML_TMPL, stageconf)
+        letkf_stage_list = parse_j2yaml(self.task_config.MARINE_LETKF_STAGE_YAML_TMPL, jcb_config)
         FileHandler(letkf_stage_list).sync()
 
-        obs_list = parse_j2yaml(self.task_config.MARINE_OBS_LIST_YAML, self.task_config)
+        # stage ensemble background files
+        soca_ens_bkg_stage_list = parse_j2yaml(self.task_config.MARINE_ENSDA_STAGE_BKG_YAML_TMPL, stageconfig)
+        FileHandler(soca_ens_bkg_stage_list).sync()
+
+        # "observations" is expected by later JCB code to populate it with config info,
+        jcb_config['observations'] = parse_j2yaml(self.task_config.MARINE_OBS_LIST_YAML, jcb_config)['observations']
+
+        obsconfigfile = os.path.join(self.task_config['PARMgfs'], 'gdas/soca/obs/obs_list_base_yaml.j2')
+        jcb_config['observations'] = parse_j2yaml(obsconfigfile, jcb_config)
 
-        # get the list of observations
+        # get the list of expected observation files
         obs_files = []
-        for ob in obs_list['observers']:
-            obs_name = ob['obs space']['name'].lower()
+        for observer in jcb_config['observations']['observers']:
+            obs_name = observer['obs space']['name'].lower()
             # TODO(AFE) - this should be removed when the obs config yamls are jinjafied
-            if 'distribution' not in ob['obs space']:
-                ob['obs space']['distribution'] = {'name': 'Halo', 'halo size': self.task_config['DIST_HALO_SIZE']}
+            if 'distribution' not in observer['obs space']:
+                observer['obs space']['distribution'] = {'name': 'Halo', 'halo size': self.task_config['DIST_HALO_SIZE']}
             obs_filename = f"{self.task_config.OPREFIX}{obs_name}.{to_YMDH(self.task_config.current_cycle)}.nc4"
-            obs_files.append((obs_filename, ob))
+            obs_files.append((obs_filename, observer))
 
         obs_files_to_copy = []
         obs_to_use = []
@@ -131,10 +150,12 @@ def initialize(self):
         FileHandler({'copy': obs_files_to_copy}).sync()
 
         # make the letkf.yaml
-        letkf_yaml = parse_j2yaml(self.task_config.MARINE_LETKF_YAML_TMPL, stageconf)
+        # TODO (AFE) switch to fully JCB version
+        letkf_yaml = parse_j2yaml(self.task_config.MARINE_LETKF_YAML_TMPL, jcb_config)
         letkf_yaml.observations.observers = obs_to_use
         letkf_yaml.save(self.task_config.letkf_yaml_file)
 
+        # TODO(AFE) get rid of this, I think
         # swap date and stack size in mom_input.nml
         domain_stack_size = self.task_config.DOMAIN_STACK_SIZE
         ymdhms = [int(s) for s in self.task_config.WINDOW_BEGIN.strftime('%Y,%m,%d,%H,%M,%S').split(',')]
@@ -190,19 +211,18 @@ def finalize(self):
         for key in keys:
             letkfsaveconf[key] = self.task_config[key]
 
-        # get the list of obs output files
-        obs_list = parse_j2yaml(letkfsaveconf.MARINE_OBS_LIST_YAML, self.task_config)
+        # get the list of obs output file - letkf yaml is already complete
+        letkf_config = parse_j2yaml(self.task_config.letkf_yaml_file, AttrDict())
         obs_files = []
-        for ob in obs_list['observers']:
-            obs_files.append(ob['obs space']['obsdataout']['engine']['obsfile'])
+        for observer in letkf_config['observations']['observers']:
+            obs_files.append(observer['obs space']['obsdataout']['engine']['obsfile'])
         obs_files_to_copy = []
-        # copy obs from diags to COMOUT
+        # copy files from diags to COMOUT
         for obs_src in obs_files:
             obs_dst = os.path.join(letkfsaveconf.COMOUT_OCEAN_LETKF, 'diags',
                                    os.path.basename(obs_src))
             if os.path.exists(obs_src):
                 obs_files_to_copy.append([obs_src, obs_dst])
-        # stage the desired diag files
         FileHandler({'mkdir': [os.path.join(letkfsaveconf.COMOUT_OCEAN_LETKF, 'diags')]}).sync()
         FileHandler({'copy': obs_files_to_copy}).sync()
         letkf_save_list = parse_j2yaml(self.task_config.MARINE_LETKF_SAVE_YAML_TMPL, letkfsaveconf)