from __future__ import annotations
import shutil
import subprocess
from logging import debug, error, info
from os import PathLike
from typing import TextIO
[docs]
def run_command(
*args: str | PathLike,
cwd: str | PathLike | None = None,
logfile: TextIO | None = None,
) -> None:
stdout = subprocess.PIPE if logfile is None else logfile
stderr = subprocess.PIPE
args_str = list(map(str, args))
command = " ".join(args_str)
debug(f"Running command: '{command}'")
with subprocess.Popen(args_str, cwd=cwd, stdout=stdout, stderr=stderr) as process:
_, stderr = process.communicate()
stderr = stderr.decode('utf-8')
if logfile is not None:
logfile.write(stderr)
error_message = f"Error while running command '{command}':\n{stderr}"
if stderr.strip():
error(error_message)
if process.returncode != 0:
raise RuntimeError(error_message)
[docs]
def get_make_command(self) -> list[str]:
args = [
"make",
"-j",
f"MODE={getattr(self, 'compile-mode')}",
f"FC={self.compiler}",
f"TARGET={getattr(self, 'compile-acc-target')}",
f"GRID={self.grid}",
f"PHYSICS={self.physics}",
]
if self.grid == "regular":
args.append(f"NLON={self.domain.nlon}")
args.append(f"NLAT={self.domain.nlat}")
elif self.grid == "dynamico":
args.append(f"NBP={self.domain.get_nbp()}")
else:
raise ValueError(f"Unknown grid type: '{self.grid}'")
args.append(f"NLEV={self.domain.nlev}")
if hasattr(self, "netcdf_inc"):
args.append(f"NETCDF_INC={self.netcdf_inc}")
if hasattr(self, "netcdf_lib"):
args.append(f"NETCDF_LIB={self.netcdf_lib}")
return args
# pylint: disable=redefined-builtin
[docs]
def compile(self) -> None:
target_executable = self.model_dir / "dispersion.e"
if not getattr(self, "force-recompile"):
if hasattr(self, "executable"):
shutil.copy(self.executable, target_executable)
with open(self.model_dir / "README.txt", "w") as f:
f.write(f"'dispersion.e' was copied from '{self.executable}'\n")
return
elif not getattr(self, "auto-recompile"):
raise RuntimeError(
"The input argumentt 'auto-recompile' is False while 'executable' "
"was not provided. Specify the 'executable' input argument or "
"'source_dir' while setting 'auto-recompile' to True."
)
if not self.source_dir.is_dir():
raise NotADirectoryError(
f"LMDZ sources directory '{self.source_dir}' not found. "
"Please provide a valid path to LMDZ sources with the 'source_dir' "
"input argument."
)
# Otherwise, re-compile
run_command("rsync", "-ar", self.source_dir / "src", self.model_dir)
run_command("rsync", "-a", self.source_dir / "makefile", self.model_dir)
make_comand = get_make_command(self)
if getattr(self, "compile-clean"):
run_command(*make_comand, "clean", cwd=self.model_dir)
if hasattr(self, "modules"):
# Writting a script that load the modules and compile LMDZ
compile_script = self.model_dir / "compile.sh"
lines = [
"#!/usr/bin/bash",
"",
"module purge",
f"module load {' '.join(self.modules)}",
"module list",
"",
' '.join(make_comand),
"",
]
with open(compile_script, "w") as f:
f.write("\n".join(lines))
args = ["bash", compile_script]
else:
# Directly calling make with the same environment as PyCIF
args = make_comand
# Now compile LMDZ
info("Compiling LMDZ")
try:
logfile = self.model_dir / "compilation.log"
with open(logfile, "w") as f:
run_command(*args, cwd=self.model_dir, logfile=f)
except RuntimeError as e:
raise RuntimeError("Error while compiling LMDZ") from e
if not target_executable.is_file():
raise FileNotFoundError("LMDZ compilation did not produce an executable")
