jmri_core.env

  1"""``jmri_core.env`` — one place to parse ``JMRI_*`` environment variables.
  2
  3Every operator-facing knob is a ``JMRI_``-prefixed env var. Historically
  4these were read directly from ``os.environ`` at each call site — which
  5made it easy to typo names, forget to strip whitespace, or add a knob
  6that never got mentioned in the docs. This module funnels every read
  7through one typed getter, so:
  8
  9* the full name registry lives in one file (search for ``JMRI_`` here
 10  to see every knob);
 11* callers can't confuse ``JMRI_HTTP_PORT`` with ``JMRI_HTTP_PORT_``;
 12* defaults are documented alongside the getter;
 13* tests can monkeypatch a single ``os.environ`` dict.
 14
 15Naming convention: getters are named after the concept, not the env
 16var. So :func:`http_port` reads ``JMRI_HTTP_PORT`` — the operator-facing
 17name is stable, the Python API stays readable.
 18"""
 19from __future__ import annotations
 20
 21import json
 22import logging
 23import os
 24import pathlib
 25from typing import Any
 26
 27_log = logging.getLogger(__name__)
 28
 29# ---- registry ------------------------------------------------------------
 30#
 31# Every env var this project reads. Documented here so operators have a
 32# single place to look. Order matches importance / freqency of use.
 33
 34ENV_CONNECTION = "JMRI_CONNECTION"
 35ENV_CONNECTION_SETTINGS = "JMRI_CONNECTION_SETTINGS"
 36ENV_SETTINGS_DIR = "JMRI_SETTINGS_DIR"
 37ENV_HTTP_PORT = "JMRI_HTTP_PORT"
 38ENV_WITHROTTLE_PORT = "JMRI_WITHROTTLE_PORT"
 39ENV_WITHROTTLE_HOST = "JMRI_WITHROTTLE_HOST"
 40
 41ENV_LOG_LEVEL = "JMRI_LOG_LEVEL"
 42ENV_LOG_FILE = "JMRI_LOG_FILE"
 43ENV_LOG_MAX_MB = "JMRI_LOG_MAX_MB"
 44ENV_LOG_BACKUPS = "JMRI_LOG_BACKUPS"
 45
 46ENV_NCE_DEBUG = "JMRI_NCE_DEBUG"
 47ENV_TIME_HOME = "JMRI_TIME_HOME"
 48
 49ENV_SIGNAL_XML_ROOT = "JMRI_SIGNAL_XML_ROOT"
 50ENV_DECODER_XML_ROOT = "JMRI_DECODER_XML_ROOT"
 51
 52
 53def all_names() -> list[str]:
 54    """List of every env var name this project reads. Useful for docs
 55    generation and for the ``/logs`` endpoint's "configured knobs"
 56    diagnostic block."""
 57    return [
 58        ENV_CONNECTION,
 59        ENV_CONNECTION_SETTINGS,
 60        ENV_SETTINGS_DIR,
 61        ENV_HTTP_PORT,
 62        ENV_WITHROTTLE_PORT,
 63        ENV_WITHROTTLE_HOST,
 64        ENV_LOG_LEVEL,
 65        ENV_LOG_FILE,
 66        ENV_LOG_MAX_MB,
 67        ENV_LOG_BACKUPS,
 68        ENV_NCE_DEBUG,
 69        ENV_TIME_HOME,
 70        ENV_SIGNAL_XML_ROOT,
 71        ENV_DECODER_XML_ROOT,
 72    ]
 73
 74
 75# ---- primitive parsers ---------------------------------------------------
 76
 77
 78def _get(env: dict[str, str] | None, name: str) -> str | None:
 79    """Read *name* from *env* (default ``os.environ``). Trims whitespace
 80    on both sides — a stray newline in a systemd EnvironmentFile is a
 81    frequent-enough footgun that we absorb it here rather than at
 82    every call site."""
 83    source = os.environ if env is None else env
 84    val = source.get(name)
 85    if val is None:
 86        return None
 87    val = val.strip()
 88    if not val:
 89        return None
 90    return val
 91
 92
 93def _get_int(
 94    env: dict[str, str] | None, name: str, default: int,
 95) -> int:
 96    raw = _get(env, name)
 97    if raw is None:
 98        return default
 99    try:
100        return int(raw)
101    except ValueError:
102        _log.warning(
103            "%s=%r isn't an int; falling back to %d", name, raw, default,
104        )
105        return default
106
107
108def _get_bool(env: dict[str, str] | None, name: str) -> bool:
109    """Truthy strings: ``1``, ``true``, ``yes``, ``on`` (case-insensitive).
110    Anything else — including unset — is false. Deliberately narrow so
111    ``JMRI_NCE_DEBUG=0`` reads as off."""
112    raw = _get(env, name)
113    if raw is None:
114        return False
115    return raw.lower() in ("1", "true", "yes", "on")
116
117
118# ---- typed getters -------------------------------------------------------
119
120
121def connection_name(env: dict[str, str] | None = None) -> str | None:
122    """Which hardware driver to start (``nce``, ``loconet``, ...). ``None``
123    means the server picks its default (Internal)."""
124    return _get(env, ENV_CONNECTION)
125
126
127def connection_settings(env: dict[str, str] | None = None) -> dict[str, Any]:
128    """Driver-specific settings blob, JSON-encoded in the env var. Missing
129    or malformed → empty dict, and the parse error is logged so the
130    operator sees why their config didn't take."""
131    raw = _get(env, ENV_CONNECTION_SETTINGS)
132    if raw is None:
133        return {}
134    try:
135        parsed = json.loads(raw)
136    except json.JSONDecodeError as e:
137        _log.warning(
138            "%s isn't valid JSON (%s); ignoring", ENV_CONNECTION_SETTINGS, e,
139        )
140        return {}
141    if not isinstance(parsed, dict):
142        _log.warning(
143            "%s must be a JSON object; got %s",
144            ENV_CONNECTION_SETTINGS, type(parsed).__name__,
145        )
146        return {}
147    return parsed
148
149
150def settings_dir(env: dict[str, str] | None = None) -> pathlib.Path | None:
151    """Override for the JMRI profile-root directory. ``None`` means the
152    OS default (handled by :func:`jmri_core.persistence.profile.default_settings_dir`)."""
153    raw = _get(env, ENV_SETTINGS_DIR)
154    return pathlib.Path(raw).expanduser() if raw else None
155
156
157def http_port(env: dict[str, str] | None = None, default: int = 12080) -> int:
158    return _get_int(env, ENV_HTTP_PORT, default)
159
160
161def withrottle_port(
162    env: dict[str, str] | None = None, default: int = 12090,
163) -> int:
164    return _get_int(env, ENV_WITHROTTLE_PORT, default)
165
166
167def withrottle_host(
168    env: dict[str, str] | None = None, default: str = "0.0.0.0",
169) -> str:
170    return _get(env, ENV_WITHROTTLE_HOST) or default
171
172
173def log_level(
174    env: dict[str, str] | None = None, default: str = "INFO",
175) -> str:
176    """Log level for the root logger (``DEBUG`` / ``INFO`` / ``WARNING`` /
177    ``ERROR``). Case-insensitive; unknown values fall back to the
178    default with a warning logged once the logging is up."""
179    raw = _get(env, ENV_LOG_LEVEL)
180    if raw is None:
181        return default.upper()
182    return raw.upper()
183
184
185def log_file(env: dict[str, str] | None = None) -> pathlib.Path | None:
186    """Path to write logs to. ``None`` means "console only" — matches
187    the pre-Tier-2 behaviour so unattended installs don't suddenly
188    grow files."""
189    raw = _get(env, ENV_LOG_FILE)
190    return pathlib.Path(raw).expanduser() if raw else None
191
192
193def log_max_mb(env: dict[str, str] | None = None, default: int = 5) -> int:
194    """Max size of a single log file before rotation, in MiB."""
195    return max(_get_int(env, ENV_LOG_MAX_MB, default), 1)
196
197
198def log_backups(env: dict[str, str] | None = None, default: int = 3) -> int:
199    """Number of rotated log files to keep."""
200    return max(_get_int(env, ENV_LOG_BACKUPS, default), 0)
201
202
203def nce_debug(env: dict[str, str] | None = None) -> bool:
204    """Verbose NCE frame logging. Off in production; NCE drivers flip
205    their own logger to DEBUG when this is true."""
206    return _get_bool(env, ENV_NCE_DEBUG)
207
208
209def time_home(env: dict[str, str] | None = None) -> bool:
210    """Emit per-phase timing warnings on the home-page render path.
211    Used to diagnose the "profile-load takes 9 s" class of bug. Noisy
212    if left on — hence env-var gated."""
213    return _get_bool(env, ENV_TIME_HOME)
214
215
216def signal_xml_root(env: dict[str, str] | None = None) -> pathlib.Path | None:
217    """Root directory of JMRI's signal-system XML corpus (typically
218    ``<jmri>/xml/signals``). ``None`` means no root is configured —
219    the loader treats that as "no systems available" and the JSON
220    service returns an empty list."""
221    raw = _get(env, ENV_SIGNAL_XML_ROOT)
222    return pathlib.Path(raw).expanduser() if raw else None
223
224
225def decoder_xml_root(env: dict[str, str] | None = None) -> pathlib.Path | None:
226    """Root directory of JMRI's decoder-definition XML corpus
227    (typically ``<jmri>/xml/decoders``). When set, the programmer
228    page pre-populates its "Load catalog from directory" input so
229    the operator only needs to click Load. Unset → same
230    empty-input behaviour as before this knob existed."""
231    raw = _get(env, ENV_DECODER_XML_ROOT)
232    return pathlib.Path(raw).expanduser() if raw else None