"""Where axqua keeps configuration, job state and scratch - platform-aware.
Four separate roles, kept apart because they have different backup, size and
portability characteristics:
``config_dir``
Hand-edited settings that belong to the user: ``profiles.yml``. Small, worth
backing up, never machine-generated in bulk.
``data_dir``
Derived state axqua owns: ``jobs.sqlite``. Reconstructible (see
:mod:`axqua.jobs.index`), so losing it costs a rescan and nothing else.
``state_dir`` / ``cache_dir``
Runtime scraps and throwaway intermediates.
The **job root** is separate from all four and deliberately overridable: CFD output runs
to tens of gigabytes and belongs on a scratch volume, not under ``~``.
No ``platformdirs`` dependency. It would be ~40 lines saved against a new runtime
requirement on a package whose dependency policy (plan §28) is to stay light, and the
QGIS plugin never imports axqua, so there is no second consumer to share it with.
"""
from __future__ import annotations
import logging
import os
import sys
from pathlib import Path
APP_NAME = "axqua"
#: Environment overrides, checked before any platform convention. These exist so a test
#: suite (and a sysadmin) can relocate everything without patching code; the autouse
#: fixture in ``tests/conftest.py`` sets all of them into ``tmp_path``.
ENV_CONFIG = "AXQUA_CONFIG_DIR"
ENV_DATA = "AXQUA_DATA_DIR"
ENV_STATE = "AXQUA_STATE_DIR"
ENV_CACHE = "AXQUA_CACHE_DIR"
ENV_JOB_ROOT = "AXQUA_JOB_ROOT"
#: The pre-rename application name. Its directories are still *read* when the aXqua ones
#: do not exist yet, so an existing job root, index and profiles.yml survive the rename
#: without anyone having to move anything.
LEGACY_APP_NAME = "hydromate"
_LEGACY_ENV_PREFIX = "HYDROMATE_"
_warned_legacy: set[str] = set()
def _env(name: str) -> Path | None:
"""Read *name*, falling back to its pre-rename spelling once, with a notice.
A shell profile, a cron entry or a systemd unit written before the rename still
exports ``HYDROMATE_JOB_ROOT``. Ignoring it would silently relocate someone's jobs,
which is far worse than accepting it and saying so.
"""
value = os.environ.get(name)
if value:
return Path(value).expanduser()
legacy_name = _LEGACY_ENV_PREFIX + name.split("_", 1)[1]
legacy = os.environ.get(legacy_name)
if not legacy:
return None
if legacy_name not in _warned_legacy:
_warned_legacy.add(legacy_name)
logging.getLogger("axqua").warning(
"%s is the pre-rename name for %s; it still works, but rename it.",
legacy_name, name)
return Path(legacy).expanduser()
def _with_legacy(current: Path) -> Path:
"""Prefer *current*, but use the pre-rename sibling while only that one exists."""
if current.exists() or LEGACY_APP_NAME == APP_NAME:
return current
legacy = current.parent / current.name.replace(APP_NAME, LEGACY_APP_NAME)
return legacy if legacy != current and legacy.is_dir() else current
def _windows_base(var: str, fallback: str) -> Path:
raw = os.environ.get(var)
if raw:
return Path(raw)
return Path.home() / fallback
[docs]
def config_dir() -> Path:
"""User configuration: ``profiles.yml`` lives here."""
override = _env(ENV_CONFIG)
if override:
return override
if os.name == "nt":
return _with_legacy(_windows_base("APPDATA", "AppData/Roaming") / APP_NAME)
if sys.platform == "darwin":
return _with_legacy(Path.home() / "Library" / "Application Support" / APP_NAME)
xdg = os.environ.get("XDG_CONFIG_HOME")
return _with_legacy((Path(xdg) if xdg else Path.home() / ".config") / APP_NAME)
[docs]
def data_dir() -> Path:
"""Derived, reconstructible state: the job index."""
override = _env(ENV_DATA)
if override:
return override
if os.name == "nt":
return _with_legacy(_windows_base("LOCALAPPDATA", "AppData/Local") / APP_NAME)
if sys.platform == "darwin":
return _with_legacy(Path.home() / "Library" / "Application Support" / APP_NAME)
xdg = os.environ.get("XDG_DATA_HOME")
return _with_legacy((Path(xdg) if xdg else Path.home() / ".local" / "share") / APP_NAME)
[docs]
def state_dir() -> Path:
override = _env(ENV_STATE)
if override:
return override
if os.name == "nt" or sys.platform == "darwin":
return data_dir()
xdg = os.environ.get("XDG_STATE_HOME")
return (Path(xdg) if xdg else Path.home() / ".local" / "state") / APP_NAME
[docs]
def cache_dir() -> Path:
override = _env(ENV_CACHE)
if override:
return override
if os.name == "nt":
return _windows_base("LOCALAPPDATA", "AppData/Local") / APP_NAME / "cache"
if sys.platform == "darwin":
return Path.home() / "Library" / "Caches" / APP_NAME
xdg = os.environ.get("XDG_CACHE_HOME")
return (Path(xdg) if xdg else Path.home() / ".cache") / APP_NAME
[docs]
def job_root(*, explicit: str | os.PathLike | None = None,
profile_root: str | os.PathLike | None = None,
config_root: str | os.PathLike | None = None) -> Path:
"""Resolve the job root, first hit wins.
``--job-root`` beats ``$AXQUA_JOB_ROOT`` beats the solver profile's
``working_root`` beats ``jobs.root`` of the profile of this computer beats the case
config's ``project.job_root`` beats ``data_dir()/jobs``. The ordering runs
most-specific to least: an explicit flag is about *this* submission, a profile is
about this machine, and the config is about this case.
The profile of this computer is consulted here and not only by ``submit``, because
``list``, ``status``, ``logs`` and ``cancel`` resolve the root through this function
too. When only ``submit`` knew the folder a profile names, a job was created there
and then not found by any of the commands that follow it.
"""
for candidate in (explicit, _env(ENV_JOB_ROOT), profile_root,
_profile_job_root(), config_root):
if candidate:
return Path(candidate).expanduser().resolve()
return data_dir() / "jobs"
def _profile_job_root() -> str | None:
"""``jobs.root`` of the active ``.axq-profile``; ``None`` without one.
Imported here and not at the top: this module is standard library only, so that a
job verb costs no more than a capability listing, and a computer without a profile
never loads the YAML reader for this.
"""
try:
from axqua.core import profile as profiles
active = profiles.load_active()
except Exception: # noqa: BLE001 - never fatal here
return None
root = getattr(active, "job_root", None) if active is not None else None
return str(root) if root else None
[docs]
def index_path() -> Path:
"""The SQLite job index. Derived - deleting it is safe, ``list --rebuild`` restores."""
return data_dir() / "jobs.sqlite"
[docs]
def profiles_path() -> Path:
"""The solver-profile file. Machine-local by design: it names install paths."""
return config_dir() / "profiles.yml"
[docs]
class JobDir:
"""Every path inside one job directory, in one place.
The per-job directory is **authoritative** (plan §5); the SQLite index is derived
from it. So this class is the schema of a job on disk, and anything that needs a path
asks here rather than joining strings.
"""
def __init__(self, root: str | os.PathLike) -> None:
self.root = Path(root)
def __fspath__(self) -> str:
return str(self.root)
def __repr__(self) -> str:
return f"JobDir({str(self.root)!r})"
def __eq__(self, other: object) -> bool:
return isinstance(other, JobDir) and self.root == other.root
@property
def job_id(self) -> str:
return self.root.name
# -- the five files at the top level -----------------------------------------
@property
def spec(self) -> Path:
"""``job.json`` - the immutable submission record."""
return self.root / "job.json"
@property
def status(self) -> Path:
"""``status.json`` - mutable, replaced atomically, runner-only."""
return self.root / "status.json"
@property
def lock(self) -> Path:
return self.root / "job.lock"
@property
def cancel_request(self) -> Path:
"""Its own file, so the one-writer-per-file rule survives cancellation."""
return self.root / "cancel.request"
@property
def runner_log(self) -> Path:
"""axqua's narration. The solver's own listing is a separate artifact."""
return self.root / "runner.log"
@property
def launch_env(self) -> Path:
"""The captured solver environment, as systemd's ``EnvironmentFile``."""
return self.root / "launch.env"
@property
def answers(self) -> Path:
"""Answers to questions a study asked while detached."""
return self.root / "answers.json"
# -- the directories ---------------------------------------------------------
@property
def input(self) -> Path:
return self.root / "input"
@property
def frozen_config(self) -> Path:
return self.input / "case-config.yml"
@property
def preprocessing(self) -> Path:
return self.input / "preprocessing"
@property
def simulation(self) -> Path:
return self.input / "simulation-cases"
@property
def solver_logs(self) -> Path:
return self.root / "solver"
@property
def calibration(self) -> Path:
return self.root / "hydrobayescal"
@property
def results(self) -> Path:
return self.root / "results"
@property
def results_json(self) -> Path:
return self.results / "results.json"
@property
def qgis(self) -> Path:
"""Where :meth:`SolverBackend.export_qgis_results` writes. Never PyQGIS output."""
return self.results / "qgis"
[docs]
def exists(self) -> bool:
return self.spec.is_file()
[docs]
def create(self) -> "JobDir":
"""Claim the directory. Raises ``FileExistsError`` if the id is taken.
The claim is the collision check - ``mkdir(exist_ok=False)`` is atomic, so two
concurrent submits cannot both win the same id no matter how the tails were
drawn.
"""
self.root.mkdir(parents=True, exist_ok=False)
for sub in (self.input, self.solver_logs, self.results, self.qgis):
sub.mkdir(parents=True, exist_ok=True)
return self