jmri_core.persistence

 59@dataclass(slots=True)
 60class Profile:
 61    """One JMRI railroad profile.
 62
 63    Lives at ``<root>/<name>`` (e.g. ``~/JMRI/DRGW.jmri``). The bare name
 64    is whatever appears in ``profiles.xml`` — JMRI conventionally
 65    suffixes it with ``.jmri`` but we don't enforce that.
 66    """
 67
 68    root: pathlib.Path
 69    name: str = "default"
 70
 71    @classmethod
 72    def default(cls) -> Profile:
 73        """Construct using the platform-default settings directory."""
 74        return cls(root=default_settings_dir())
 75
 76    # ----- known paths ----------------------------------------------------
 77
 78    @property
 79    def profile_dir(self) -> pathlib.Path:
 80        """``<root>/<name>`` — the profile's root directory."""
 81        return self.root / self.name
 82
 83    @property
 84    def config_file(self) -> pathlib.Path:
 85        """The bean-tables file. Prefers ``PanelPro.xml`` (JMRI 5.x) if
 86        either it exists already; otherwise falls back to the older
 87        ``Configuration.xml``. Brand-new profiles get ``PanelPro.xml``.
 88        """
 89        panel = self.profile_dir / "PanelPro.xml"
 90        legacy = self.profile_dir / "Configuration.xml"
 91        if panel.exists():
 92            return panel
 93        if legacy.exists():
 94            return legacy
 95        return panel
 96
 97    @property
 98    def roster_dir(self) -> pathlib.Path:
 99        """Where per-loco ``locomotive-config`` XML files live."""
100        return self.profile_dir / "roster"
101
102    @property
103    def roster_index(self) -> pathlib.Path:
104        """Roster index — ``<profile>/roster.xml``, alongside the roster dir."""
105        return self.profile_dir / "roster.xml"
106
107    @property
108    def rosters_cv_dir(self) -> pathlib.Path:
109        """Per-loco CV state files (programmer output). Loaded on demand."""
110        return self.profile_dir / "rosters"
111
112    @property
113    def throttle_dir(self) -> pathlib.Path:
114        return self.profile_dir / "throttle"
115
116    @property
117    def signal_dir(self) -> pathlib.Path:
118        """Per-profile signal-mast definitions."""
119        return self.profile_dir / "signal"
120
121    @property
122    def programmers_dir(self) -> pathlib.Path:
123        return self.profile_dir / "programmers"
124
125    @property
126    def jython_dir(self) -> pathlib.Path:
127        """Legacy autoload dir (JMRI Java's Jython 2.7 tree). Kept as a
128        model attribute so imports from an older rewrite install still
129        parse — the autoloader no longer scans it. Use :attr:`python_dir`
130        for CPython 3 scripts."""
131        return self.profile_dir / "jython"
132
133    @property
134    def python_dir(self) -> pathlib.Path:
135        """Per-profile CPython 3 script dir.
136
137        Autoload sequence (see
138        :func:`jmri_server.app._autolaunch_profile_scripts`):
139
140        1. ``<settings-root>/python/`` — cross-profile shared scripts.
141        2. ``<profile>/python/`` — per-profile scripts.
142
143        Files ending in ``.py`` at the top level run in alphabetical
144        order; files starting with ``_`` are treated as private
145        modules that other scripts may import but aren't run standalone.
146        Missing directories are silently ignored — running without
147        either dir is a valid setup for a bare install.
148        """
149        return self.profile_dir / "python"
150
151    @property
152    def shared_python_dir(self) -> pathlib.Path:
153        """Cross-profile CPython 3 script dir at
154        ``<settings-root>/python/``. See :attr:`python_dir` for the
155        autoload sequence."""
156        return self.root / "python"
157
158    @property
159    def operations_dir(self) -> pathlib.Path:
160        return self.profile_dir / "operations"
161
162    @property
163    def panel_dir(self) -> pathlib.Path:
164        return self.profile_dir / "panels"
165
166    @property
167    def logixng_dir(self) -> pathlib.Path:
168        """Where LogixNG programs are stored as YAML files.
169
170        Each program lives at ``<profile>/logixng/<system_name>.yaml``
171        so a profile switch swaps the whole automation set atomically.
172        Directory is created on first save; loads treat a missing dir
173        as "no programs" rather than an error.
174        """
175        return self.profile_dir / "logixng"
176
177    @property
178    def log_dir(self) -> pathlib.Path:
179        return self.root / "log"
180
181    @property
182    def preferences_dir(self) -> pathlib.Path:
183        return self.profile_dir / "profile"
184
185    @property
186    def signal_systems_dir(self) -> pathlib.Path:
187        """User-defined signal systems shared across profiles."""
188        return self.root / "resources" / "signals"
189
190    # ----- helpers ---------------------------------------------------------
191
192    def ensure(self) -> None:
193        """Create every standard per-profile subdirectory if missing.
194
195        Idempotent. Does not touch the settings-root level directories
196        (those are shared across profiles).
197        """
198        for d in (
199            self.root,
200            self.profile_dir,
201            self.roster_dir,
202            self.rosters_cv_dir,
203            self.operations_dir,
204            self.panel_dir,
205            self.throttle_dir,
206            self.signal_dir,
207            self.log_dir,
208            self.signal_systems_dir,
209        ):
210            d.mkdir(parents=True, exist_ok=True)
211
212    def __repr__(self) -> str:
213        return f"Profile(name={self.name!r}, root={str(self.root)!r})"

