diff --git a/compass/landice/ismip7/archive.py b/compass/landice/ismip7/archive.py index 1e4d773f85..87e951bb41 100644 --- a/compass/landice/ismip7/archive.py +++ b/compass/landice/ismip7/archive.py @@ -149,8 +149,8 @@ def resolve_fracture_source(config, file_pattern): scenario = config.get('ismip7', 'scenario') base_path = os.path.join( _ice_sheet_path(config), model, scenario, 'fracture') - requested_version = config.get( - 'ismip7_fracture', 'version', fallback='latest') + requested_version = config['ismip7_fracture'].get( + 'version', fallback='latest') version_path, version, files = resolve_version_directory( base_path, requested_version, file_pattern) diff --git a/compass/landice/ismip7/remap.py b/compass/landice/ismip7/remap.py index 720ef39e85..6f6eaa4459 100644 --- a/compass/landice/ismip7/remap.py +++ b/compass/landice/ismip7/remap.py @@ -10,6 +10,60 @@ from scipy.ndimage import distance_transform_edt +def netcdf_file_is_valid(path, varname=None, require_time=False, + logger=None): + """ + Return True if ``path`` exists, opens cleanly, contains ``varname`` and + (optionally) a ``time`` coordinate, and its last time record can + actually be read. Used to detect incomplete intermediate files left by + an interrupted (timed-out) run so they can be regenerated instead of + silently reused. The last-record read catches a file whose header is + complete but whose data section was truncated mid-write (which opens + cleanly but fails later at concatenation). + + Parameters + ---------- + path : str + Path to the NetCDF file to check + + varname : str, optional + Variable name to verify is present; if None, only checks file opens + + require_time : bool, optional + Whether to require a ``time`` coordinate/variable + + logger : logging.Logger, optional + Logger for status messages (logs incomplete-file info if provided) + + Returns + ------- + bool + True if the file is valid and complete; False otherwise + """ + if not os.path.exists(path): + return False + try: + with xr.open_dataset(path, decode_times=False) as ds: + if varname is not None and varname not in ds.variables: + return False + if require_time and "time" not in ds.variables: + return False + # Force-read the last record to detect a truncated data + # section (header OK, but bytes cut off mid-file). + if varname is not None: + var = ds[varname] + if "time" in var.dims: + var.isel(time=-1).load() + else: + var.load() + except Exception as exc: + if logger is not None: + logger.info(f" Ignoring incomplete/unreadable file " + f"{os.path.basename(path)}: {exc}") + return False + return True + + def extrapolate_source(input_file, output_file, varnames, logger): """ Extrapolate fill/missing values on the source polar stereographic grid diff --git a/compass/landice/tests/ismip7_forcing/atmosphere/process_runoff.py b/compass/landice/tests/ismip7_forcing/atmosphere/process_runoff.py index 94db1faf9c..f1976b03bb 100644 --- a/compass/landice/tests/ismip7_forcing/atmosphere/process_runoff.py +++ b/compass/landice/tests/ismip7_forcing/atmosphere/process_runoff.py @@ -9,7 +9,10 @@ mapping_file_name, resolve_atmosphere_source, ) -from compass.landice.ismip7.remap import extrapolate_source +from compass.landice.ismip7.remap import ( + extrapolate_source, + netcdf_file_is_valid, +) from compass.step import Step @@ -93,6 +96,16 @@ def run(self): logger.info(f"Found {len(input_files)} runoff files for years " f"{start_year}-{end_year}") + # Check if final output already exists; skip if so + output_file = (f"{mali_mesh_name}_runoff_{model}_{scenario}_" + f"{source.version}_{start_year}-{end_year}.nc") + output_path = os.path.join(output_base_path, forcing_group, + "atmosphere") + dst = os.path.join(output_path, output_file) + if os.path.exists(dst): + logger.info(f"Output already exists, skipping: {dst}") + return + # The mapping file is supplied by the build_mapping_file step. mapping_file = mapping_file_name( config, "atm", source.source_grid, method_remap) @@ -104,16 +117,25 @@ def run(self): remapped_file = f"remapped_{basename}" remapped_files.append(remapped_file) - if os.path.exists(remapped_file): + if netcdf_file_is_valid(remapped_file, "mrro", + require_time=True, logger=logger): logger.info(f" Remapped file exists, skipping: {basename}") continue + # Incomplete remapped file from an interrupted run: drop and + # redo + if os.path.exists(remapped_file): + logger.info(f" Reprocessing incomplete remapped file: " + f"{basename}") + os.remove(remapped_file) # Extrapolate fill values on source grid before remapping # so they don't pollute neighboring cells during interpolation extrap_file = f"extrap_{basename}" - if not os.path.exists(extrap_file): - extrapolate_source(input_file, extrap_file, "mrro", - logger) + if not netcdf_file_is_valid(extrap_file, "mrro", + logger=logger): + if os.path.exists(extrap_file): + os.remove(extrap_file) + extrapolate_source(input_file, extrap_file, "mrro", logger) logger.info(f" Remapping: {basename}") args = ["ncremap", @@ -129,9 +151,6 @@ def run(self): # Combine remapped files and rename to MALI conventions logger.info("Combining remapped files and renaming variables...") - output_file = (f"{mali_mesh_name}_runoff_{model}_{scenario}_" - f"{start_year}-{end_year}.nc") - self._combine_and_rename(remapped_files, output_file) # Clean up remapped files @@ -141,12 +160,9 @@ def run(self): os.remove(f) # Place output in appropriate directory - output_path = os.path.join(output_base_path, forcing_group, - "atmosphere") if not os.path.exists(output_path): os.makedirs(output_path) - dst = os.path.join(output_path, output_file) shutil.copy(output_file, dst) logger.info(f"Done. Output: {dst}") diff --git a/compass/landice/tests/ismip7_forcing/atmosphere/process_smb.py b/compass/landice/tests/ismip7_forcing/atmosphere/process_smb.py index e84ea3c0c0..d169dda839 100644 --- a/compass/landice/tests/ismip7_forcing/atmosphere/process_smb.py +++ b/compass/landice/tests/ismip7_forcing/atmosphere/process_smb.py @@ -9,7 +9,10 @@ mapping_file_name, resolve_atmosphere_source, ) -from compass.landice.ismip7.remap import extrapolate_source +from compass.landice.ismip7.remap import ( + extrapolate_source, + netcdf_file_is_valid, +) from compass.step import Step @@ -93,6 +96,16 @@ def run(self): logger.info(f"Found {len(input_files)} SMB files for years " f"{start_year}-{end_year}") + # Check if final output already exists; skip if so + output_file = (f"{mali_mesh_name}_SMB_{model}_{scenario}_" + f"{source.version}_{start_year}-{end_year}.nc") + output_path = os.path.join(output_base_path, forcing_group, + "atmosphere") + dst = os.path.join(output_path, output_file) + if os.path.exists(dst): + logger.info(f"Output already exists, skipping: {dst}") + return + # The mapping file is supplied by the build_mapping_file step. mapping_file = mapping_file_name( config, "atm", source.source_grid, method_remap) @@ -104,14 +117,24 @@ def run(self): remapped_file = f"remapped_{basename}" remapped_files.append(remapped_file) - if os.path.exists(remapped_file): + if netcdf_file_is_valid(remapped_file, "acabf", + require_time=True, logger=logger): logger.info(f" Remapped file exists, skipping: {basename}") continue + # Incomplete remapped file from an interrupted run: drop and + # redo + if os.path.exists(remapped_file): + logger.info(f" Reprocessing incomplete remapped file: " + f"{basename}") + os.remove(remapped_file) # Extrapolate fill values on source grid before remapping # so they don't pollute neighboring cells during interpolation extrap_file = f"extrap_{basename}" - if not os.path.exists(extrap_file): + if not netcdf_file_is_valid(extrap_file, "acabf", + logger=logger): + if os.path.exists(extrap_file): + os.remove(extrap_file) extrapolate_source(input_file, extrap_file, "acabf", logger) logger.info(f" Remapping: {basename}") @@ -128,9 +151,6 @@ def run(self): # Combine remapped files and rename to MALI conventions logger.info("Combining remapped files and renaming variables...") - output_file = (f"{mali_mesh_name}_SMB_{model}_{scenario}_" - f"{start_year}-{end_year}.nc") - self._combine_and_rename(remapped_files, output_file) # Clean up remapped files @@ -140,12 +160,9 @@ def run(self): os.remove(f) # Place output in appropriate directory - output_path = os.path.join(output_base_path, forcing_group, - "atmosphere") if not os.path.exists(output_path): os.makedirs(output_path) - dst = os.path.join(output_path, output_file) shutil.copy(output_file, dst) logger.info(f"Done. Output: {dst}") diff --git a/compass/landice/tests/ismip7_forcing/atmosphere/process_smb_gradient.py b/compass/landice/tests/ismip7_forcing/atmosphere/process_smb_gradient.py index 9f38bde8d7..20473acf00 100644 --- a/compass/landice/tests/ismip7_forcing/atmosphere/process_smb_gradient.py +++ b/compass/landice/tests/ismip7_forcing/atmosphere/process_smb_gradient.py @@ -9,7 +9,10 @@ mapping_file_name, resolve_atmosphere_source, ) -from compass.landice.ismip7.remap import extrapolate_source +from compass.landice.ismip7.remap import ( + extrapolate_source, + netcdf_file_is_valid, +) from compass.step import Step @@ -94,6 +97,16 @@ def run(self): logger.info(f"Found {len(input_files)} SMB gradient files for years " f"{start_year}-{end_year}") + # Check if final output already exists; skip if so + output_file = (f"{mali_mesh_name}_SMB_gradient_{model}_{scenario}_" + f"{source.version}_{start_year}-{end_year}.nc") + output_path = os.path.join(output_base_path, forcing_group, + "atmosphere") + dst = os.path.join(output_path, output_file) + if os.path.exists(dst): + logger.info(f"Output already exists, skipping: {dst}") + return + # The mapping file is supplied by the build_mapping_file step. mapping_file = mapping_file_name( config, "atm", source.source_grid, method_remap) @@ -105,14 +118,24 @@ def run(self): remapped_file = f"remapped_{basename}" remapped_files.append(remapped_file) - if os.path.exists(remapped_file): + if netcdf_file_is_valid(remapped_file, "dacabfdz", + require_time=True, logger=logger): logger.info(f" Remapped file exists, skipping: {basename}") continue + # Incomplete remapped file from an interrupted run: drop and + # redo + if os.path.exists(remapped_file): + logger.info(f" Reprocessing incomplete remapped file: " + f"{basename}") + os.remove(remapped_file) # Extrapolate fill values on source grid before remapping # so they don't pollute neighboring cells during interpolation extrap_file = f"extrap_{basename}" - if not os.path.exists(extrap_file): + if not netcdf_file_is_valid(extrap_file, "dacabfdz", + logger=logger): + if os.path.exists(extrap_file): + os.remove(extrap_file) extrapolate_source(input_file, extrap_file, "dacabfdz", logger) @@ -130,9 +153,6 @@ def run(self): # Combine remapped files and rename to MALI conventions logger.info("Combining remapped files and renaming variables...") - output_file = (f"{mali_mesh_name}_SMB_gradient_{model}_{scenario}_" - f"{start_year}-{end_year}.nc") - self._combine_and_rename(remapped_files, output_file) # Clean up remapped files @@ -142,12 +162,9 @@ def run(self): os.remove(f) # Place output in appropriate directory - output_path = os.path.join(output_base_path, forcing_group, - "atmosphere") if not os.path.exists(output_path): os.makedirs(output_path) - dst = os.path.join(output_path, output_file) shutil.copy(output_file, dst) logger.info(f"Done. Output: {dst}") diff --git a/compass/landice/tests/ismip7_forcing/atmosphere/process_temperature.py b/compass/landice/tests/ismip7_forcing/atmosphere/process_temperature.py index dfd145fe7c..5c9cbf4a09 100644 --- a/compass/landice/tests/ismip7_forcing/atmosphere/process_temperature.py +++ b/compass/landice/tests/ismip7_forcing/atmosphere/process_temperature.py @@ -9,7 +9,10 @@ mapping_file_name, resolve_atmosphere_source, ) -from compass.landice.ismip7.remap import extrapolate_source +from compass.landice.ismip7.remap import ( + extrapolate_source, + netcdf_file_is_valid, +) from compass.step import Step @@ -93,6 +96,16 @@ def run(self): logger.info(f"Found {len(input_files)} temperature files for years " f"{start_year}-{end_year}") + # Check if final output already exists; skip if so + output_file = (f"{mali_mesh_name}_temperature_{model}_{scenario}_" + f"{source.version}_{start_year}-{end_year}.nc") + output_path = os.path.join(output_base_path, forcing_group, + "atmosphere") + dst = os.path.join(output_path, output_file) + if os.path.exists(dst): + logger.info(f"Output already exists, skipping: {dst}") + return + # The mapping file is supplied by the build_mapping_file step. mapping_file = mapping_file_name( config, "atm", source.source_grid, method_remap) @@ -104,14 +117,23 @@ def run(self): remapped_file = f"remapped_{basename}" remapped_files.append(remapped_file) - if os.path.exists(remapped_file): + if netcdf_file_is_valid(remapped_file, "ts", require_time=True, + logger=logger): logger.info(f" Remapped file exists, skipping: {basename}") continue + # Incomplete remapped file from an interrupted run: drop and + # redo + if os.path.exists(remapped_file): + logger.info(f" Reprocessing incomplete remapped file: " + f"{basename}") + os.remove(remapped_file) # Extrapolate fill values on source grid before remapping # so they don't pollute neighboring cells during interpolation extrap_file = f"extrap_{basename}" - if not os.path.exists(extrap_file): + if not netcdf_file_is_valid(extrap_file, "ts", logger=logger): + if os.path.exists(extrap_file): + os.remove(extrap_file) extrapolate_source(input_file, extrap_file, "ts", logger) logger.info(f" Remapping: {basename}") @@ -128,9 +150,6 @@ def run(self): # Combine remapped files and rename to MALI conventions logger.info("Combining remapped files and renaming variables...") - output_file = (f"{mali_mesh_name}_temperature_{model}_{scenario}_" - f"{start_year}-{end_year}.nc") - self._combine_and_rename(remapped_files, output_file) # Clean up remapped files @@ -140,12 +159,9 @@ def run(self): os.remove(f) # Place output in appropriate directory - output_path = os.path.join(output_base_path, forcing_group, - "atmosphere") if not os.path.exists(output_path): os.makedirs(output_path) - dst = os.path.join(output_path, output_file) shutil.copy(output_file, dst) logger.info(f"Done. Output: {dst}") diff --git a/compass/landice/tests/ismip7_forcing/atmosphere/process_temperature_gradient.py b/compass/landice/tests/ismip7_forcing/atmosphere/process_temperature_gradient.py index 47541a0798..5ebb001c9a 100644 --- a/compass/landice/tests/ismip7_forcing/atmosphere/process_temperature_gradient.py +++ b/compass/landice/tests/ismip7_forcing/atmosphere/process_temperature_gradient.py @@ -9,7 +9,10 @@ mapping_file_name, resolve_atmosphere_source, ) -from compass.landice.ismip7.remap import extrapolate_source +from compass.landice.ismip7.remap import ( + extrapolate_source, + netcdf_file_is_valid, +) from compass.step import Step @@ -95,6 +98,17 @@ def run(self): logger.info(f"Found {len(input_files)} temperature gradient files " f"for years {start_year}-{end_year}") + # Check if final output already exists; skip if so + output_file = (f"{mali_mesh_name}_temperature_gradient_{model}_" + f"{scenario}_{source.version}_" + f"{start_year}-{end_year}.nc") + output_path = os.path.join(output_base_path, forcing_group, + "atmosphere") + dst = os.path.join(output_path, output_file) + if os.path.exists(dst): + logger.info(f"Output already exists, skipping: {dst}") + return + # The mapping file is supplied by the build_mapping_file step. mapping_file = mapping_file_name( config, "atm", source.source_grid, method_remap) @@ -106,14 +120,24 @@ def run(self): remapped_file = f"remapped_{basename}" remapped_files.append(remapped_file) - if os.path.exists(remapped_file): + if netcdf_file_is_valid(remapped_file, "dtsdz", + require_time=True, logger=logger): logger.info(f" Remapped file exists, skipping: {basename}") continue + # Incomplete remapped file from an interrupted run: drop and + # redo + if os.path.exists(remapped_file): + logger.info(f" Reprocessing incomplete remapped file: " + f"{basename}") + os.remove(remapped_file) # Extrapolate fill values on source grid before remapping # so they don't pollute neighboring cells during interpolation extrap_file = f"extrap_{basename}" - if not os.path.exists(extrap_file): + if not netcdf_file_is_valid(extrap_file, "dtsdz", + logger=logger): + if os.path.exists(extrap_file): + os.remove(extrap_file) extrapolate_source(input_file, extrap_file, "dtsdz", logger) logger.info(f" Remapping: {basename}") @@ -130,9 +154,6 @@ def run(self): # Combine remapped files and rename to MALI conventions logger.info("Combining remapped files and renaming variables...") - output_file = (f"{mali_mesh_name}_temperature_gradient_{model}_" - f"{scenario}_{start_year}-{end_year}.nc") - self._combine_and_rename(remapped_files, output_file) # Clean up remapped files @@ -142,12 +163,9 @@ def run(self): os.remove(f) # Place output in appropriate directory - output_path = os.path.join(output_base_path, forcing_group, - "atmosphere") if not os.path.exists(output_path): os.makedirs(output_path) - dst = os.path.join(output_path, output_file) shutil.copy(output_file, dst) logger.info(f"Done. Output: {dst}") diff --git a/compass/landice/tests/ismip7_forcing/fracture/process_excess_melt.py b/compass/landice/tests/ismip7_forcing/fracture/process_excess_melt.py index 9a68fcd753..882bf513be 100644 --- a/compass/landice/tests/ismip7_forcing/fracture/process_excess_melt.py +++ b/compass/landice/tests/ismip7_forcing/fracture/process_excess_melt.py @@ -104,6 +104,16 @@ def run(self): basename = os.path.basename(input_file) logger.info(f"Processing excess melt: {basename}") + # Check if final output already exists; skip if so + stem, ext = os.path.splitext(basename) + output_file = f"{mali_mesh_name}_{stem}_{source.version}{ext}" + output_path = os.path.join(output_base_path, f"{model}_{scenario}", + "excess_melt") + dst = os.path.join(output_path, output_file) + if os.path.exists(dst): + logger.info(f"Output already exists, skipping: {dst}") + return + # Build a source file with x/y coordinates and correct orientation gridded_file = f"gridded_{basename}" self._prepare_source_grid(input_file, input_path, gridded_file, @@ -116,8 +126,7 @@ def run(self): # Extrapolate fill values on the source grid before remapping so # they don't pollute neighboring cells during interpolation extrap_file = f"extrap_{basename}" - extrapolate_source(gridded_file, extrap_file, "excess_melt", - logger, decode_times=False) + extrapolate_source(gridded_file, extrap_file, "excess_melt", logger) # Remap the excess melt onto the MALI mesh remapped_file = f"remapped_{basename}" @@ -131,7 +140,6 @@ def run(self): # Rename to MALI conventions logger.info("Renaming variables to MALI conventions...") - output_file = f"{mali_mesh_name}_{basename}" self._rename_to_mali_vars(remapped_file, output_file, start_year, end_year) @@ -141,12 +149,9 @@ def run(self): os.remove(f) # Place output in the appropriate directory - output_path = os.path.join(output_base_path, f"{model}_{scenario}", - "excess_melt") if not os.path.exists(output_path): os.makedirs(output_path) - dst = os.path.join(output_path, output_file) shutil.copy(output_file, dst) logger.info(f"Done. Output: {dst}") diff --git a/compass/landice/tests/ismip7_forcing/fracture/process_lake_properties.py b/compass/landice/tests/ismip7_forcing/fracture/process_lake_properties.py index 839af87ea7..abb6301731 100644 --- a/compass/landice/tests/ismip7_forcing/fracture/process_lake_properties.py +++ b/compass/landice/tests/ismip7_forcing/fracture/process_lake_properties.py @@ -107,6 +107,16 @@ def run(self): basename = os.path.basename(input_file) logger.info(f"Processing lake properties: {basename}") + # Check if final output already exists; skip if so + stem, ext = os.path.splitext(basename) + output_file = f"{mali_mesh_name}_{stem}_{source.version}{ext}" + output_path = os.path.join(output_base_path, f"{model}_{scenario}", + "lake_properties") + dst = os.path.join(output_path, output_file) + if os.path.exists(dst): + logger.info(f"Output already exists, skipping: {dst}") + return + # The mapping file is supplied by the build_mapping_file step. mapping_file = mapping_file_name( config, "fracture", source.source_grid, method_remap) @@ -115,8 +125,7 @@ def run(self): # they don't pollute neighboring cells during interpolation extrap_file = f"extrap_{basename}" extrapolate_source(input_file, extrap_file, - list(self._variables.keys()), logger, - decode_times=False) + list(self._variables.keys()), logger) # Remap both lake property variables onto the MALI mesh remapped_file = f"remapped_{basename}" @@ -130,7 +139,6 @@ def run(self): # Rename to MALI conventions logger.info("Renaming variables to MALI conventions...") - output_file = f"{mali_mesh_name}_{basename}" self._rename_to_mali_vars(remapped_file, output_file, start_year, end_year) @@ -140,12 +148,9 @@ def run(self): os.remove(f) # Place output in the appropriate directory - output_path = os.path.join(output_base_path, f"{model}_{scenario}", - "lake_properties") if not os.path.exists(output_path): os.makedirs(output_path) - dst = os.path.join(output_path, output_file) shutil.copy(output_file, dst) logger.info(f"Done. Output: {dst}") diff --git a/compass/landice/tests/ismip7_forcing/fracture/process_shelf_collapse.py b/compass/landice/tests/ismip7_forcing/fracture/process_shelf_collapse.py index df5d1b6d7c..3d0f9e2625 100644 --- a/compass/landice/tests/ismip7_forcing/fracture/process_shelf_collapse.py +++ b/compass/landice/tests/ismip7_forcing/fracture/process_shelf_collapse.py @@ -9,6 +9,7 @@ ) from compass.landice.ismip7.remap import ( add_xtime_and_write, + netcdf_file_is_valid, open_rename_and_trim, ) from compass.step import Step @@ -94,13 +95,28 @@ def run(self): basename = os.path.basename(input_file) logger.info(f"Processing ice shelf collapse mask: {basename}") + # Check if final output already exists; skip if so + stem, ext = os.path.splitext(basename) + output_file = f"{mali_mesh_name}_{stem}_{source.version}{ext}" + output_path = os.path.join(output_base_path, f"{model}_{scenario}", + "shelf_collapse") + dst = os.path.join(output_path, output_file) + if os.path.exists(dst): + logger.info(f"Output already exists, skipping: {dst}") + return + # The mapping file is supplied by the build_mapping_file step. mapping_file = mapping_file_name( config, "fracture", source.source_grid, method_remap) # Remap the collapse mask onto the MALI mesh remapped_file = f"remapped_{basename}" - if not os.path.exists(remapped_file): + if not netcdf_file_is_valid(remapped_file, "mask", + require_time=True, logger=logger): + if os.path.exists(remapped_file): + logger.info(f"Reprocessing incomplete remapped file: " + f"{basename}") + os.remove(remapped_file) logger.info(f"Remapping: {basename}") args = ["ncremap", "-i", input_file, @@ -111,7 +127,6 @@ def run(self): # Combine time slice and rename to MALI conventions logger.info("Renaming variables to MALI conventions...") - output_file = f"{mali_mesh_name}_{basename}" self._rename_to_mali_vars(remapped_file, output_file, start_year, end_year) @@ -120,12 +135,9 @@ def run(self): os.remove(remapped_file) # Place output in the appropriate directory - output_path = os.path.join(output_base_path, f"{model}_{scenario}", - "shelf_collapse") if not os.path.exists(output_path): os.makedirs(output_path) - dst = os.path.join(output_path, output_file) shutil.copy(output_file, dst) logger.info(f"Done. Output: {dst}") diff --git a/compass/landice/tests/ismip7_forcing/greenland_3d_tf_config.json b/compass/landice/tests/ismip7_forcing/greenland_3d_tf_config.json index b1c90072ec..5dfd730ba7 100644 --- a/compass/landice/tests/ismip7_forcing/greenland_3d_tf_config.json +++ b/compass/landice/tests/ismip7_forcing/greenland_3d_tf_config.json @@ -46,5 +46,5 @@ "b_degC": 0.0901, "c_degC_per_m": 0.000753 }, - "overwrite": true + "overwrite": false } diff --git a/compass/landice/tests/ismip7_forcing/ocean_thermal/build_3d_thermal_forcing.py b/compass/landice/tests/ismip7_forcing/ocean_thermal/build_3d_thermal_forcing.py index ec2a2173a0..fa7bebbb82 100644 --- a/compass/landice/tests/ismip7_forcing/ocean_thermal/build_3d_thermal_forcing.py +++ b/compass/landice/tests/ismip7_forcing/ocean_thermal/build_3d_thermal_forcing.py @@ -71,11 +71,11 @@ def run(self): forcing_2d = os.path.join( ocean_dir, f"{mali_mesh_name}_2dThermalForcing_{label}_" - f"{start_year}-{end_year}.nc") + f"{source.version}_{start_year}-{end_year}.nc") output_file = os.path.join( ocean_dir, f"{mali_mesh_name}_3dThermalForcing_{label}_" - f"{start_year}-{end_year}.nc") + f"{source.version}_{start_year}-{end_year}.nc") # DeltaT/gamma0/basin are calibrated once against OCX and held fixed # for every ESM (see Config.calibrate_delta_t), so they always live diff --git a/compass/landice/tests/ismip7_forcing/ocean_thermal/greenland_3d.py b/compass/landice/tests/ismip7_forcing/ocean_thermal/greenland_3d.py index 87edb2b623..4150e8dcaa 100644 --- a/compass/landice/tests/ismip7_forcing/ocean_thermal/greenland_3d.py +++ b/compass/landice/tests/ismip7_forcing/ocean_thermal/greenland_3d.py @@ -1500,11 +1500,6 @@ def _write_melt_params( ) return temporary = output.with_name(output.name + ".partial") - if temporary.exists(): - raise FileExistsError( - f"Partial melt-parameters output already exists: {temporary}. " - "Remove or rename it after inspecting it." - ) delta_t_by_cell = regional_delta_t[basin_ids - 1] target = xr.Dataset( data_vars={ @@ -1687,11 +1682,6 @@ def _write_forcing_chunk( ) return temporary = output_path.with_name(output_path.name + ".partial") - if temporary.exists(): - raise FileExistsError( - f"Partial output already exists: {temporary}. Remove or rename " - "it after inspecting it." - ) hdf5_temporary = output_path.with_name(output_path.name + ".h5.partial") target = xr.Dataset( data_vars={ @@ -2400,6 +2390,17 @@ def _warn_if_seafloor_below_max_depth( def run(cfg: Config, logger, prepare_only: bool = False) -> None: + # Remove any leftover .partial files from a previous interrupted run. + # These are incomplete temps; the real outputs are named differently. + seen = set() + for d in (cfg.output_file.parent, cfg.melt_params_file.parent): + if d in seen: + continue + seen.add(d) + for stale in sorted(d.glob("*.partial")): + logger.info(f"Removing stale partial file: {stale}") + stale.unlink() + mesh, basin_ids = _load_mesh_and_basins(cfg) profiles = _build_regional_profiles(cfg, mesh, basin_ids, logger) _warn_if_seafloor_below_max_depth(cfg, mesh, basin_ids, profiles, logger) diff --git a/compass/landice/tests/ismip7_forcing/ocean_thermal/process_thermal_forcing.py b/compass/landice/tests/ismip7_forcing/ocean_thermal/process_thermal_forcing.py index b897a2ae69..4f07d89a3e 100644 --- a/compass/landice/tests/ismip7_forcing/ocean_thermal/process_thermal_forcing.py +++ b/compass/landice/tests/ismip7_forcing/ocean_thermal/process_thermal_forcing.py @@ -11,7 +11,10 @@ resolve_version_directory, ) from compass.landice.ismip7.ice_sheet_params import get_params -from compass.landice.ismip7.remap import extrapolate_source +from compass.landice.ismip7.remap import ( + extrapolate_source, + netcdf_file_is_valid, +) from compass.step import Step @@ -255,6 +258,17 @@ def _process_ocean_forcing(self, job, mapping_file, ocean_3d, label = job.label all_files = job.files + # Check if final output already exists; skip if so + tf_label = "3dThermalForcing" if ocean_3d else "2dThermalForcing" + output_file = (f"{mali_mesh_name}_{tf_label}_{label}_" + f"{job.version}_{start_year}-{end_year}.nc") + output_path = os.path.join(output_base_path, forcing_group, + "ocean_thermal_forcing") + dst = os.path.join(output_path, output_file) + if os.path.exists(dst): + logger.info(f"Output already exists, skipping: {dst}") + return + # Filter to files that overlap with the requested year range. # AIS files are named with decade or multi-decade ranges (e.g., # 1850-1859, 1950-2025). GrIS files are named with single years. @@ -283,16 +297,24 @@ def _process_ocean_forcing(self, job, mapping_file, ocean_3d, remapped_file = f"remapped_{basename}" remapped_files.append(remapped_file) - if os.path.exists(remapped_file): + if netcdf_file_is_valid(remapped_file, "tf", require_time=True, + logger=logger): logger.info(f" Remapped file exists, skipping: {basename}") continue + # Incomplete remapped file from an interrupted run: drop and + # redo + if os.path.exists(remapped_file): + logger.info(f" Reprocessing incomplete remapped file: " + f"{basename}") + os.remove(remapped_file) # Extrapolate fill values on source grid before remapping # so they don't pollute neighboring cells during interpolation extrap_file = f"extrap_{basename}" - if not os.path.exists(extrap_file): - extrapolate_source(input_file, extrap_file, "tf", - logger) + if not netcdf_file_is_valid(extrap_file, "tf", logger=logger): + if os.path.exists(extrap_file): + os.remove(extrap_file) + extrapolate_source(input_file, extrap_file, "tf", logger) logger.info(f" Remapping: {basename}") args = ["ncremap", @@ -308,10 +330,6 @@ def _process_ocean_forcing(self, job, mapping_file, ocean_3d, # Combine remapped files and rename to MALI conventions logger.info("Combining remapped files and renaming variables...") - tf_label = "3dThermalForcing" if ocean_3d else "2dThermalForcing" - output_file = (f"{mali_mesh_name}_{tf_label}_{label}_" - f"{start_year}-{end_year}.nc") - if ocean_3d: self._combine_and_rename_3d(remapped_files, output_file, start_year, end_year) @@ -326,12 +344,9 @@ def _process_ocean_forcing(self, job, mapping_file, ocean_3d, os.remove(f) # Place output in appropriate directory - output_path = os.path.join(output_base_path, forcing_group, - "ocean_thermal_forcing") if not os.path.exists(output_path): os.makedirs(output_path) - dst = os.path.join(output_path, output_file) shutil.copy(output_file, dst) logger.info(f"Done. Output: {dst}") @@ -364,6 +379,17 @@ def _run_climatology(self): logger.info(f"Processing ocean TF climatology: " f"{os.path.basename(input_file)}") + # Check if final output already exists; skip if so + version = os.path.basename(os.path.dirname(input_file)) + output_file = (f"{mali_mesh_name}_thermal_forcing_climatology_" + f"{version}.nc") + output_path = os.path.join(output_base_path, + "ocean_thermal_forcing", "climatology") + dst = os.path.join(output_path, output_file) + if os.path.exists(dst): + logger.info(f"Output already exists, skipping: {dst}") + return + # The mapping file is supplied by the build_mapping_file step. mapping_file = mapping_file_name( config, "ocean", "climatology", method_remap) @@ -372,11 +398,15 @@ def _run_climatology(self): basename = os.path.basename(input_file) remapped_file = f"remapped_{basename}" - if not os.path.exists(remapped_file): + if not netcdf_file_is_valid(remapped_file, "tf", + require_time=False, logger=logger): + if os.path.exists(remapped_file): + os.remove(remapped_file) extrap_file = f"extrap_{basename}" - if not os.path.exists(extrap_file): - extrapolate_source(input_file, extrap_file, "tf", - logger) + if not netcdf_file_is_valid(extrap_file, "tf", logger=logger): + if os.path.exists(extrap_file): + os.remove(extrap_file) + extrapolate_source(input_file, extrap_file, "tf", logger) logger.info(f" Remapping: {basename}") args = ["ncremap", @@ -392,10 +422,6 @@ def _run_climatology(self): # Rename to MALI conventions logger.info("Renaming variables to MALI conventions...") - version = os.path.basename(os.path.dirname(input_file)) - output_file = (f"{mali_mesh_name}_thermal_forcing_climatology_" - f"{version}.nc") - self._rename_climatology_3d(remapped_file, output_file) # Clean up remapped file @@ -403,12 +429,9 @@ def _run_climatology(self): os.remove(remapped_file) # Place output in appropriate directory - output_path = os.path.join(output_base_path, "ocean_thermal_forcing", - "climatology") if not os.path.exists(output_path): os.makedirs(output_path) - dst = os.path.join(output_path, output_file) shutil.copy(output_file, dst) logger.info(f"Done. Output: {dst}") diff --git a/docs/developers_guide/landice/api.rst b/docs/developers_guide/landice/api.rst index 9e362e97e7..71329383e9 100644 --- a/docs/developers_guide/landice/api.rst +++ b/docs/developers_guide/landice/api.rst @@ -34,6 +34,7 @@ ISMIP7 framework archive.resolve_version_directory archive.mapping_file_name mapping.build_mapping_file + remap.netcdf_file_is_valid remap.extrapolate_source remap.open_rename_and_trim remap.add_xtime_and_write