"""Configuration management for the seaview processor.
This module handles loading and managing configuration settings using
Dynaconf. Settings are loaded from multiple locations in order of
increasing priority:
1. Global settings (/etc/seaview/)
2. User settings (~/.config/seaview/)
3. Current directory settings (./)
4. Any ``trm_*.toml`` file in the current directory (e.g. ``trm_omi.toml``),
sorted by filename
5. Environment variable specified file (SEAVIEW_SETTINGS_FILE_FOR_DYNACONF)
Attributes
----------
ENV : str
Default environment name.
USER_DIR : pathlib.Path
Path to user configuration directory.
GLOB_DIR : pathlib.Path
Path to global configuration directory.
CURR_DIR : pathlib.Path
Path to current working directory.
settings : Dynaconf
The Dynaconf settings object with loaded configuration.
"""
import os
import pathlib
import logging
from dynaconf import Dynaconf
USER_DIR = pathlib.Path("~/.config/tracmass").expanduser()
USER_DIR.mkdir(parents=True, exist_ok=True)
GLOB_DIR = pathlib.Path("/etc/tracmass/")
CURR_DIR = pathlib.Path("./").absolute()
# Any ``trm_*.toml`` file sitting in the current directory is picked up
# automatically (e.g. a per-run ``trm_omi.toml`` dropped next to a script),
# sorted by filename so the merge order is deterministic; later files in
# settings_files win on conflicting keys.
TRM_FILES = sorted(CURR_DIR.glob("trm_*.toml"))
settings_files = [
GLOB_DIR / "tracmass.toml",
GLOB_DIR / ".secrets.toml",
USER_DIR / "tracmass.toml",
USER_DIR / ".secrets.toml",
CURR_DIR / "tracmass.toml",
CURR_DIR / ".secrets.toml",
*TRM_FILES,
]
extra_file = os.getenv("SEAVIEW_SETTINGS_FILE_FOR_DYNACONF")
if extra_file:
settings_files.append(pathlib.Path(extra_file).absolute())
settings = Dynaconf(
merge_enabled = False,
envvar_prefix="SEAVIEW",
DEBUG_LEVEL_FOR_DYNACONF='DEBUG',
settings_files=settings_files,
#secrets=[
# "/etc/seaview/.secrets.toml",
# "~/.config/seaview/.secrets.toml",
# "./.seaview.toml",
#],
environments=True,
load_dotenv=True,
)
[docs]
def change_env(new_env):
"""Change the active Dynaconf environment.
Parameters
----------
new_env : str
The environment name to switch to (e.g., 'development', 'production').
"""
settings.setenv(new_env)
settings.reload()
[docs]
def move_vertically() -> bool:
"""Return the ``move_vertically`` flag from settings.
Controls whether particles are advected by the vertical velocity ``w``.
Read from the active ``tracmass.toml`` environment; defaults to ``True``
(particles do move vertically) when the key is absent, matching the
historical ``no_wflux=False`` / ``include_wflux=True`` defaults.
Returns
-------
bool
``True`` if vertical advection is enabled. The integrators map this
to ``no_wflux = not move_vertically()`` and the ECCO2 adapter maps it
to ``include_wflux = move_vertically()`` whenever those arguments are
not passed explicitly.
"""
value = settings.get("move_vertically", True)
return True if value is None else bool(value)
[docs]
def n_particle_batches() -> int:
"""Return the ``n_particle_batches`` factor from settings.
Number of equal-sized particle groups each ``lax.while_loop`` call is
split into by the full-grid integrators
(:func:`tracmass.steady_state`, :func:`tracmass.time_dependent`), to cap
peak per-call GPU scratch memory. Read from the active ``tracmass.toml``
environment; defaults to ``1`` (no batching) when the key is absent or
unparseable. ``time_dependent_patched`` / ``time_dependent_streaming``
accept the argument for API symmetry but do not batch.
Returns
-------
int
Batching factor, clamped to ``>= 1``.
"""
value = settings.get("n_particle_batches", 1)
try:
return max(1, int(value))
except (TypeError, ValueError):
return 1
[docs]
def gcm_name(default: str | None = None) -> str | None:
"""Return the ``gcm.name`` adapter name from the active settings environment.
Reads the ``name`` key of the ``[env.gcm]`` table in ``tracmass.toml``
(e.g. ``"glorys"``, ``"ecco2"``, ``"theoretical"``, ``"fabboni"``). A
bare ``gcm = "<name>"`` string (the pre-``[gcm]``-table form) is still
accepted.
Parameters
----------
default : str or None, optional
Value to return when neither form is set. Default ``None``.
Returns
-------
str or None
The adapter name, or *default* if unset.
"""
gcm = settings.get("gcm", None)
if hasattr(gcm, "get"):
return gcm.get("name", default)
if isinstance(gcm, str):
return gcm
return default
# Canonical ordering of the GCM spatial-subset keys. ``i``/``j`` index the
# source grid's x/y axes as 0-based Python slice endpoints; ``k1``/``k2`` are
# 1-based TRACMASS layer numbers; ``*step`` are optional slice strides.
GCM_SUBSET_KEYS: tuple = (
"i1", "i2", "istep",
"j1", "j2", "jstep",
"k1", "k2", "kstep",
)
[docs]
def gcm_subset() -> dict:
"""Return the GCM spatial subset from the active settings environment.
Reads the ``subset`` key of the ``[env.gcm]`` table in ``tracmass.toml``
and returns it as a plain ``dict`` with every key in
:data:`GCM_SUBSET_KEYS` present. Missing keys — and keys written as
``"@none"`` or ``false`` — come back as ``None`` ("no limit on that
edge").
Returns
-------
dict
``{"i1": .., "i2": .., "istep": .., "j1": .., "j2": .., "jstep": ..,
"k1": .., "k2": .., "kstep": ..}`` — each value an ``int`` or
``None``.
See Also
--------
apply_gcm_subset : push this dict into a GCM adapter's ``settings``.
Examples
--------
>>> settings.setenv("omi") # [omi.gcm] subset = { i1 = 2600, ... }
>>> gcm_subset()["i1"]
2600
"""
gcm = settings.get("gcm", None)
raw = gcm.get("subset", None) if hasattr(gcm, "get") else None
try:
raw = dict(raw) if raw else {}
except (TypeError, ValueError):
raw = {}
out: dict = {}
for key in GCM_SUBSET_KEYS:
val = raw.get(key, None)
if val is None or val is False or val == "" or val == "@none":
out[key] = None
else:
out[key] = int(val)
return out
[docs]
def apply_gcm_subset(module, subset: dict | None = None) -> dict:
"""Copy the settings-driven spatial subset into a GCM adapter.
Updates ``module.settings["subset"]`` in place so that a subsequent
``module.fluxes(...)`` call slices the source grid to the configured
region. Only keys with a non-``None`` value are written, so a
partially-specified ``[env.gcm]`` ``subset`` table in ``tracmass.toml``
composes with any limits set directly on ``module.settings["subset"]``.
Adapters that generate their grid and expose no ``settings["subset"]``
(:mod:`tracmass.gcm.theoretical`, :mod:`tracmass.gcm.fabboni`) are left
untouched.
Parameters
----------
module : module
A ``tracmass.gcm`` adapter (e.g. ``tracmass.gcm.ecco2``).
subset : dict, optional
Subset mapping to apply. Defaults to :func:`gcm_subset` (the
active settings environment).
Returns
-------
dict
The fully-resolved subset (all :data:`GCM_SUBSET_KEYS`, ``int`` or
``None``), whether or not the adapter had a slot to receive it.
Examples
--------
>>> from tracmass import gcm
>>> apply_gcm_subset(gcm.glorys) # honour [env.gcm] subset
>>> gcm.glorys.settings["subset"]["k1"]
1
"""
subset = gcm_subset() if subset is None else subset
target = getattr(module, "settings", None)
if isinstance(target, dict) and isinstance(target.get("subset"), dict):
target["subset"].update(
{key: val for key, val in subset.items() if val is not None}
)
return subset
#print("Loaded files:", settings.loaded_files)
#print("\nSettings:", dict(settings))
def config_log():
logpath = pathlib.Path.home() / ".local/state/seaview/"
logpath.mkdir(parents=True, exist_ok=True)
logger = logging.getLogger("seaview")
logging.basicConfig(format='%(levelname)s:%(message)s', level=logging.DEBUG)
logging.getLogger('copernicusmarine').setLevel("WARNING")
logging.getLogger('copernicusmarine').handlers.clear()
logging.getLogger('copernicusmarine_blank_logger').handlers.clear()
logging.getLogger('copernicusmarine').addHandler(logging.NullHandler())