35def default_settings_dir() -> pathlib.Path:
36    """OS-appropriate default settings directory matching JMRI's Java behavior.
37
38    Honours the ``JMRI_SETTINGS_DIR`` environment variable when set —
39    useful for pointing a Linux host at a Windows-formatted JMRI tree
40    (e.g. an SMB-mounted ``//Pi/JMRI`` or an rsync of the dev machine's
41    ``%USERPROFILE%\\JMRI``). The read goes through
42    :func:`jmri_core.env.settings_dir` so operator-facing env-var
43    names live in one registry.
44    """
45    from jmri_core.env import settings_dir  # noqa: PLC0415 — avoid an import cycle at module load
46    override = settings_dir()
47    if override is not None:
48        return override
49    system = platform.system()
50    if system == "Windows":
51        # JMRI installer creates %userprofile%\JMRI on Windows.
52        return pathlib.Path(os.environ.get("USERPROFILE", "~")).expanduser() / "JMRI"
53    if system == "Darwin":
54        return pathlib.Path("~/Library/Preferences/JMRI").expanduser()
55    # Linux + everything else.
56    return pathlib.Path("~/.jmri").expanduser()

40@dataclass(slots=True)
41class ProfileRef:
42    """One ``<profile>`` entry from ``profiles.xml``."""
43
44    id: str
45    path: str   # raw ``path`` attribute, possibly ``settings:Foo.jmri/``
46    name: str   # derived display name — directory name stripped of suffix
47
48    def to_profile(self, settings_root: pathlib.Path) -> Profile:
49        """Resolve to a runtime :class:`Profile` rooted at *settings_root*."""
50        dir_name = _strip_settings_prefix(self.path).rstrip("/").split("/")[-1]
51        if not dir_name:
52            raise ValueError(f"profile {self.id!r} has empty resolved path")
53        return Profile(root=settings_root, name=dir_name)

 56@dataclass
 57class ProfileRegistry:
 58    """In-memory view of ``<settings>/profiles.xml`` and the active profile."""
 59
 60    settings_root: pathlib.Path
 61    profiles: list[ProfileRef]
 62
 63    # ----- accessors -------------------------------------------------------
 64
 65    def by_id(self, profile_id: str) -> ProfileRef | None:
 66        for p in self.profiles:
 67            if p.id == profile_id:
 68                return p
 69        return None
 70
 71    def by_name(self, name: str) -> ProfileRef | None:
 72        for p in self.profiles:
 73            if p.name == name:
 74                return p
 75        return None
 76
 77    def active(self) -> ProfileRef | None:
 78        """Return the currently-active profile, or ``None`` if unresolved.
 79
 80        Resolution order:
 81        1. ``<settings>/.active_profile`` marker file (our own bookkeeping).
 82        2. First profile in ``profiles.xml`` (JMRI's implicit default).
 83        """
 84        marker = self.settings_root / ACTIVE_MARKER
 85        if marker.exists():
 86            stored = marker.read_text(encoding="utf-8").strip()
 87            hit = self.by_id(stored) or self.by_name(stored)
 88            if hit:
 89                return hit
 90        return self.profiles[0] if self.profiles else None
 91
 92    def set_active(self, profile_id: str) -> None:
 93        """Write the ``.active_profile`` marker. Raises if id is unknown."""
 94        if self.by_id(profile_id) is None:
 95            raise KeyError(f"no profile with id {profile_id!r}")
 96        marker = self.settings_root / ACTIVE_MARKER
 97        marker.parent.mkdir(parents=True, exist_ok=True)
 98        marker.write_text(profile_id, encoding="utf-8")
 99
