from __future__ import annotations
from pathlib import Path
[docs]
def to_output(path: Path) -> Path:
"""Return the output-file counterpart of an input file path.
Appends ``_out`` before the suffix: ``foo.nc`` → ``foo_out.nc``.
Args:
path: input file path.
Returns:
Path: derived output file path.
"""
return path.parent.joinpath(f"{path.stem}_out{path.suffix}")
[docs]
def remove_file(runsubdir: Path, filename: str, input_only: bool = True) -> None:
"""Remove a file (and optionally its ``_out`` counterpart) from *runsubdir*.
Args:
runsubdir: period run directory.
filename: file name relative to *runsubdir*.
input_only: if ``False``, also delete the ``_out`` variant.
"""
path = runsubdir.joinpath(filename)
path.unlink(missing_ok=True)
if not input_only:
to_output(path).unlink(missing_ok=True)
[docs]
def remove_spec_files(
runsubdir: Path,
filename: str,
species: list[str],
input_only: bool = True,
) -> None:
"""Remove a file and its per-species variants from *runsubdir*.
Deletes *filename* and ``{stem}_{spec}{suffix}`` for each species.
When ``input_only=False``, also deletes the corresponding ``_out``
output files.
Args:
runsubdir: period run directory.
filename: base file name.
species: list of species suffixes to also remove.
input_only: if ``False``, also delete output-file counterparts.
"""
path = runsubdir.joinpath(filename)
path.unlink(missing_ok=True)
if not input_only:
to_output(path).unlink(missing_ok=True)
for spec in species:
spec_path = runsubdir.joinpath(f"{path.stem}_{spec}{path.suffix}")
spec_path.unlink(missing_ok=True)
[docs]
def flush(self, runsubdir: Path, input_only: bool = True) -> None:
"""Clean a simulation sub directory
Parameters
----------
runsubdir : Path
Path to the simulation sub directory
input_only : bool, optional
Only remove input files and not output files, by default True
"""
runsubdir = Path(runsubdir)
chem = self.chemistry
# Auxiliaries files
remove_file(runsubdir, "run.sh")
remove_file(runsubdir, "dispersion.e")
remove_file(runsubdir, "parameters.nml")
remove_file(runsubdir, "chemical_scheme.nml")
# Domain files
remove_file(runsubdir, "domain.nc")
remove_file(runsubdir, "mesh.nc")
# Mass fluxes files
remove_file(runsubdir, "fluxstoke.nc")
remove_file(runsubdir, "fluxstokev.nc")
remove_file(runsubdir, "phystoke.nc")
# Main input/output files
remove_file(runsubdir, "obs.nc", input_only)
remove_spec_files(runsubdir, "start.nc", chem.active_species)
remove_spec_files(runsubdir, "start_tl.nc", chem.active_species)
remove_spec_files(runsubdir, "trajq.bin", chem.active_species)
remove_spec_files(runsubdir, "flux.nc", chem.emitted_species)
# Chemistry files
if self.do_chemistry:
remove_file(runsubdir, "kinetic.nc")
remove_spec_files(
runsubdir, "prescrconcs.nc", chem.prescribed_species, input_only
)
remove_spec_files(runsubdir, "prodloss.nc", chem.prodloss_species, input_only)
remove_spec_files(runsubdir, "deposition.nc", chem.deposition_species)
[docs]
def flushrun(self, rundir, mode, transform_id, full_flush=True):
"""Cleaning the simulation directories to limit space usage"""
# Run subdirectories
for date in self.subsimu_dates:
runsubdir = Path(rundir, date.strftime("%Y-%m-%d_%H-%M"))
flush(self, runsubdir, input_only=False)
# "chain" directory
if mode == "adj":
chain_dir = Path(rundir, "chain")
remove_spec_files(chain_dir, "trajq.bin", self.chemistry.active_species)
if full_flush:
remove_spec_files(chain_dir, "restart.nc", self.chemistry.active_species)
remove_spec_files(chain_dir, "restart.nc", self.chemistry.active_species)