100    def active_profile(self) -> Profile | None:
101        """Convenience: resolved :class:`Profile` for the active entry."""
102        ref = self.active()
103        return ref.to_profile(self.settings_root) if ref else None

129def load_profile_registry(settings_root: pathlib.Path | None = None) -> ProfileRegistry:
130    """Load ``profiles.xml`` from *settings_root* (default: platform default)."""
131    if settings_root is None:
132        settings_root = default_settings_dir()
133    path = settings_root / PROFILES_XML
134    if not path.exists():
135        return ProfileRegistry(settings_root=settings_root, profiles=[])
136    parser = etree.XMLParser(
137        remove_blank_text=False,
138        resolve_entities=False,
139        load_dtd=False,
140        no_network=True,
141    )
142    root = etree.fromstring(path.read_bytes(), parser=parser)
143    if root.tag != "profileConfig":
144        raise ValueError(
145            f"expected <profileConfig> root in {path}, got <{root.tag}>"
146        )
147    profiles_elem = root.find("profiles")
148    profiles: list[ProfileRef] = []
149    if profiles_elem is not None:
150        for child in profiles_elem.findall("profile"):
151            ref = _parse_profile_elem(child)
152            if ref is not None:
153                profiles.append(ref)
154    return ProfileRegistry(settings_root=settings_root, profiles=profiles)

157def save_profile_registry(registry: ProfileRegistry) -> None:
158    """Write the registry back to ``<settings>/profiles.xml``.
159
160    Preserves a simple shape — we don't model ``<searchPaths>`` because
161    we never need to vary it. If a user's existing ``profiles.xml`` has
162    one, it gets dropped on save; this is acceptable for our alpha
163    since we don't create new profiles from the UI yet.
164    """
165    root = etree.Element("profileConfig")
166    profiles_elem = etree.SubElement(root, "profiles")
167    for ref in registry.profiles:
168        etree.SubElement(profiles_elem, "profile", attrib={
169            "id": ref.id,
170            "path": ref.path,
171        })
172    search = etree.SubElement(root, "searchPaths")
173    etree.SubElement(search, "path", attrib={
174        "path": "settings:",
175        "default": "true",
176    })
177    out = etree.tostring(
178        root,
179        xml_declaration=True,
180        encoding="utf-8",
181        pretty_print=True,
182        standalone=False,
183    )
184    registry.settings_root.mkdir(parents=True, exist_ok=True)
185    (registry.settings_root / PROFILES_XML).write_bytes(out)

19class LocoCvState(BaseModel):
20    """Decoder CV state for one locomotive.
21
22    Keys are CV names (strings, because JMRI uses ``"17.18"`` notation
23    for split CVs and similar non-integer cases). Values are byte
24    integers 0..255.
25    """
26
27    model_config = ConfigDict(populate_by_name=True)
28
29    loco_id: str
30    cv_values: dict[str, int] = Field(default_factory=dict)

116def load_cv_values_from_locomotive_config(
117    path: str | pathlib.Path, *, loco_id: str | None = None,
118) -> LocoCvState | None:
119    """Extract CV state from a ``locomotive-config`` file's ``<values>`` block.
120
121    The JMRI-canonical location for per-loco CV state. Returns ``None`` if
122    the file doesn't exist or has no ``<values>`` block; returns an
123    (empty) :class:`LocoCvState` when the block is present but has no
124    ``<CV>`` rows yet.
125
126    *loco_id* narrows the search when the file lists multiple
127    ``<locomotive>`` entries (rare — JMRI writes one per file). When
128    omitted, the first ``<locomotive>`` wins.
129    """
130    p = pathlib.Path(path)
131    if not p.exists():
132        return None
133    return load_cv_values_from_locomotive_config_bytes(
134        p.read_bytes(), loco_id=loco_id,
135    )

253def load_cv_values_from_locomotive_config_bytes(
254    data: bytes, *, loco_id: str | None = None,
255) -> LocoCvState | None:
256    """Bytes-in variant of :func:`load_cv_values_from_locomotive_config`."""
257    parser = etree.XMLParser(
258        remove_blank_text=False,
259        resolve_entities=False,
260        load_dtd=False,
261        no_network=True,
262    )
263    try:
264        root = etree.fromstring(data, parser=parser)
265    except etree.XMLSyntaxError:
266        return None
267    if root.tag != "locomotive-config":
268        return None
269    for loco in root.iter("locomotive"):
270        this_id = loco.get("id", "")
271        if loco_id is not None and this_id != loco_id:
272            continue
273        values = loco.find("values")
274        if values is None:
275            return None
276        cv_values: dict[str, int] = {}
277        # Accept both element shapes:
278        # * ``<CVvalue name=... value=.../>`` — what real JMRI Java
279        #   writes inside ``<values>``. This is the JMRI-canonical form.
280        # * ``<CV name=... value=.../>`` — what earlier versions of our
281        #   own code wrote. Kept for backwards compat on files we've
282        #   already touched.
283        for tag in ("CVvalue", "CV"):
284            for cv in values.iter(tag):
285                name = cv.get("name")
286                val = cv.get("value")
287                if not name or val is None:
288                    continue
289                try:
290                    cv_values[name] = int(val)
291                except ValueError:
292                    continue
293        return LocoCvState(loco_id=this_id or (loco_id or ""), cv_values=cv_values)
294    return None

65def load_loco_cv(path: str | pathlib.Path) -> LocoCvState:
66    """Load a per-loco CV file. Returns an empty state if missing."""
67    p = pathlib.Path(path)
68    if not p.exists():
69        # The id is the file stem (e.g. ``459.xml`` → ``"459"``).
70        return LocoCvState(loco_id=p.stem)
71    return load_loco_cv_bytes(p.read_bytes())

37def load_loco_cv_bytes(data: bytes) -> LocoCvState:
38    """Parse a ``rosters/<id>.xml`` from bytes."""
39    parser = etree.XMLParser(
40        remove_blank_text=False,
41        resolve_entities=False,
42        load_dtd=False,
43        no_network=True,
44    )
45    root = etree.fromstring(data, parser=parser)
46    if root.tag != "locomotive-cv-values":
47        raise ValueError(
48            f"expected <locomotive-cv-values> root, got <{root.tag}>"
49        )
50    loco_id = root.get("id") or ""
51    cv_values: dict[str, int] = {}
52    for cv in root.findall("CV"):
53        name = cv.get("name")
54        val = cv.get("value")
55        if not name or val is None:
56            continue
57        try:
58            cv_values[name] = int(val)
59        except ValueError:
60            # Skip malformed entries; don't blow up the rest of the file.
61            continue
62    return LocoCvState(loco_id=loco_id, cv_values=cv_values)

138def save_cv_values_into_locomotive_config(
139    path: str | pathlib.Path,
140    cv_values: dict[str, int],
141    *,
142    loco_id: str | None = None,
143) -> None:
144    """Update or insert a ``<values>`` block inside a ``locomotive-config``
145    file, preserving every other element.
146
147    * If *path* doesn't exist, does nothing (caller is expected to have
148      created the roster entry first via the roster service).
149    * If *cv_values* is empty, an existing ``<values>`` block is removed
150      entirely — the file still round-trips through the roster loader.
151    * When the file lists multiple locomotives (rare), *loco_id* picks
152      which one gets the update; otherwise the first is used.
153
154    Writes are atomic (tmp + rename) so a crash mid-write can't corrupt
155    the roster entry.
156    """
157    p = pathlib.Path(path)
158    if not p.exists():
159        return
160    parser = etree.XMLParser(
161        remove_blank_text=False,
162        resolve_entities=False,
163        load_dtd=False,
164        no_network=True,
165    )
166    tree = etree.parse(str(p), parser=parser)
167    root = tree.getroot()
168    if root.tag != "locomotive-config":
169        return
170    target_loco = None
171    for loco in root.iter("locomotive"):
172        if loco_id is None or loco.get("id", "") == loco_id:
173            target_loco = loco
174            break
175    if target_loco is None:
176        return
177
178    # Update in place: preserve any existing ``<values>`` sub-elements
179    # that AREN'T our per-CV entries. In particular JMRI Java writes a
180    # sibling ``<decoderDef>`` block inside ``<values>`` that tracks
181    # per-variable human values; wiping the whole ``<values>`` block
182    # (as an earlier revision did) destroys that data. Same story for
183    # any other unrecognised child element we don't own.
184    values_el = target_loco.find("values")
185    if values_el is None:
186        if not cv_values:
187            return
188        values_el = etree.SubElement(target_loco, "values")
189    else:
190        # Remove existing per-CV entries so we don't stack duplicates.
191        # Accept both element shapes we've written across versions.
192        for entry in list(values_el):
193            if isinstance(entry.tag, str) and entry.tag in ("CVvalue", "CV"):
194                values_el.remove(entry)
195
196    def _key(name: str) -> tuple[int, str]:
197        try:
198            return (0, f"{int(name):05d}")
199        except ValueError:
200            return (1, name)
201
202    if cv_values:
203        # Emit ``<CVvalue>`` — that's the tag name JMRI Java writes and
204        # reads. Interop-safe: a file we write will round-trip through
205        # JMRI Java, and a file JMRI Java wrote (with ``<CVvalue>``)
206        # gets picked up by our reader too.
207        #
208        # lxml's ``pretty_print`` only formats truly-new subtrees; when
209        # we append into an already-parsed element with existing
210        # whitespace nodes it silently packs siblings onto one line.
211        # Explicitly set ``.tail`` on each new entry to a newline +
212        # indent so the file stays hand-editable and diff-friendly.
213        # The last entry gets the closing-tag indent so ``</values>``
214        # lines up under its parent.
215        parent_depth = 0
216        node = values_el
217        while node is not None and node.tag != "locomotive-config":
218            parent_depth += 1
219            node = node.getparent()
220        indent_unit = "  "
221        entry_indent = "\n" + indent_unit * (parent_depth + 1)
222        close_indent = "\n" + indent_unit * parent_depth
223        sorted_names = sorted(cv_values, key=_key)
224        # Make sure the last-existing child's tail (if any) leads into
225        # the first new entry cleanly.
226        if len(values_el) > 0:
227            values_el[-1].tail = entry_indent
228        for i, name in enumerate(sorted_names):
229            el = etree.SubElement(values_el, "CVvalue", attrib={
230                "name": name,
231                "value": str(cv_values[name]),
232            })
233            el.tail = (
234                close_indent if i == len(sorted_names) - 1 else entry_indent
235            )
236    elif len(values_el) == 0:
237        # No per-CV entries left and no other children (decoderDef etc.)
238        # → drop the empty ``<values>`` element rather than leaving
239        # dead XML behind.
240        target_loco.remove(values_el)
241
242    tmp = p.with_suffix(p.suffix + ".tmp")
243    tree.write(
244        str(tmp),
245        xml_declaration=True,
246        encoding="utf-8",
247        pretty_print=True,
248        standalone=False,
249    )
250    tmp.replace(p)

 97def save_loco_cv(path: str | pathlib.Path, state: LocoCvState) -> None:
 98    target = pathlib.Path(path)
 99    target.parent.mkdir(parents=True, exist_ok=True)
100    # Atomic write: tmp + rename so a crash mid-write can't corrupt the file.
101    tmp = target.with_suffix(target.suffix + ".tmp")
102    tmp.write_bytes(save_loco_cv_bytes(state))
103    tmp.replace(target)

74def save_loco_cv_bytes(state: LocoCvState) -> bytes:
75    """Serialize a :class:`LocoCvState` to XML."""
76    root = etree.Element("locomotive-cv-values", attrib={"id": state.loco_id})
77    # Sort by numeric CV when possible, fall back to lexicographic.
78    def _key(name: str) -> tuple[int, str]:
79        try:
80            return (0, f"{int(name):05d}")
81        except ValueError:
82            return (1, name)
83    for name in sorted(state.cv_values, key=_key):
84        etree.SubElement(root, "CV", attrib={
85            "name": name,
86            "value": str(state.cv_values[name]),
87        })
88    return etree.tostring(
89        root,
90        xml_declaration=True,
91        encoding="utf-8",
92        pretty_print=True,
93        standalone=False,
94    )

183@dataclass
184class ConfigFile:
185    """A parsed ``<layout-config>`` PanelPro.xml / Configuration.xml.
186
187    Typed sections (``turnout_sections``, ``sensor_sections``,
188    ``light_sections``) are lists because JMRI emits one element per
189    connection prefix. Other top-level elements (signal heads, memories,
190    routes, blocks, panels, LogixNG, etc.) live in ``preserved`` and are
191    written back verbatim.
192    """
193
194    root_attribs: dict[str, str] = field(default_factory=dict)
195    doctype: str | None = None
196    turnout_sections: list[TurnoutSection] = field(default_factory=list)
197    sensor_sections: list[SensorSection] = field(default_factory=list)
198    light_sections: list[LightSection] = field(default_factory=list)
199    # Other top-level elements we don't model, kept verbatim:
200    preserved: list[etree._Element] = field(default_factory=list)
201
202    # ----- accessors -------------------------------------------------------
203
204    def all_turnouts(self) -> Iterable[TurnoutRow]:
205        for section in self.turnout_sections:
206            yield from section.rows
207
208    def all_sensors(self) -> Iterable[SensorRow]:
209        for section in self.sensor_sections:
210            yield from section.rows
211
212    def all_lights(self) -> Iterable[LightRow]:
213        for section in self.light_sections:
214            yield from section.rows