jmri_web.routes
Template routes — every human-facing page lives here.
The routes call into the registered JsonService instances directly
rather than going over HTTP, since we're in the same process. htmx-driven
mutations (POST endpoints) update the state and return a small HTML
fragment that replaces just the affected table row or status block.
1"""Template routes — every human-facing page lives here. 2 3The routes call into the registered :class:`JsonService` instances directly 4rather than going over HTTP, since we're in the same process. htmx-driven 5mutations (POST endpoints) update the state and return a small HTML 6fragment that replaces just the affected table row or status block. 7""" 8from __future__ import annotations 9 10import re 11from typing import TYPE_CHECKING, Any 12 13from fastapi import Body, FastAPI, Form, HTTPException, Request, Response, UploadFile 14from fastapi.responses import HTMLResponse 15from jmri_server.envelope import JmriRequest 16from jmri_server.service import JsonServiceError 17 18if TYPE_CHECKING: 19 from fastapi.templating import Jinja2Templates 20 from jmri_server.app import AppHandle 21 22 23# ---------- helpers --------------------------------------------------------- 24 25 26import time as _time # noqa: E402 27_STATIC_BUILD_TAG = str(int(_time.time())) 28 29 30def _topics_with_help(handle: AppHandle) -> list[dict[str, Any]]: 31 """List every JSON topic paired with its :class:`TopicHelp` entry. 32 33 Feeds the accordion under the "Topics available on the JSON API" 34 card on the home tab. Topics with no registered help still show 35 as expandable badges — the panel just says "no help authored yet" 36 — so a newly-added service without a help entry doesn't hide 37 from the operator. 38 """ 39 from jmri_server import topic_help # noqa: PLC0415 40 out: list[dict[str, Any]] = [] 41 for topic in handle.registry.topics(): 42 entry = topic_help.get(topic) 43 out.append({ 44 "topic": topic, 45 "help": entry, 46 }) 47 return out 48 49 50def _base_context(handle: AppHandle, active: str) -> dict[str, Any]: 51 m = handle.ctx.metadata 52 return { 53 "active": active, 54 "railroad": m.railroad, 55 "jmri_version": m.jmri_version, 56 # Cache-busting query string appended to /static/* URLs. Every 57 # server restart bumps it so browsers always fetch the fresh 58 # JS/CSS regardless of their cache policy — otherwise a code 59 # change on our side is invisible until the user does a hard 60 # refresh (which some setups have their own problems with). 61 "static_v": _STATIC_BUILD_TAG, 62 } 63 64 65async def _dispatch( 66 handle: AppHandle, request: Request | None = None, **kwargs: Any, 67) -> Any: 68 req = JmriRequest.model_validate(kwargs) 69 try: 70 # When auth is enabled the middleware populates 71 # ``request.state.role`` and ``request.state.user``; 72 # forward both so the same gating rules apply to 73 # web-driven mutations as to raw JSON calls, and 74 # self-service topics (``configUser``) can attribute 75 # the request to the current user. Without a request 76 # (some internal calls) the registry defaults apply. 77 if request is not None: 78 role = getattr(request.state, "role", None) 79 user = getattr(request.state, "user", None) 80 username = user.username if user is not None else None 81 if role is not None: 82 resp = await handle.registry.dispatch( 83 req, role=role, username=username, 84 ) 85 else: 86 resp = await handle.registry.dispatch( 87 req, username=username, 88 ) 89 else: 90 resp = await handle.registry.dispatch(req) 91 except JsonServiceError as e: 92 raise HTTPException(status_code=e.code, detail=e.message) from None 93 return resp.data 94 95 96# ---------- routes --------------------------------------------------------- 97 98 99def install_template_routes( 100 app: FastAPI, handle: AppHandle, templates: Jinja2Templates 101) -> None: 102 # ---- auth: /login + /logout ---------------------------------------- 103 # These routes only do something meaningful when 104 # ``JMRI_AUTH_ENABLED=1``. With auth off, ``ctx.auth`` is None and 105 # /login shows a friendly "auth is disabled" message rather than 106 # a form that couldn't do anything anyway. 107 108 from jmri_server.auth.middleware import SESSION_COOKIE # noqa: PLC0415 109 110 def _csrf_token(request: Request) -> str: 111 """Get-or-set a CSRF token in the ``JMRICSRF`` cookie. 112 113 Double-submit-cookie CSRF pattern: the same random token is 114 embedded in the form AND in a cookie; on POST the server 115 checks they match. A cross-site attacker can't read the 116 cookie (SameSite=Lax) so can't forge a matching form field. 117 """ 118 existing = request.cookies.get("JMRICSRF") 119 if existing: 120 return existing 121 import secrets # noqa: PLC0415 122 return secrets.token_urlsafe(24) 123 124 def _verify_csrf(request: Request, submitted: str) -> bool: 125 cookie = request.cookies.get("JMRICSRF", "") 126 if not cookie or not submitted: 127 return False 128 import hmac # noqa: PLC0415 129 return hmac.compare_digest(cookie, submitted) 130 131 @app.get("/login", response_class=HTMLResponse) 132 async def login_page( 133 request: Request, next: str = "/", 134 ) -> Any: # type: ignore[valid-type] 135 auth = getattr(handle.ctx, "auth", None) 136 error = None 137 if auth is None: 138 error = "Auth is disabled on this server (JMRI_AUTH_ENABLED is unset). Every request runs as admin." 139 token = _csrf_token(request) 140 response = templates.TemplateResponse( 141 request, "login.html", 142 _base_context(handle, "login") | { 143 "csrf_token": token, 144 "next": next, 145 "error": error, 146 }, 147 ) 148 # SameSite=Lax so cross-site GETs don't leak the token; the 149 # form POST is same-origin so it lines up. 150 response.set_cookie( 151 "JMRICSRF", token, samesite="lax", 152 httponly=False, max_age=3600, 153 ) 154 return response 155 156 @app.post("/login", response_class=HTMLResponse) 157 async def login_submit( 158 request: Request, 159 username: str = Form(...), 160 password: str = Form(...), 161 next: str = Form("/"), 162 csrf: str = Form("", alias="_csrf"), 163 ) -> Any: # type: ignore[valid-type] 164 from fastapi.responses import RedirectResponse # noqa: PLC0415 165 auth = getattr(handle.ctx, "auth", None) 166 if auth is None: 167 # Auth off — no login needed; just bounce to home. 168 return RedirectResponse(url=next or "/", status_code=303) 169 if not _verify_csrf(request, csrf): 170 token = _csrf_token(request) 171 response = templates.TemplateResponse( 172 request, "login.html", 173 _base_context(handle, "login") | { 174 "csrf_token": token, 175 "next": next, 176 "error": "Session expired — try again.", 177 }, 178 status_code=400, 179 ) 180 response.set_cookie( 181 "JMRICSRF", token, samesite="lax", 182 httponly=False, max_age=3600, 183 ) 184 return response 185 result = auth.login_password(username, password) 186 if result is None: 187 token = _csrf_token(request) 188 response = templates.TemplateResponse( 189 request, "login.html", 190 _base_context(handle, "login") | { 191 "csrf_token": token, 192 "next": next, 193 "error": "Invalid username or password.", 194 }, 195 status_code=401, 196 ) 197 response.set_cookie( 198 "JMRICSRF", token, samesite="lax", 199 httponly=False, max_age=3600, 200 ) 201 return response 202 response = RedirectResponse(url=next or "/", status_code=303) 203 cookie_value = auth.mint_session_cookie(result.user) 204 # httponly so JS can't touch it (defends against XSS 205 # session theft). secure=False for now since we run on 206 # plain HTTP LANs; flip to True when we grow HTTPS. 207 response.set_cookie( 208 SESSION_COOKIE, cookie_value, 209 httponly=True, samesite="lax", 210 max_age=7 * 24 * 3600, 211 ) 212 return response 213 214 @app.post("/logout") 215 async def logout(request: Request) -> Any: # type: ignore[valid-type] 216 from fastapi.responses import RedirectResponse # noqa: PLC0415 217 response = RedirectResponse(url="/login", status_code=303) 218 response.delete_cookie(SESSION_COOKIE) 219 return response 220 221 # ---- /users admin + self-service ----------------------------------- 222 223 async def _render_users( 224 request: Request, 225 *, 226 notice: str | None = None, 227 new_token: str | None = None, 228 new_token_user: str | None = None, 229 ) -> Any: 230 """Fetch the user list (as admin, if the caller is admin) and 231 render the /users template. Non-admin callers see an empty 232 list — the template hides the admin sections in that case.""" 233 from jmri_server.auth.models import Role # noqa: PLC0415 234 role = getattr(request.state, "role", Role.ADMIN) 235 is_admin = role is Role.ADMIN 236 users: list[dict[str, Any]] = [] 237 if is_admin: 238 try: 239 users = await _dispatch( 240 handle, request, type="configUser", method="list", 241 ) 242 except HTTPException: 243 users = [] 244 token = _csrf_token(request) 245 response = templates.TemplateResponse( 246 request, "users.html", 247 _base_context(handle, "users") | { 248 "users": users, 249 "is_admin": is_admin, 250 "csrf_token": token, 251 "notice": notice, 252 "new_token": new_token, 253 "new_token_user": new_token_user, 254 }, 255 ) 256 response.set_cookie( 257 "JMRICSRF", token, samesite="lax", 258 httponly=False, max_age=3600, 259 ) 260 return response 261 262 @app.get("/users", response_class=HTMLResponse) 263 async def users_home(request: Request) -> Any: # type: ignore[valid-type] 264 return await _render_users(request) 265 266 @app.post("/users/create", response_class=HTMLResponse) 267 async def users_create( 268 request: Request, 269 username: str = Form(...), 270 password: str = Form(...), 271 role: str = Form("operator"), 272 csrf: str = Form("", alias="_csrf"), 273 ) -> Any: # type: ignore[valid-type] 274 if not _verify_csrf(request, csrf): 275 return await _render_users(request, notice="Session expired — try again.") 276 try: 277 await _dispatch( 278 handle, request, type="configUser", method="put", 279 data={"username": username, "password": password, "role": role}, 280 ) 281 except HTTPException as exc: 282 return await _render_users(request, notice=f"Create failed: {exc.detail}") 283 return await _render_users(request, notice=f"Created user {username}.") 284 285 # ---- self-service routes register FIRST so the /users/{username} 286 # ---- parameterized routes below don't shadow /users/self/* with 287 # ---- username="self". Starlette matches routes in registration 288 # ---- order; the more specific paths must come first. 289 290 @app.post("/users/self/change-password", response_class=HTMLResponse) 291 async def users_self_change_password_early( 292 request: Request, 293 new_password: str = Form(...), 294 csrf: str = Form("", alias="_csrf"), 295 ) -> Any: # type: ignore[valid-type] 296 if not _verify_csrf(request, csrf): 297 return await _render_users(request, notice="Session expired.") 298 user = getattr(request.state, "user", None) 299 if user is None: 300 return await _render_users( 301 request, notice="Not logged in — password change unavailable.", 302 ) 303 try: 304 await _dispatch( 305 handle, request, type="configUser", method="post", 306 data={ 307 "action": "change_password", 308 "username": user.username, 309 "new_password": new_password, 310 }, 311 ) 312 except HTTPException as exc: 313 return await _render_users( 314 request, notice=f"Password change failed: {exc.detail}", 315 ) 316 return await _render_users( 317 request, notice="Password updated.", 318 ) 319 320 @app.post("/users/self/regenerate-token", response_class=HTMLResponse) 321 async def users_self_regen_token_early( 322 request: Request, 323 csrf: str = Form("", alias="_csrf"), 324 ) -> Any: # type: ignore[valid-type] 325 if not _verify_csrf(request, csrf): 326 return await _render_users(request, notice="Session expired.") 327 user = getattr(request.state, "user", None) 328 if user is None: 329 return await _render_users( 330 request, notice="Not logged in — token regenerate unavailable.", 331 ) 332 try: 333 resp = await _dispatch( 334 handle, request, type="configUser", method="post", 335 data={ 336 "action": "regenerate_token", 337 "username": user.username, 338 }, 339 ) 340 except HTTPException as exc: 341 return await _render_users( 342 request, notice=f"Regenerate failed: {exc.detail}", 343 ) 344 return await _render_users( 345 request, 346 new_token=resp["token"], 347 new_token_user=user.username, 348 ) 349 350 @app.post("/users/{username}/delete", response_class=HTMLResponse) 351 async def users_delete( 352 request: Request, 353 username: str, 354 csrf: str = Form("", alias="_csrf"), 355 ) -> Any: # type: ignore[valid-type] 356 if not _verify_csrf(request, csrf): 357 return await _render_users(request, notice="Session expired.") 358 try: 359 await _dispatch( 360 handle, request, type="configUser", method="delete", 361 data={"username": username}, 362 ) 363 except HTTPException as exc: 364 return await _render_users(request, notice=f"Delete failed: {exc.detail}") 365 return await _render_users(request, notice=f"Deleted user {username}.") 366 367 @app.post("/users/{username}/set-role", response_class=HTMLResponse) 368 async def users_set_role( 369 request: Request, 370 username: str, 371 role: str = Form(...), 372 csrf: str = Form("", alias="_csrf"), 373 ) -> Any: # type: ignore[valid-type] 374 if not _verify_csrf(request, csrf): 375 return await _render_users(request, notice="Session expired.") 376 try: 377 await _dispatch( 378 handle, request, type="configUser", method="put", 379 data={"username": username, "role": role}, 380 ) 381 except HTTPException as exc: 382 return await _render_users( 383 request, notice=f"Role change failed: {exc.detail}", 384 ) 385 return await _render_users( 386 request, notice=f"{username} → {role}", 387 ) 388 389 @app.post("/users/{username}/regenerate-token", response_class=HTMLResponse) 390 async def users_regen_token( 391 request: Request, 392 username: str, 393 csrf: str = Form("", alias="_csrf"), 394 ) -> Any: # type: ignore[valid-type] 395 if not _verify_csrf(request, csrf): 396 return await _render_users(request, notice="Session expired.") 397 try: 398 resp = await _dispatch( 399 handle, request, type="configUser", method="post", 400 data={"action": "regenerate_token", "username": username}, 401 ) 402 except HTTPException as exc: 403 return await _render_users( 404 request, notice=f"Regenerate failed: {exc.detail}", 405 ) 406 return await _render_users( 407 request, 408 new_token=resp["token"], 409 new_token_user=username, 410 ) 411 412 # The decoder catalog is used by both the roster page (to gate the 413 # Program button) and the programmer routes further down. Kept on 414 # ``handle.ctx._decoder_catalog`` so tests / other subsystems can 415 # populate the catalog directly. Routes fetch it via ``_decoder_catalog`` 416 # on every request so a runtime swap is honoured. 417 from jmri_decoderpro import DecoderCatalog # noqa: PLC0415 418 419 if not hasattr(handle.ctx, "_decoder_catalog"): 420 handle.ctx._decoder_catalog = DecoderCatalog() # type: ignore[attr-defined] 421 422 def _catalog() -> DecoderCatalog: 423 return handle.ctx._decoder_catalog # type: ignore[attr-defined,no-any-return] 424 425 @app.get("/", response_class=HTMLResponse) 426 async def home(request: Request) -> Any: 427 # Diagnostic timing — turned on with ``JMRI_TIME_HOME=1``. Logs 428 # per-dispatch and total render time at WARNING so it shows up 429 # in ``uvicorn.log`` without cranking log verbosity globally. 430 # Off by default so production users see no extra chatter. 431 import time as _time # noqa: PLC0415 432 from jmri_core.env import time_home as _time_home # noqa: PLC0415 433 _diag = _time_home() 434 _t0 = _time.monotonic() if _diag else 0.0 435 power = await _dispatch(handle, type="power", method="get") 436 _t1 = _time.monotonic() if _diag else 0.0 437 fast_clock = await _dispatch(handle, type="time", method="get") 438 _t2 = _time.monotonic() if _diag else 0.0 439 if _diag: 440 import logging as _logging # noqa: PLC0415 441 _logging.getLogger("jmri_web.timing").warning( 442 "GET / dispatch power=%.0fms time=%.0fms", 443 (_t1 - _t0) * 1000, (_t2 - _t1) * 1000, 444 ) 445 registry = getattr(handle.ctx, "profile_registry", None) 446 profiles: list[dict[str, Any]] = [] 447 active_id: str | None = None 448 settings_root_str: str | None = None 449 if registry is not None: 450 active = registry.active() 451 active_id = active.id if active else None 452 profiles = [ 453 {"id": p.id, "name": p.name, "path": p.path} 454 for p in registry.profiles 455 ] 456 settings_root_str = str(registry.settings_root) 457 _t3 = _time.monotonic() if _diag else 0.0 458 log_tail_ctx = await _log_tail_context(handle) 459 resp = templates.TemplateResponse( 460 request, "home.html", 461 _base_context(handle, "home") | { 462 "power": power, 463 "fast_clock": fast_clock, 464 "connection": handle.ctx.connection, 465 "command_station_on": bool( 466 getattr(handle.ctx.connection, 467 "is_command_station_on", False) 468 ), 469 "topics": _topics_with_help(handle), 470 "profiles": profiles, 471 "active_profile_id": active_id, 472 "settings_root": settings_root_str, 473 } | log_tail_ctx, 474 ) 475 if _diag: 476 _t4 = _time.monotonic() 477 import logging as _logging # noqa: PLC0415 478 _logging.getLogger("jmri_web.timing").warning( 479 "GET / render ctx=%.0fms tmpl=%.0fms TOTAL=%.0fms", 480 (_t3 - _t2) * 1000, (_t4 - _t3) * 1000, 481 (_t4 - _t0) * 1000, 482 ) 483 return resp 484 485 @app.post("/power/set", response_class=HTMLResponse) 486 async def set_power(request: Request, state: int = Form(...)) -> Any: 487 data = await _dispatch(handle, type="power", method="post", data={"state": state}) 488 return templates.TemplateResponse( 489 request, "_power_panel.html", {"power": data}, 490 ) 491 492 @app.post("/profiles/load", response_class=HTMLResponse) 493 async def load_profile(request: Request, profile_id: str = Form(...)) -> Any: 494 """Switch the active profile. Clears in-memory beans, re-points 495 ``ConfigStore`` and the roster store, repopulates from the new 496 profile's ``PanelPro.xml``. Hardware connection is untouched.""" 497 from jmri_server.app import switch_active_profile # noqa: PLC0415 498 # Diagnostic timing — opt in with ``JMRI_TIME_HOME=1``. Logs 499 # per-phase durations at WARNING so they surface in uvicorn.log 500 # without dialing up global log verbosity. Off by default. 501 import logging as _logging # noqa: PLC0415 502 import time as _time # noqa: PLC0415 503 from jmri_core.env import time_home as _time_home # noqa: PLC0415 504 _diag = _time_home() 505 _log = _logging.getLogger("jmri_web.timing") 506 _t0 = _time.monotonic() if _diag else 0.0 507 try: 508 await switch_active_profile(handle, profile_id) 509 except (KeyError, RuntimeError) as e: 510 raise HTTPException(status_code=400, detail=str(e)) from None 511 _t1 = _time.monotonic() if _diag else 0.0 512 # Re-render the home page with the new active profile. 513 power = await _dispatch(handle, type="power", method="get") 514 _t2 = _time.monotonic() if _diag else 0.0 515 fast_clock = await _dispatch(handle, type="time", method="get") 516 _t3 = _time.monotonic() if _diag else 0.0 517 registry = getattr(handle.ctx, "profile_registry", None) 518 active = registry.active() if registry is not None else None 519 if _diag: 520 _log.warning( 521 "POST /profiles/load switch=%.0fms power=%.0fms time=%.0fms", 522 (_t1 - _t0) * 1000, (_t2 - _t1) * 1000, (_t3 - _t2) * 1000, 523 ) 524 return templates.TemplateResponse( 525 request, "home.html", 526 _base_context(handle, "home") | { 527 "power": power, 528 "fast_clock": fast_clock, 529 "connection": handle.ctx.connection, 530 "command_station_on": bool( 531 getattr(handle.ctx.connection, 532 "is_command_station_on", False) 533 ), 534 "topics": _topics_with_help(handle), 535 "profiles": [ 536 {"id": p.id, "name": p.name, "path": p.path} 537 for p in (registry.profiles if registry is not None else []) 538 ], 539 "active_profile_id": active.id if active else None, 540 "load_notice": f"Loaded profile: {active.name}" if active else None, 541 }, 542 ) 543 544 # ----- throttle --------------------------------------------------------- 545 546 @app.get("/throttle", response_class=HTMLResponse) 547 async def throttle( 548 request: Request, 549 address: int | None = None, 550 long: int | None = None, 551 ) -> Any: 552 """Throttle page. Without query args, returns to the most recently 553 acquired throttle (single-user alpha; ``ctx._last_throttle`` stores 554 ``(address, long_address)``). Falls back to address 3 if no history. 555 """ 556 last = getattr(handle.ctx, "_last_throttle", None) 557 if address is None: 558 if isinstance(last, tuple) and len(last) == 2: 559 address, last_long = int(last[0]), bool(last[1]) 560 if long is None: 561 long = 1 if last_long else 0 562 else: 563 address = 3 564 # NMRA default: addresses >= 128 are long unless explicitly overridden. 565 long_address = (address >= 128) if long is None else bool(long) 566 data = await _dispatch( 567 handle, type="throttle", method="put", 568 data={ 569 "address": address, 570 "name": str(address), 571 "isLongAddress": long_address, 572 }, 573 ) 574 # Remember this acquisition so the next visit lands here. 575 handle.ctx._last_throttle = (address, long_address) # type: ignore[attr-defined] 576 # Surface roster entries so the page can build a picker. 577 roster = await _dispatch(handle, type="roster", method="list") 578 return templates.TemplateResponse( 579 request, "throttle.html", 580 _base_context(handle, "throttle") | { 581 "address": address, 582 "long_address": long_address, 583 "throttle": data, 584 "roster": roster, 585 }, 586 ) 587 588 @app.post("/throttle/{address}/state", response_class=HTMLResponse) 589 async def throttle_set_state( 590 request: Request, address: int, 591 speed: float = Form(None), 592 forward: int = Form(None), 593 ) -> Any: 594 payload: dict[str, Any] = {"name": str(address)} 595 if speed is not None: 596 payload["speed"] = speed 597 if forward is not None: 598 payload["forward"] = bool(forward) 599 data = await _dispatch(handle, type="throttle", method="post", data=payload) 600 return templates.TemplateResponse( 601 request, "_throttle_status.html", {"throttle": data}, 602 ) 603 604 @app.post("/throttle/{address}/function/{index}", response_class=HTMLResponse) 605 async def throttle_set_function( 606 request: Request, address: int, index: int, on: int = Form(...), 607 ) -> Any: 608 payload = {"name": str(address), f"F{index}": bool(on)} 609 data = await _dispatch(handle, type="throttle", method="post", data=payload) 610 return templates.TemplateResponse( 611 request, "_throttle_status.html", {"throttle": data}, 612 ) 613 614 @app.post("/throttle/{address}/release", response_class=HTMLResponse) 615 async def throttle_release(request: Request, address: int) -> Any: 616 await _dispatch( 617 handle, type="throttle", method="delete", data={"name": str(address)}, 618 ) 619 return templates.TemplateResponse( 620 request, "_throttle_status.html", 621 {"throttle": {"name": str(address), "released": True}}, 622 ) 623 624 # ----- turnouts --------------------------------------------------------- 625 626 @app.get("/turnouts", response_class=HTMLResponse) 627 async def turnouts(request: Request) -> Any: 628 items = await _dispatch(handle, type="turnout", method="list") 629 return templates.TemplateResponse( 630 request, "turnouts.html", 631 _base_context(handle, "turnouts") | {"turnouts": items}, 632 ) 633 634 @app.post("/turnouts/create", response_class=HTMLResponse) 635 async def create_turnout(request: Request, name: str = Form(...)) -> Any: 636 await _dispatch(handle, type="turnout", method="put", data={"name": name}) 637 items = await _dispatch(handle, type="turnout", method="list") 638 return templates.TemplateResponse( 639 request, "_turnouts_table.html", {"turnouts": items}, 640 ) 641 642 @app.post("/turnouts/{name}/state", response_class=HTMLResponse) 643 async def set_turnout_state(request: Request, name: str, state: int = Form(...)) -> Any: 644 data = await _dispatch( 645 handle, type="turnout", method="post", 646 data={"name": name, "state": state}, 647 ) 648 return templates.TemplateResponse( 649 request, "_turnout_row.html", {"t": data}, 650 ) 651 652 @app.get("/turnouts/{name}/row", response_class=HTMLResponse) 653 async def turnout_row(request: Request, name: str) -> Any: 654 """Read-only row partial. Used by the Cancel button to revert an edit.""" 655 data = await _dispatch( 656 handle, type="turnout", method="get", data={"name": name}, 657 ) 658 return templates.TemplateResponse(request, "_turnout_row.html", {"t": data}) 659 660 @app.get("/turnouts/{name}/edit", response_class=HTMLResponse) 661 async def turnout_edit_row(request: Request, name: str) -> Any: 662 data = await _dispatch( 663 handle, type="turnout", method="get", data={"name": name}, 664 ) 665 return templates.TemplateResponse( 666 request, "_turnout_row_edit.html", {"t": data}, 667 ) 668 669 @app.post("/turnouts/{name}/update", response_class=HTMLResponse) 670 async def turnout_update( 671 request: Request, 672 name: str, 673 userName: str = Form(""), 674 inverted: int = Form(0), 675 ) -> Any: 676 payload: dict[str, Any] = { 677 "name": name, 678 "userName": userName or None, 679 "inverted": bool(inverted), 680 } 681 data = await _dispatch(handle, type="turnout", method="post", data=payload) 682 return templates.TemplateResponse(request, "_turnout_row.html", {"t": data}) 683 684 @app.delete("/turnouts/{name}", response_class=HTMLResponse) 685 async def turnout_delete(request: Request, name: str) -> Any: 686 await _dispatch(handle, type="turnout", method="delete", data={"name": name}) 687 # HTMX swap target is the row; returning empty deletes it from the DOM. 688 return HTMLResponse("") 689 690 # ----- sensors ---------------------------------------------------------- 691 692 @app.get("/sensors", response_class=HTMLResponse) 693 async def sensors(request: Request) -> Any: 694 items = await _dispatch(handle, type="sensor", method="list") 695 return templates.TemplateResponse( 696 request, "sensors.html", 697 _base_context(handle, "sensors") | {"sensors": items}, 698 ) 699 700 @app.post("/sensors/create", response_class=HTMLResponse) 701 async def create_sensor(request: Request, name: str = Form(...)) -> Any: 702 await _dispatch(handle, type="sensor", method="put", data={"name": name}) 703 items = await _dispatch(handle, type="sensor", method="list") 704 return templates.TemplateResponse( 705 request, "_sensors_table.html", {"sensors": items}, 706 ) 707 708 @app.post("/sensors/{name}/state", response_class=HTMLResponse) 709 async def set_sensor_state(request: Request, name: str, state: int = Form(...)) -> Any: 710 """For the internal-connection simulator: force a sensor active/inactive.""" 711 data = await _dispatch( 712 handle, type="sensor", method="post", 713 data={"name": name, "state": state}, 714 ) 715 return templates.TemplateResponse( 716 request, "_sensor_row.html", {"s": data}, 717 ) 718 719 @app.get("/sensors/{name}/row", response_class=HTMLResponse) 720 async def sensor_row(request: Request, name: str) -> Any: 721 data = await _dispatch( 722 handle, type="sensor", method="get", data={"name": name}, 723 ) 724 return templates.TemplateResponse(request, "_sensor_row.html", {"s": data}) 725 726 @app.get("/sensors/{name}/edit", response_class=HTMLResponse) 727 async def sensor_edit_row(request: Request, name: str) -> Any: 728 data = await _dispatch( 729 handle, type="sensor", method="get", data={"name": name}, 730 ) 731 return templates.TemplateResponse( 732 request, "_sensor_row_edit.html", {"s": data}, 733 ) 734 735 @app.post("/sensors/{name}/update", response_class=HTMLResponse) 736 async def sensor_update( 737 request: Request, 738 name: str, 739 userName: str = Form(""), 740 inverted: int = Form(0), 741 ) -> Any: 742 payload: dict[str, Any] = { 743 "name": name, 744 "userName": userName or None, 745 "inverted": bool(inverted), 746 } 747 data = await _dispatch(handle, type="sensor", method="post", data=payload) 748 return templates.TemplateResponse(request, "_sensor_row.html", {"s": data}) 749 750 @app.delete("/sensors/{name}", response_class=HTMLResponse) 751 async def sensor_delete(request: Request, name: str) -> Any: 752 await _dispatch(handle, type="sensor", method="delete", data={"name": name}) 753 return HTMLResponse("") 754 755 # ----- lights ----------------------------------------------------------- 756 757 @app.get("/lights", response_class=HTMLResponse) 758 async def lights(request: Request) -> Any: 759 items = await _dispatch(handle, type="light", method="list") 760 return templates.TemplateResponse( 761 request, "lights.html", 762 _base_context(handle, "lights") | {"lights": items}, 763 ) 764 765 @app.post("/lights/create", response_class=HTMLResponse) 766 async def create_light(request: Request, name: str = Form(...)) -> Any: 767 await _dispatch(handle, type="light", method="put", data={"name": name}) 768 items = await _dispatch(handle, type="light", method="list") 769 return templates.TemplateResponse( 770 request, "_lights_table.html", {"lights": items}, 771 ) 772 773 @app.post("/lights/{name}/state", response_class=HTMLResponse) 774 async def set_light_state(request: Request, name: str, state: int = Form(...)) -> Any: 775 data = await _dispatch( 776 handle, type="light", method="post", 777 data={"name": name, "state": state}, 778 ) 779 return templates.TemplateResponse( 780 request, "_light_row.html", {"l": data}, 781 ) 782 783 @app.get("/lights/{name}/row", response_class=HTMLResponse) 784 async def light_row(request: Request, name: str) -> Any: 785 data = await _dispatch( 786 handle, type="light", method="get", data={"name": name}, 787 ) 788 return templates.TemplateResponse(request, "_light_row.html", {"l": data}) 789 790 @app.get("/lights/{name}/edit", response_class=HTMLResponse) 791 async def light_edit_row(request: Request, name: str) -> Any: 792 data = await _dispatch( 793 handle, type="light", method="get", data={"name": name}, 794 ) 795 return templates.TemplateResponse( 796 request, "_light_row_edit.html", {"l": data}, 797 ) 798 799 @app.post("/lights/{name}/update", response_class=HTMLResponse) 800 async def light_update( 801 request: Request, 802 name: str, 803 userName: str = Form(""), 804 ) -> Any: 805 payload: dict[str, Any] = { 806 "name": name, 807 "userName": userName or None, 808 } 809 data = await _dispatch(handle, type="light", method="post", data=payload) 810 return templates.TemplateResponse(request, "_light_row.html", {"l": data}) 811 812 @app.delete("/lights/{name}", response_class=HTMLResponse) 813 async def light_delete(request: Request, name: str) -> Any: 814 await _dispatch(handle, type="light", method="delete", data={"name": name}) 815 return HTMLResponse("") 816 817 # ----- roster ----------------------------------------------------------- 818 819 async def _roster_view(request: Request, format: str | None = None) -> Any: 820 """Engine Driver fetches the roster as XML at ``/roster?format=xml``; 821 the web UI fetches the same data as HTML. Branch on the query.""" 822 if format == "xml": 823 # Serve the on-disk roster.xml verbatim if it exists; that's 824 # what JMRI's RosterServlet would emit. Engine Driver parses 825 # the ``<roster-config>`` shape directly. 826 profile = getattr(handle.ctx, "_roster_profile", None) 827 if profile is not None and profile.roster_index.exists(): 828 return Response( 829 content=profile.roster_index.read_bytes(), 830 media_type="application/xml", 831 ) 832 # No roster.xml yet — synthesize an empty one. 833 return Response( 834 content=b'<?xml version="1.0" encoding="UTF-8"?>\n' 835 b'<roster-config><roster/></roster-config>\n', 836 media_type="application/xml", 837 ) 838 items = await _dispatch(handle, type="roster", method="list") 839 return templates.TemplateResponse( 840 request, "roster.html", 841 _base_context(handle, "roster") | { 842 "roster": items, 843 "catalog_loaded": len(_catalog()) > 0, 844 # For the decoder-picker dropdown. Empty when catalog 845 # isn't loaded — the template falls back to a hint. 846 "manufacturers": _catalog().manufacturers(), 847 }, 848 ) 849 850 # Both the canonical and trailing-slash forms — Engine Driver issues 851 # ``GET /roster/?format=xml`` and won't follow the FastAPI 307 redirect 852 # that the default routing emits for a slash mismatch. 853 @app.get("/roster", response_class=HTMLResponse) 854 async def roster(request: Request, format: str | None = None) -> Any: 855 return await _roster_view(request, format) 856 857 @app.get("/roster/", response_class=HTMLResponse, include_in_schema=False) 858 async def roster_slashed(request: Request, format: str | None = None) -> Any: 859 return await _roster_view(request, format) 860 861 # ----- Engine Driver / WiThrottle side-channel HTTP endpoints ---------- 862 863 @app.get("/prefs/{app_name}/{file_name}", include_in_schema=False) 864 async def withrottle_prefs(app_name: str, file_name: str) -> Response: 865 """Engine Driver pulls/pushes its own preferences from the server. 866 867 JMRI's WiThrottle server stores these at ``<profile>/preferences/ 868 engine_driver/auto_preferences.ed`` and serves them back. For the 869 alpha we return an empty 200 instead of 404 so Engine Driver 870 doesn't panic — its later "save" will not be persisted, but the 871 connection stays healthy. 872 """ 873 profile = getattr(handle.ctx, "_roster_profile", None) 874 if profile is not None: 875 path = profile.preferences_dir / app_name / file_name 876 if path.exists(): 877 return Response( 878 content=path.read_bytes(), 879 media_type="application/octet-stream", 880 ) 881 return Response(content=b"", media_type="application/octet-stream") 882 883 @app.put("/prefs/{app_name}/{file_name}", include_in_schema=False) 884 async def withrottle_prefs_put( 885 request: Request, app_name: str, file_name: str, 886 ) -> Response: 887 """Accept (and persist) Engine Driver's saved preferences.""" 888 profile = getattr(handle.ctx, "_roster_profile", None) 889 body = await request.body() 890 if profile is not None: 891 target_dir = profile.preferences_dir / app_name 892 target_dir.mkdir(parents=True, exist_ok=True) 893 (target_dir / file_name).write_bytes(body) 894 return Response(status_code=204) 895 896 @app.post("/roster/create", response_class=HTMLResponse) 897 async def roster_create( 898 request: Request, 899 id: str = Form(...), 900 dccAddress: str = Form("3"), 901 longAddress: int = Form(0), 902 decoderFamily: str = Form(""), 903 decoderModel: str = Form(""), 904 roadName: str = Form(""), 905 roadNumber: str = Form(""), 906 mfg: str = Form(""), 907 model: str = Form(""), 908 owner: str = Form(""), 909 maxSpeedPct: int = Form(100), 910 ) -> Any: 911 await _dispatch(handle, type="roster", method="put", data={ 912 "id": id, 913 "dccAddress": dccAddress, 914 "longAddress": bool(longAddress), 915 "decoderFamily": decoderFamily, 916 "decoderModel": decoderModel, 917 "roadName": roadName, 918 "roadNumber": roadNumber, 919 "mfg": mfg, 920 "model": model, 921 "owner": owner, 922 "maxSpeedPct": maxSpeedPct, 923 }) 924 items = await _dispatch(handle, type="roster", method="list") 925 return templates.TemplateResponse( 926 request, "_roster_table.html", 927 {"roster": items, "catalog_loaded": len(_catalog()) > 0}, 928 ) 929 930 @app.get("/roster/{entry_id}/row", response_class=HTMLResponse) 931 async def roster_row(request: Request, entry_id: str) -> Any: 932 data = await _dispatch( 933 handle, type="roster", method="get", data={"id": entry_id}, 934 ) 935 return templates.TemplateResponse( 936 request, "_roster_row.html", 937 {"e": data, "catalog_loaded": len(_catalog()) > 0}, 938 ) 939 940 @app.get("/roster/{entry_id}/edit", response_class=HTMLResponse) 941 async def roster_edit_row(request: Request, entry_id: str) -> Any: 942 data = await _dispatch( 943 handle, type="roster", method="get", data={"id": entry_id}, 944 ) 945 return templates.TemplateResponse( 946 request, "_roster_row_edit.html", {"e": data}, 947 ) 948 949 @app.post("/roster/{entry_id}/update", response_class=HTMLResponse) 950 async def roster_update( 951 request: Request, 952 entry_id: str, 953 dccAddress: str = Form(""), 954 longAddress: int = Form(0), 955 decoderFamily: str = Form(""), 956 decoderModel: str = Form(""), 957 roadName: str = Form(""), 958 roadNumber: str = Form(""), 959 mfg: str = Form(""), 960 model: str = Form(""), 961 owner: str = Form(""), 962 maxSpeedPct: int = Form(100), 963 ) -> Any: 964 data = await _dispatch(handle, type="roster", method="post", data={ 965 "id": entry_id, 966 "dccAddress": dccAddress, 967 "longAddress": bool(longAddress), 968 "decoderFamily": decoderFamily, 969 "decoderModel": decoderModel, 970 "roadName": roadName, 971 "roadNumber": roadNumber, 972 "mfg": mfg, 973 "model": model, 974 "owner": owner, 975 "maxSpeedPct": maxSpeedPct, 976 }) 977 return templates.TemplateResponse( 978 request, "_roster_row.html", 979 {"e": data, "catalog_loaded": len(_catalog()) > 0}, 980 ) 981 982 @app.delete("/roster/{entry_id}", response_class=HTMLResponse) 983 async def roster_delete(request: Request, entry_id: str) -> Any: 984 await _dispatch(handle, type="roster", method="delete", data={"id": entry_id}) 985 return HTMLResponse("") 986 987 # ----- panel viewer ----------------------------------------------------- 988 989 @app.get("/panel", response_class=HTMLResponse) 990 async def panel(request: Request) -> Any: 991 return templates.TemplateResponse( 992 request, "panel.html", 993 _base_context(handle, "panel") | {"panel_summary": None}, 994 ) 995 996 # ----- decoder programmer ----------------------------------------------- 997 998 from jmri_core.persistence.decoder_xml import ( # noqa: PLC0415 999 DecoderVariable, 1000 UnsupportedVariableKind, 1001 VariableKind, 1002 find_composite_current_choice, 1003 variable_apply_to_state, 1004 variable_cv_list, 1005 variable_extract_from_state, 1006 variable_extract_from_store, 1007 variable_is_indexed, 1008 variable_primary_cv, 1009 variable_store_key, 1010 variable_wire_read_plan, 1011 variable_wire_write_plan, 1012 ) 1013 from jmri_decoderpro import ( # noqa: PLC0415 1014 compose_panes, 1015 find_pane_by_slug, 1016 ) 1017 1018 def _programmer_context() -> dict[str, Any]: 1019 """The header/badge state for both the main /programmer page and 1020 the tabbed per-loco view. 1021 1022 ``decoder_xml_default`` pre-fills the catalog-load input from 1023 ``JMRI_DECODER_XML_ROOT`` when set; unset → empty string, 1024 matching the pre-Tier-3 behaviour where the operator typed 1025 the path each time. 1026 1027 ``load_in_progress`` (+ ``loaded`` / ``total`` / ``directory``) 1028 surface when either the ``JMRI_DECODER_XML_ROOT`` autoload or 1029 a manual ``/programmer/catalog/load`` is running. The catalog 1030 fragment renders the thermometer instead of the "Catalog 1031 empty" hint in that state, so a user who lands on 1032 ``/programmer`` mid-autoload doesn't see a stale message 1033 prompting them to re-click Load. 1034 """ 1035 from jmri_core.env import decoder_xml_root # noqa: PLC0415 1036 default_root = decoder_xml_root() 1037 progress = getattr(handle.ctx, "_catalog_load_progress", None) 1038 load_in_progress = bool( 1039 progress and not progress.get("done") and not progress.get("error"), 1040 ) 1041 ctx: dict[str, Any] = { 1042 "catalog_size": len(_catalog()), 1043 "manufacturers": _catalog().manufacturers(), 1044 "programmer_count": len(_catalog().programmers), 1045 "decoder_xml_default": str(default_root) if default_root else "", 1046 "load_in_progress": load_in_progress, 1047 } 1048 if load_in_progress: 1049 ctx["loaded"] = progress.get("loaded", 0) 1050 ctx["total"] = progress.get("total", 0) 1051 ctx["directory"] = progress.get("directory", "") 1052 return ctx 1053 1054 @app.get("/programmer", response_class=HTMLResponse) 1055 async def programmer_home(request: Request) -> Any: 1056 try: 1057 avail = await _dispatch(handle, type="programmer", method="list") 1058 except HTTPException: 1059 avail = [] 1060 return templates.TemplateResponse( 1061 request, "programmer.html", 1062 _base_context(handle, "programmer") 1063 | _programmer_context() 1064 | {"availability": avail}, 1065 ) 1066 1067 # ----- per-locomotive programmer (tabbed, DecoderPro-style) ------------- 1068 1069 def _pick_programmer_config(): 1070 """Return the programmer definition used for tab composition. 1071 1072 Preference (matches JMRI's default): ``Comprehensive`` > any 1073 other loaded programmer > None. 1074 """ 1075 if not _catalog().programmers: 1076 return None 1077 return ( 1078 _catalog().programmers.get("Comprehensive") 1079 or next(iter(_catalog().programmers.values()), None) 1080 ) 1081 1082 def _build_variable_row( 1083 var, cv_bytes: dict[str, int], current_value: int | None, 1084 *, definition=None, 1085 ) -> dict[str, Any]: 1086 """Turn a :class:`DecoderVariable` into the dict the pane template 1087 renders. Enum/composite choices are unpacked to plain dicts so 1088 Jinja can iterate them without knowing pydantic. 1089 1090 For composite variables, we also compute: 1091 1092 * ``current_choice`` — which preset's settings match the 1093 current sub-variable state (None if no match). 1094 * ``composite_choices[i].settings`` — the sub-variable writes 1095 each preset expands into. Templates carry this as 1096 ``data-settings`` JSON so client-side batch collection can 1097 fan a composite pick out into per-sub-variable BatchWrite 1098 entries without a server round-trip per choice. 1099 """ 1100 cv_list = variable_cv_list(var) 1101 cv_display = ".".join(str(c) for c in cv_list) if cv_list else var.cv 1102 primary_cv = variable_primary_cv(var) 1103 # For composite: settings map to sub-variable writes. 1104 composite_current_choice: str | None = None 1105 composite_choices: list[dict[str, Any]] = [] 1106 if var.kind is VariableKind.COMPOSITE and definition is not None: 1107 by_item = {v.item: v for v in definition.variables} 1108 composite_current_choice = find_composite_current_choice( 1109 var, by_item, cv_bytes, 1110 ) 1111 for cc in var.composite_choices: 1112 composite_choices.append({ 1113 "choice": cc.choice, 1114 "settings": [ 1115 {"item": s.label, "value": s.value} 1116 for s in cc.settings 1117 ], 1118 }) 1119 else: 1120 composite_choices = [ 1121 {"choice": c.choice, "settings": []} 1122 for c in var.composite_choices 1123 ] 1124 return { 1125 "item": var.item, 1126 "label": var.label, 1127 "tooltip": var.tooltip, 1128 "cv": primary_cv, 1129 "cv_display": cv_display, 1130 "kind": var.kind.value, 1131 "read_only": var.read_only, 1132 "write_only": var.write_only, 1133 "current_value": current_value, 1134 "current_choice": composite_current_choice, 1135 "wire_max": (1 << var.bit_mask.width) - 1 if var.kind in ( 1136 VariableKind.DECIMAL, VariableKind.HEXADECIMAL, VariableKind.ENUM, 1137 ) else 65535, 1138 "enum_choices": [ 1139 {"value": c.value, "choice": c.choice} 1140 for c in var.enum_choices 1141 ], 1142 "composite_choices": composite_choices, 1143 "constant_value": var.constant_value, 1144 } 1145 1146 def _resolve_cv_bytes( 1147 variables, cv_values: dict, 1148 ) -> dict[str, int]: 1149 """Build a string-keyed ``{store_key: byte}`` snapshot for the loco. 1150 1151 Layers, low → high precedence: 1152 1153 1. Per-variable defaults from the catalog: for each variable 1154 that declares a ``default`` (already-*masked* value), we 1155 apply it via :func:`variable_apply_to_state` so bits from 1156 multiple variables share the same CV byte correctly (e.g. 1157 CV 29's Address Format bit 5, Speed Steps bit 1, Power 1158 Source Conv bit 2 all compose into one default byte). 1159 Indexed variables get their default byte written under the 1160 compound key ``bank.index.cv``. 1161 2. Roster ``<values>`` from ``cv_values`` (already str-keyed) 1162 — a stored byte overrides whatever the default layer 1163 produced. 1164 1165 Result is fed to :func:`variable_extract_from_store` which 1166 looks up plain-int-string keys for non-indexed variables and 1167 compound keys for indexed ones. 1168 """ 1169 result: dict[str, int] = {} 1170 for var in variables: 1171 if var.default is None: 1172 continue 1173 if var.kind in ( 1174 VariableKind.SPEED_TABLE, 1175 VariableKind.COMPOSITE, 1176 VariableKind.SPLIT_DATETIME, 1177 VariableKind.CONSTANT, 1178 ): 1179 # These kinds don't participate in the mask-composition 1180 # model. Speed table has one CV per step (defaults are 1181 # per-step and handled by the speed-table row builder). 1182 # Composite and split-datetime aren't wire-value coded. 1183 # Constants don't live in the byte snapshot at all. 1184 continue 1185 # Fetch the current byte under the variable's store key 1186 # (needed for masked RMW). Then apply the default and put 1187 # the result back under the same key. 1188 if variable_is_indexed(var): 1189 # Indexed: apply per byte, keyed by compound name. 1190 # Multi-byte indexed splits get one key per byte. 1191 assert var.split is not None 1192 base = var.split.primary_cv 1193 for i in range(var.split.byte_count): 1194 key = variable_store_key(var, base + i) 1195 # Best-effort byte-slice of a multi-byte human default. 1196 shift = 8 * (var.split.byte_count - 1 - i) 1197 result[key] = (int(var.default) >> shift) & 0xFF 1198 continue 1199 try: 1200 # Non-indexed: use the legacy int-keyed helper against 1201 # an int-keyed view of ``result`` for backward-compat. 1202 int_view = { 1203 int(k): v for k, v in result.items() if k.isdigit() 1204 } 1205 updates = variable_apply_to_state(var, var.default, int_view) 1206 except Exception: # noqa: BLE001 1207 continue 1208 for cv_int, byte in updates.items(): 1209 result[str(cv_int)] = byte 1210 # Overlay roster/stored values — they win over defaults. 1211 for k, v in cv_values.items(): 1212 try: 1213 result[str(k)] = int(v) 1214 except (TypeError, ValueError): 1215 continue 1216 return result 1217 1218 def _build_speed_table_rows( 1219 var, cv_values: dict, 1220 ) -> list[dict[str, Any]]: 1221 """Expand a ``<speedTable>`` variable into one row per step. 1222 1223 NMRA layout: ``entries`` consecutive CVs starting at the 1224 variable's primary CV (typically CV67 → step 1, CV68 → step 2, 1225 …, CV94 → step 28). Each row is an independent 0..255 decimal 1226 input; dirty tracking + batch writes treat them as separate 1227 writes so the queue can retry a single failed step without 1228 re-sending the whole curve. 1229 """ 1230 base_cv = variable_primary_cv(var) 1231 entries = max(int(var.speed_table_entries or 28), 1) 1232 rows: list[dict[str, Any]] = [] 1233 for step_idx in range(entries): 1234 cv = base_cv + step_idx 1235 current = cv_values.get(str(cv)) 1236 try: 1237 current_int = int(current) if current is not None else None 1238 except (TypeError, ValueError): 1239 current_int = None 1240 # Each step becomes its own pseudo-variable so downstream 1241 # dirty tracking uses a per-step ``item`` id. 1242 step_label = f"Speed Step {step_idx + 1}" 1243 step_item = f"{var.item}[{step_idx + 1}]" 1244 rows.append({ 1245 "item": step_item, 1246 "label": step_label, 1247 "tooltip": var.tooltip, 1248 "cv": cv, 1249 "cv_display": str(cv), 1250 "kind": "decimal", 1251 "read_only": var.read_only, 1252 "write_only": var.write_only, 1253 "current_value": current_int, 1254 "wire_max": 255, 1255 "enum_choices": [], 1256 "composite_choices": [], 1257 "constant_value": None, 1258 }) 1259 return rows 1260 1261 _FNMAP_ITEM_RE = re.compile( 1262 r"^(?P<fn>F[^ ]+) controls output (?P<out>\d+)$", 1263 ) 1264 1265 def _fnmap_fn_sort_key(fn: str) -> tuple[int, int, str]: 1266 """JMRI ordering: STOP, DRIVE, FWD, REV, FL(f), FL(r), 1267 F0..F28, then anything else (sensors S0.., etc.) trailing. 1268 """ 1269 specials = { 1270 "STOP": (0, 0), "DRIVE": (0, 1), 1271 "FWD": (0, 2), "REV": (0, 3), 1272 "FL": (0, 4), "FL(f)": (0, 5), "FL(r)": (0, 6), 1273 "F0F": (0, 7), "F0R": (0, 8), 1274 } 1275 if fn in specials: 1276 band, idx = specials[fn] 1277 return (band, idx, "") 1278 m = re.match(r"^F(\d+)$", fn) 1279 if m: 1280 return (1, int(m.group(1)), "") 1281 m = re.match(r"^S(\d+)$", fn) 1282 if m: 1283 return (3, int(m.group(1)), "") 1284 return (2, 0, fn) 1285 1286 def _layout_tree_has_fnmapping(tree: dict) -> bool: 1287 """Walk *tree* looking for any ``fnmapping`` leaf node.""" 1288 if tree.get("type") == "fnmapping": 1289 return True 1290 for child in tree.get("children", ()): 1291 if _layout_tree_has_fnmapping(child): 1292 return True 1293 return False 1294 1295 def _build_fnmap_matrix( 1296 definition, cv_bytes: dict[str, int], 1297 row_by_item: dict[str, dict[str, Any]], 1298 *, variables=None, 1299 ) -> dict[str, Any]: 1300 """Collect all ``Fx controls output y`` variables into a matrix. 1301 1302 Mirrors JMRI's ``FnMapPanel`` layout: 1303 1304 * Rows are functions in natural order — ``FL(f)``, ``FL(r)``, 1305 ``F0``, ``F1``…``F28``, plus any special names the decoder 1306 uses (STOP/DRIVE/FWD/REV — rare, but supported). 1307 * Columns are physical outputs. Only outputs referenced by 1308 at least one variable become columns; outputs a decoder 1309 doesn't wire are suppressed so wide decoders (15+ outputs) 1310 don't get pointless empty columns. 1311 * Each column carries a two-line header — the numeric 1312 ``Output N`` and the decoder-declared output label (e.g. 1313 "Airhorn", "Bell", "FX3"). The label comes from 1314 ``<output name="N" label="…"/>`` on the model definition; 1315 JMRI's authoring convention is to use ``|`` as a line break 1316 inside the label, so we split on it for the sub-lines. 1317 1318 Returns ``{"fn_labels": […], "columns": [{"num": N, 1319 "label_lines": […]}], "cells": {(fn, out): row}}``. 1320 """ 1321 fn_labels: list[str] = [] 1322 outputs: set[int] = set() 1323 cells: dict[tuple[str, int], dict[str, Any]] = {} 1324 seen_fn: set[str] = set() 1325 if variables is None: 1326 variables = definition.variables if definition else () 1327 for var in variables: 1328 m = _FNMAP_ITEM_RE.match(var.item) 1329 if not m: 1330 continue 1331 fn = m.group("fn") 1332 try: 1333 out = int(m.group("out")) 1334 except ValueError: 1335 continue 1336 row = row_by_item.get(var.item) 1337 if row is None: 1338 try: 1339 current_int = variable_extract_from_store(var, cv_bytes) 1340 except UnsupportedVariableKind: 1341 current_int = None 1342 row = _build_variable_row( 1343 var, cv_bytes, current_int, definition=definition, 1344 ) 1345 cells[(fn, out)] = row 1346 outputs.add(out) 1347 if fn not in seen_fn: 1348 seen_fn.add(fn) 1349 fn_labels.append(fn) 1350 fn_labels.sort(key=_fnmap_fn_sort_key) 1351 1352 # Column header labels come from the decoder's <output> elements. 1353 # These live either on the model (JMRI's official spec) or on 1354 # the family — SoundTraxx Tsunami2 and Digitrax put them at 1355 # family level so the labels apply to every model. Model wins 1356 # when both define the same output; family fills the gaps. 1357 output_label_by_num: dict[int, str] = {} 1358 if definition is not None: 1359 output_sources = [] 1360 if definition.family is not None: 1361 output_sources.append(definition.family.outputs) 1362 if definition.model is not None: 1363 output_sources.append(definition.model.outputs) 1364 for source in output_sources: 1365 for o in source: 1366 try: 1367 n = int(o.name) 1368 except (TypeError, ValueError): 1369 continue 1370 if o.label: 1371 output_label_by_num[n] = o.label 1372 columns: list[dict[str, Any]] = [] 1373 for out in sorted(outputs): 1374 raw = output_label_by_num.get(out, "") 1375 # ``|`` is JMRI's authoring convention for line breaks in 1376 # output labels ("Airhorn| ", "Dynamic|Brakes"). 1377 lines = [ 1378 seg.strip() for seg in raw.split("|") if seg.strip() 1379 ] if raw else [] 1380 columns.append({"num": out, "label_lines": lines}) 1381 return { 1382 "fn_labels": fn_labels, 1383 "columns": columns, 1384 "cells": cells, 1385 } 1386 1387 def _build_pane_view( 1388 composed_pane, definition, cv_values: dict, 1389 ) -> dict[str, Any]: 1390 """Assemble the render context for one pane's HTML fragment. 1391 1392 Baseline value shown per row = roster ``<values>`` if the CV is 1393 set there, otherwise the decoder catalog's default. This merged 1394 baseline is what dirty tracking compares against; the JS on the 1395 page marks a row dirty when the user's input diverges from it. 1396 1397 SPEED_TABLE variables (``<speedTable entries="N"/>``) expand 1398 into N rows via :func:`_build_speed_table_rows` — one row per 1399 step CV so the batch queue can write each step independently. 1400 """ 1401 if composed_pane.is_synthetic: 1402 return {} 1403 # Filter variables against the current model's productID and 1404 # any ``<qualifier>`` conditions on other variables. 1405 # 1406 # * productID gate: ``<variable include="pid1,pid2,…">`` / 1407 # ``exclude="…"`` limits the variable to specific decoder 1408 # revisions. 1409 # * Qualifier gate: ``<qualifier>`` triples reference another 1410 # variable's current value (``<variableref>X</variableref> 1411 # <relation>ge</relation><value>7</value>``). All qualifiers 1412 # on a variable must pass (AND) for it to be visible. 1413 # 1414 # SoundTraxx Tsunami2 uses both heavily: HEP Mode / HEP Mode2 / 1415 # Turbine Start-Stop share CV 16.1.303 and take turns being 1416 # applicable depending on the loco model (productID) and prime 1417 # mover selection (``Sound Option 5``). Without the two gates 1418 # all three end up piled into the same grid cell. 1419 current_pid = ( 1420 definition.model.product_id 1421 if definition is not None and definition.model is not None 1422 else "" 1423 ) 1424 def _pid_applies(var) -> bool: 1425 if var.include_product_ids and current_pid not in var.include_product_ids: 1426 return False 1427 if var.exclude_product_ids and current_pid in var.exclude_product_ids: 1428 return False 1429 return True 1430 pid_vars = tuple( 1431 v for v in (definition.variables if definition else ()) 1432 if _pid_applies(v) 1433 ) 1434 # Build a preliminary by-item map from productID-applicable 1435 # variables and a preliminary cv_bytes so qualifiers can look 1436 # up their referenced variables' current values. 1437 pid_by_item = {v.item: v for v in pid_vars} 1438 pid_cv_bytes = _resolve_cv_bytes(pid_vars, cv_values) 1439 def _qualifiers_apply(var) -> bool: 1440 for q in var.qualifiers: 1441 ref = pid_by_item.get(q["variableref"]) 1442 if ref is None: 1443 # Referenced variable not present after productID 1444 # filtering — JMRI treats a missing ref as "gate 1445 # doesn't apply", so let the qualifier pass. Being 1446 # permissive here mirrors JMRI's tolerant behaviour 1447 # for decoder files that share qualifier snippets 1448 # across models. 1449 continue 1450 try: 1451 current = variable_extract_from_store(ref, pid_cv_bytes) 1452 except UnsupportedVariableKind: 1453 continue 1454 if current is None: 1455 continue 1456 rel = q["relation"] 1457 if rel == "exists": 1458 continue 1459 try: 1460 target = int(q["value"]) 1461 except (TypeError, ValueError): 1462 continue 1463 if rel == "eq" and current != target: return False 1464 if rel == "ne" and current == target: return False 1465 if rel == "gt" and current <= target: return False 1466 if rel == "ge" and current < target: return False 1467 if rel == "lt" and current >= target: return False 1468 if rel == "le" and current > target: return False 1469 return True 1470 filtered_vars = tuple(v for v in pid_vars if _qualifiers_apply(v)) 1471 by_item = {v.item: v for v in filtered_vars} 1472 rows: list[dict[str, Any]] = [] 1473 # ``<speedtable/>`` widget: JMRI's Speed Table pane uses this 1474 # instead of per-step <display> items. Find the decoder's 1475 # SPEED_TABLE variable and expand it here (before iterating 1476 # items_available) so the pane renders even when no explicit 1477 # <display item="Speed Table"/> reference exists. Track which 1478 # items we've expanded to avoid double-rendering if the pane 1479 # also lists them via <display>. 1480 expanded_items: set[str] = set() 1481 pane_decl = composed_pane.pane 1482 # Two triggers for speed-table expansion: 1483 # (a) ``<speedtable/>`` widget detected by the parser, OR 1484 # (b) the composed pane's name/slug identifies it as the Speed 1485 # Table tab — a name-based fallback so a real JMRI install 1486 # that ships an unexpected widget variant still expands 1487 # the decoder's SPEED_TABLE variable when the user opens 1488 # the "Speed Table" tab. 1489 pane_is_speed_table = ( 1490 (pane_decl is not None and pane_decl.has_speed_table) 1491 or composed_pane.slug == "speed-table" 1492 or composed_pane.name.strip().lower() == "speed table" 1493 ) 1494 if pane_is_speed_table: 1495 found_speed_table_var = False 1496 for v in (definition.variables if definition else ()): 1497 if v.kind == VariableKind.SPEED_TABLE: 1498 rows.extend(_build_speed_table_rows(v, cv_values)) 1499 expanded_items.add(v.item) 1500 found_speed_table_var = True 1501 if not found_speed_table_var: 1502 # No SPEED_TABLE variable in the catalog — many decoder 1503 # XMLs omit it, relying on the NMRA convention that CV 1504 # 67..CV 94 are the 28-step table (bytes 1..28). Emit a 1505 # synthetic 28-row expansion so the user can still write 1506 # each step. Use the class:`DecoderVariable` model with 1507 # the anchor CV and default entries so 1508 # :func:`_build_speed_table_rows` handles it. 1509 synthetic = DecoderVariable.model_validate({ 1510 "CV": "67", 1511 "item": "Speed Table", 1512 "kind": VariableKind.SPEED_TABLE, 1513 "speed_table_entries": 28, 1514 }) 1515 rows.extend(_build_speed_table_rows(synthetic, cv_values)) 1516 # Build an int-keyed CV byte snapshot, with defaults filling in 1517 # for CVs that haven't been read/written yet. For bit-mask 1518 # variables (Address Format, Speed Steps, Power Source Conv — 1519 # all sharing CV 29), we must call ``variable_extract_from_state`` 1520 # so the row's "current" shows the *masked* value, not the raw 1521 # byte. Otherwise every enum on a shared CV compares its wire 1522 # value (0/1/2/…) against the whole byte (e.g. 2) and dirty- 1523 # tracks a false positive; and switching panes shows the same 1524 # raw byte across all variables, looking wildly inconsistent. 1525 cv_bytes = _resolve_cv_bytes(filtered_vars, cv_values) 1526 for item_name in composed_pane.items_available: 1527 if item_name in expanded_items: 1528 continue 1529 var = by_item.get(item_name) 1530 if var is None: 1531 continue 1532 if var.kind == VariableKind.SPEED_TABLE: 1533 rows.extend(_build_speed_table_rows(var, cv_values)) 1534 continue 1535 try: 1536 current_int = variable_extract_from_store(var, cv_bytes) 1537 except UnsupportedVariableKind: 1538 current_int = None 1539 rows.append(_build_variable_row( 1540 var, cv_bytes, current_int, definition=definition, 1541 )) 1542 # On the Speed Table pane we synthesize CV 67..CV 94 rows before 1543 # iterating items_available, so the natural row order interleaves 1544 # oddly (67..94 first, then whatever items_available lists like 1545 # CV 65 Kickstart / CV 66 Forward Trim). Sort by CV number here 1546 # so the pane reads top-to-bottom in CV order. Only applies to 1547 # the speed-table pane — other panes rely on JMRI's explicit 1548 # display ordering. 1549 if pane_is_speed_table: 1550 def _cv_sort_key(row: dict[str, Any]) -> int: 1551 try: 1552 return int(row.get("cv") or 0) 1553 except (TypeError, ValueError): 1554 return 0 1555 rows.sort(key=_cv_sort_key) 1556 1557 # Raw CV table (CVs pane) — enumerate every CV the decoder 1558 # defines, merged with whatever's been read/written into the 1559 # store. Each row is displayed with its stored value (if any) 1560 # and gives the user a Read/Write handle on it. Matches JMRI 1561 # Java's CVs tab which lists everything up front, not just 1562 # what's been touched. 1563 raw_cv_rows: list[dict[str, Any]] = [] 1564 if pane_decl is not None and pane_decl.has_cvtable: 1565 catalog_names: set[str] = set() 1566 for var in (definition.variables if definition else ()): 1567 if var.kind in ( 1568 VariableKind.CONSTANT, VariableKind.SPEED_TABLE, 1569 VariableKind.COMPOSITE, VariableKind.SPLIT_DATETIME, 1570 ): 1571 continue 1572 if variable_is_indexed(var): 1573 assert var.split is not None 1574 base = var.split.primary_cv 1575 for i in range(var.split.byte_count): 1576 catalog_names.add(variable_store_key(var, base + i)) 1577 else: 1578 for cv in variable_cv_list(var): 1579 catalog_names.add(str(cv)) 1580 # Also include anything in the store that isn't in the 1581 # catalog (e.g. legacy roster entries, raw CVs the user 1582 # added via the "Add CV" input in a prior session). 1583 all_names = catalog_names | set(cv_values.keys()) 1584 def _raw_sort_key(name: str) -> tuple[int, int, int, str]: 1585 # Sort plain CVs first (by number), then indexed by 1586 # (bank, index, base). Falls through to string compare 1587 # for anything that doesn't parse. 1588 parts = name.split(".") 1589 try: 1590 nums = [int(p) for p in parts] 1591 except ValueError: 1592 return (2, 0, 0, name) 1593 if len(nums) == 1: 1594 return (0, nums[0], 0, "") 1595 if len(nums) == 3: 1596 return (1, nums[0], nums[1] * 1000 + nums[2], "") 1597 return (1, 0, 0, name) 1598 for name in sorted(all_names, key=_raw_sort_key): 1599 v = cv_values.get(name) 1600 try: 1601 v_int = int(v) if v is not None else None 1602 except (TypeError, ValueError): 1603 v_int = None 1604 raw_cv_rows.append({ 1605 "name": name, 1606 "value": v_int, 1607 "hex": f"{v_int:02X}" if v_int is not None else "", 1608 }) 1609 # Rich layout tree — the JMRI-authored column/row/label/ 1610 # separator structure. The template renders it recursively via 1611 # the ``layout_node`` macro. Each ``<display>`` leaf resolves 1612 # to a variable row (looked up by item), which the macro emits 1613 # inline with its label + input + Read button. When the pane 1614 # doesn't have a raw XML element (synthetic tests, or older 1615 # panes), the fallback tree in ``pane_layout_tree`` gives a 1616 # flat column of the pane's ``items`` list — same widgets as 1617 # the legacy table view, just wrapped in divs. 1618 from jmri_core.persistence.decoder_xml import pane_layout_tree # noqa: PLC0415 1619 layout_tree = None 1620 row_by_item: dict[str, dict[str, Any]] = {} 1621 if composed_pane.pane is not None: 1622 layout_tree = pane_layout_tree(composed_pane.pane) 1623 row_by_item = {r["item"]: r for r in rows if r.get("item")} 1624 1625 # ----- Function Map matrix ----- 1626 # JMRI's ``<fnmapping/>`` marker is an app-generated grid of 1627 # checkboxes keyed by every ``Fx controls output y`` variable 1628 # the decoder defines. The pane XML doesn't list these 1629 # individually (that's the point of the marker), so we collect 1630 # them here from ``definition.variables`` and shape them into 1631 # a matrix keyed by function label + output number. When the 1632 # pane has no fnmapping marker we skip the whole thing. 1633 fnmap_matrix: dict[str, Any] | None = None 1634 if layout_tree is not None and _layout_tree_has_fnmapping(layout_tree): 1635 fnmap_matrix = _build_fnmap_matrix( 1636 definition, cv_bytes, row_by_item, variables=filtered_vars, 1637 ) 1638 return { 1639 "variable_rows": rows, 1640 "raw_cv_rows": raw_cv_rows, 1641 "layout_tree": layout_tree, 1642 "row_by_item": row_by_item, 1643 "is_speed_table_pane": pane_is_speed_table, 1644 "fnmap_matrix": fnmap_matrix, 1645 } 1646 1647 @app.get("/programmer/loco/{loco_id}", response_class=HTMLResponse) 1648 async def programmer_loco( 1649 request: Request, loco_id: str, tab: str | None = None, 1650 ) -> Any: 1651 """Per-loco tabbed CV programmer. 1652 1653 Roster entry drives the tab set: the loco's decoder family/model 1654 picks a :class:`DecoderDefinition` from the catalog, then 1655 ``compose_panes`` unions the standard programmer panes (from 1656 Comprehensive.xml if loaded) with any decoder-declared extras. 1657 """ 1658 try: 1659 loco = await _dispatch( 1660 handle, type="roster", method="get", data={"id": loco_id}, 1661 ) 1662 except HTTPException as e: 1663 return HTMLResponse( 1664 f"no such roster entry: {loco_id}", status_code=e.status_code, 1665 ) 1666 1667 family = loco.get("decoderFamily") or "" 1668 model = loco.get("decoderModel") or "" 1669 definition = ( 1670 _catalog().find(family, model) if (family and model) else None 1671 ) 1672 programmer_cfg = _pick_programmer_config() 1673 1674 variables = definition.variables if definition else () 1675 decoder_panes = definition.panes if definition else () 1676 tab_panes = compose_panes(list(variables), programmer_cfg, list(decoder_panes)) 1677 1678 # Per-loco CV state. If the store sees CVs in both the JMRI- 1679 # canonical roster/<id>.xml <values> block AND our legacy 1680 # rosters/<id>.xml, ask the user which is authoritative before 1681 # rendering panes with potentially-wrong values. 1682 file_name = loco.get("fileName") or None 1683 from jmri_server.services.programmer import _loco_cv_store # noqa: PLC0415 1684 from jmri_server.loco_cv_store import AmbiguousCvSourceError # noqa: PLC0415 1685 store = _loco_cv_store(handle.ctx) 1686 try: 1687 state = store.get(loco_id, file_name) 1688 except AmbiguousCvSourceError as ambig: 1689 return templates.TemplateResponse( 1690 request, "programmer_cv_source_choose.html", 1691 _base_context(handle, "programmer") 1692 | _programmer_context() 1693 | { 1694 "loco": loco, 1695 "loco_id": loco_id, 1696 "roster_path": str(ambig.roster_path), 1697 "legacy_path": str(ambig.legacy_path), 1698 }, 1699 ) 1700 1701 # Default active tab = Roster Entry. Query-param override lets 1702 # deep links land on a specific tab. 1703 active_slug = tab or (tab_panes[0].slug if tab_panes else "roster-entry") 1704 active_pane = find_pane_by_slug(tab_panes, active_slug) or ( 1705 tab_panes[0] if tab_panes else None 1706 ) 1707 if active_pane is not None: 1708 active_slug = active_pane.slug 1709 1710 pane_ctx = _build_pane_view(active_pane, definition, state.cv_values) if active_pane else {} 1711 1712 return templates.TemplateResponse( 1713 request, "programmer_loco.html", 1714 _base_context(handle, "programmer") 1715 | _programmer_context() 1716 | { 1717 "loco": loco, 1718 "loco_id": loco_id, 1719 "definition": definition, 1720 "family": family, 1721 "model": model, 1722 "cv_values": state.cv_values, 1723 "tab_panes": tab_panes, 1724 "active_slug": active_slug, 1725 "pane": active_pane, 1726 **pane_ctx, 1727 }, 1728 ) 1729 1730 @app.post( 1731 "/programmer/loco/{loco_id}/resolve-source", 1732 response_class=HTMLResponse, 1733 ) 1734 async def programmer_loco_resolve_source( 1735 request: Request, loco_id: str, source: str = Form(...), 1736 ) -> Any: 1737 """Persist the user's pick for which CV-state source is 1738 authoritative on this loco, then bounce back to the programmer 1739 view so the panes render with the chosen source.""" 1740 from jmri_server.loco_cv_store import CvSource # noqa: PLC0415 1741 from jmri_server.services.programmer import _loco_cv_store # noqa: PLC0415 1742 try: 1743 picked = CvSource(source) 1744 except ValueError: 1745 raise HTTPException(400, f"unknown source: {source}") from None 1746 _loco_cv_store(handle.ctx).set_preferred_source(loco_id, picked) 1747 # 303 See Other → re-run the GET so panes render. 1748 return HTMLResponse( 1749 "", status_code=303, 1750 headers={"Location": f"/programmer/loco/{loco_id}"}, 1751 ) 1752 1753 @app.get( 1754 "/programmer/loco/{loco_id}/pane/{pane_slug}", 1755 response_class=HTMLResponse, 1756 ) 1757 async def programmer_loco_pane( 1758 request: Request, loco_id: str, pane_slug: str, 1759 ) -> Any: 1760 """Return one pane's HTML fragment for HTMX tab-swap.""" 1761 try: 1762 loco = await _dispatch( 1763 handle, type="roster", method="get", data={"id": loco_id}, 1764 ) 1765 except HTTPException as e: 1766 return HTMLResponse(str(e.detail), status_code=e.status_code) 1767 1768 family = loco.get("decoderFamily") or "" 1769 model = loco.get("decoderModel") or "" 1770 definition = ( 1771 _catalog().find(family, model) if (family and model) else None 1772 ) 1773 programmer_cfg = _pick_programmer_config() 1774 1775 variables = definition.variables if definition else () 1776 decoder_panes = definition.panes if definition else () 1777 tab_panes = compose_panes(list(variables), programmer_cfg, list(decoder_panes)) 1778 active_pane = find_pane_by_slug(tab_panes, pane_slug) 1779 if active_pane is None: 1780 return HTMLResponse("no such pane", status_code=404) 1781 1782 file_name = loco.get("fileName") or None 1783 from jmri_server.services.programmer import _loco_cv_store # noqa: PLC0415 1784 store = _loco_cv_store(handle.ctx) 1785 state = store.get(loco_id, file_name) 1786 1787 pane_ctx = _build_pane_view(active_pane, definition, state.cv_values) 1788 return templates.TemplateResponse( 1789 request, "_programmer_pane_content.html", 1790 { 1791 "loco": loco, 1792 "loco_id": loco_id, 1793 "family": family, 1794 "model": model, 1795 "active_slug": pane_slug, 1796 "pane": active_pane, 1797 "cv_values": state.cv_values, 1798 **pane_ctx, 1799 }, 1800 ) 1801 1802 @app.post("/programmer/loco/{loco_id}/read") 1803 async def programmer_loco_read( 1804 request: Request, loco_id: str, 1805 payload: dict[str, Any] = Body(...), 1806 ) -> Any: 1807 """JSON-shaped single-CV read. 1808 1809 When the payload includes the row's ``item`` and the catalog 1810 knows a variable for that name, the response also carries the 1811 variable's *extracted* masked value ("humanValue") — the JS 1812 needs that to update the row's ``data-current`` correctly. 1813 Otherwise a masked-bit read (Address Format, Speed Steps, etc.) 1814 would put the raw CV byte into ``data-current`` and every 1815 subsequent dirty-check on that row would compare the byte 1816 against the enum's masked-value option and flip false-dirty. 1817 """ 1818 cv = payload.get("cv") 1819 if not isinstance(cv, int): 1820 raise HTTPException(400, "read payload requires integer 'cv'") 1821 item = str(payload.get("item") or "") 1822 dispatch_payload: dict[str, Any] = { 1823 "cv": cv, 1824 "mode": str(payload.get("mode", "service")), 1825 "locoId": loco_id, 1826 } 1827 file_name = payload.get("fileName") or payload.get("file_name") 1828 if file_name: 1829 dispatch_payload["fileName"] = file_name 1830 address = payload.get("address") 1831 if address is not None: 1832 dispatch_payload["address"] = int(address) 1833 dispatch_payload["isLongAddress"] = bool(payload.get("isLongAddress")) 1834 try: 1835 data = await _dispatch( 1836 handle, type="programmer", method="get", data=dispatch_payload, 1837 ) 1838 except HTTPException as e: 1839 raise HTTPException(e.status_code, e.detail) from None 1840 wire_value = int(data.get("value", 0)) 1841 human_value: int | None = None 1842 var = _find_variable_for_write(loco_id, item) if item else None 1843 if var is not None: 1844 # Snapshot the post-write CV bytes (the read already pushed 1845 # this CV into the store) and extract per-variable so the 1846 # client can set ``data-current`` to the masked value. 1847 from jmri_server.services.programmer import _loco_cv_store # noqa: PLC0415 1848 store = _loco_cv_store(handle.ctx) 1849 state = store.get(loco_id, file_name) 1850 # String-keyed snapshot supports indexed CVs (``16.1.275``) 1851 # coexisting with plain CVs (``275``) — the extract helper 1852 # picks the right key per variable. 1853 cv_bytes = {str(k): int(v) for k, v in state.cv_values.items()} 1854 try: 1855 human_value = variable_extract_from_store(var, cv_bytes) 1856 except UnsupportedVariableKind: 1857 human_value = None 1858 return { 1859 "cv": cv, 1860 "value": wire_value, 1861 "humanValue": human_value if human_value is not None else wire_value, 1862 "mode": str(data.get("mode", "service_direct")), 1863 } 1864 1865 @app.post("/programmer/loco/{loco_id}/write", response_class=HTMLResponse) 1866 async def programmer_loco_write( 1867 request: Request, loco_id: str, 1868 cv: int = Form(...), value: int = Form(...), 1869 mode: str = Form("service"), address: int | None = Form(None), 1870 file_name: str = Form(""), 1871 ) -> Any: 1872 payload: dict[str, Any] = { 1873 "cv": cv, "value": value, "mode": mode, "locoId": loco_id, 1874 } 1875 if file_name: 1876 payload["fileName"] = file_name 1877 if address is not None: 1878 payload["address"] = address 1879 try: 1880 data = await _dispatch(handle, type="programmer", method="post", data=payload) 1881 except HTTPException as e: 1882 return templates.TemplateResponse( 1883 request, "_programmer_cv_result.html", 1884 {"cv": cv, "error": e.detail}, status_code=e.status_code, 1885 ) 1886 return templates.TemplateResponse( 1887 request, "_programmer_cv_result.html", 1888 {"cv": cv, "value": data["value"], "mode": data["mode"], "loco_id": loco_id}, 1889 ) 1890 1891 # ----- raw CV access on the CVs tab --------------------------------- 1892 1893 from jmri_core.persistence.decoder_xml import SplitCv # noqa: PLC0415 1894 1895 def _parse_raw_cv_name(name: str) -> tuple[int, SplitCv | None]: 1896 """Split a raw-CV table entry into ``(wire_cv, split_or_None)``. 1897 1898 ``"275"`` → (275, None) — plain CV, no indexed setup. 1899 ``"16.1.275"`` → (275, SplitCv(bank=16, index=1, primary_cv=275)) 1900 — compound CV, caller must prepend CV31/CV32 setup writes. 1901 1902 Raises :class:`ValueError` on unparseable input so the caller 1903 can 400 out with a clear message. 1904 """ 1905 name = name.strip() 1906 if not name: 1907 raise ValueError("CV name must not be empty") 1908 if "." in name: 1909 split = SplitCv.parse(name) 1910 if split.primary_cv <= 0 or split.bank is None: 1911 raise ValueError(f"malformed indexed CV name: {name!r}") 1912 return split.primary_cv, split 1913 try: 1914 return int(name), None 1915 except ValueError as e: 1916 raise ValueError(f"CV name must be int or bank.index.base, got {name!r}") from e 1917 1918 @app.post("/programmer/loco/{loco_id}/cv/read") 1919 async def programmer_loco_raw_cv_read( 1920 loco_id: str, payload: dict[str, Any] = Body(...), 1921 ) -> Any: 1922 """Read one CV on the CVs tab. Accepts either plain (``275``) 1923 or compound (``16.1.275``) names. 1924 1925 For compound CVs, dispatches CV31 = bank and CV32 = index setup 1926 writes before the read, and stashes the result in the roster 1927 under the compound key so subsequent renders show it. 1928 """ 1929 cv_name = str(payload.get("cvName") or "") 1930 try: 1931 wire_cv, split = _parse_raw_cv_name(cv_name) 1932 except ValueError as e: 1933 raise HTTPException(400, str(e)) from None 1934 mode = str(payload.get("mode", "service")) 1935 address = payload.get("address") 1936 file_name = payload.get("fileName") 1937 1938 common: dict[str, Any] = {"mode": mode} 1939 if address is not None: 1940 common["address"] = int(address) 1941 common["isLongAddress"] = bool(payload.get("isLongAddress")) 1942 if file_name: 1943 common["fileName"] = file_name 1944 1945 # Compound: prepend CV31/CV32 setup as service writes (no store). 1946 if split is not None: 1947 for setup_cv, setup_byte in ((31, split.bank), (32, split.index or 0)): 1948 await _dispatch( 1949 handle, type="programmer", method="post", 1950 data={**common, "cv": setup_cv, "value": int(setup_byte or 0)}, 1951 ) 1952 # Read the base CV; skip service-side persistence, we'll set the 1953 # store manually with the correct key below. 1954 try: 1955 data = await _dispatch( 1956 handle, type="programmer", method="get", 1957 data={**common, "cv": wire_cv}, 1958 ) 1959 except HTTPException as e: 1960 raise HTTPException(e.status_code, e.detail) from None 1961 value = int(data.get("value", 0)) 1962 from jmri_server.services.programmer import _loco_cv_store # noqa: PLC0415 1963 _loco_cv_store(handle.ctx).record_read_result( 1964 loco_id, cv_name, value, 1965 file_name if isinstance(file_name, str) else None, 1966 ) 1967 return {"cvName": cv_name, "value": value} 1968 1969 @app.post("/programmer/loco/{loco_id}/cv/write") 1970 async def programmer_loco_raw_cv_write( 1971 loco_id: str, payload: dict[str, Any] = Body(...), 1972 ) -> Any: 1973 """Write one CV on the CVs tab. Plain or compound; same shape 1974 as the read endpoint but with a ``value`` byte. 1975 1976 Compound writes prepend CV31/CV32 setup writes and persist the 1977 payload under the compound key (matching JMRI Java's roster 1978 format ``<CVvalue name="16.1.275" value="X"/>``).""" 1979 cv_name = str(payload.get("cvName") or "") 1980 try: 1981 wire_cv, split = _parse_raw_cv_name(cv_name) 1982 except ValueError as e: 1983 raise HTTPException(400, str(e)) from None 1984 try: 1985 value = int(payload["value"]) 1986 except (KeyError, TypeError, ValueError): 1987 raise HTTPException(400, "value must be an int 0..255") from None 1988 if not 0 <= value <= 255: 1989 raise HTTPException(400, "value out of range 0..255") 1990 1991 mode = str(payload.get("mode", "service")) 1992 address = payload.get("address") 1993 file_name = payload.get("fileName") 1994 common: dict[str, Any] = {"mode": mode} 1995 if address is not None: 1996 common["address"] = int(address) 1997 common["isLongAddress"] = bool(payload.get("isLongAddress")) 1998 if file_name: 1999 common["fileName"] = file_name 2000 2001 if split is not None: 2002 for setup_cv, setup_byte in ((31, split.bank), (32, split.index or 0)): 2003 await _dispatch( 2004 handle, type="programmer", method="post", 2005 data={**common, "cv": setup_cv, "value": int(setup_byte or 0)}, 2006 ) 2007 # Base-CV write. Omit ``locoId`` so the service's built-in store 2008 # update doesn't fire under the plain int key — we do it below 2009 # with the compound (or plain) name. 2010 try: 2011 data = await _dispatch( 2012 handle, type="programmer", method="post", 2013 data={**common, "cv": wire_cv, "value": value}, 2014 ) 2015 except HTTPException as e: 2016 raise HTTPException(e.status_code, e.detail) from None 2017 from jmri_server.services.programmer import _loco_cv_store # noqa: PLC0415 2018 _loco_cv_store(handle.ctx).set_cv( 2019 loco_id, cv_name, value, 2020 file_name if isinstance(file_name, str) else None, 2021 ) 2022 return {"cvName": cv_name, "value": int(data.get("value", value))} 2023 2024 # ----- programmer step 4: dirty-tracked batch writes ---------------- 2025 2026 from jmri_server.programming_queue import ( # noqa: PLC0415 2027 BatchInFlightError, 2028 BatchWrite, 2029 get_or_create_queue, 2030 ) 2031 2032 def _make_write_callback(loco_id: str) -> Any: 2033 """Build the callback the queue calls per CV. 2034 2035 Two responsibilities on top of just dispatching to 2036 ``programmer.post``: 2037 2038 * **Read-modify-write for masked variables.** The client sends 2039 the variable's *human* value (e.g. Address Format=1 for long). 2040 The wire needs the full CV byte with the other variables' 2041 bits preserved. We look up the variable by ``item`` name in 2042 the decoder catalog, compute the full byte via 2043 :func:`variable_apply_to_state` against the current CV 2044 snapshot, and issue writes for the resulting CV bytes. 2045 * **Split-CV / long-address writes.** ``variable_apply_to_state`` 2046 returns one entry per touched CV; long-address writes two 2047 bytes (CV 17 + CV 18) — we dispatch each so the wire sees a 2048 real per-CV WRITE_DIR_CV_CMD. 2049 2050 A write that can't be resolved to a catalog variable (no 2051 catalog loaded, item name missing) falls back to writing the 2052 raw value to the raw CV — matches the pre-fix behaviour so 2053 single-CV panes / raw-CV writes still work. 2054 """ 2055 async def _dispatch_one( 2056 cv: int, value: int, extra: dict[str, Any], 2057 ) -> int: 2058 payload: dict[str, Any] = { 2059 "cv": cv, "value": value, 2060 "mode": extra.get("mode", "service"), 2061 "locoId": loco_id, 2062 } 2063 file_name = extra.get("file_name") 2064 if file_name: 2065 payload["fileName"] = file_name 2066 addr = extra.get("address") 2067 if addr is not None: 2068 payload["address"] = addr 2069 payload["isLongAddress"] = bool(extra.get("long_address")) 2070 data = await _dispatch( 2071 handle, type="programmer", method="post", data=payload, 2072 ) 2073 return int(data.get("value", value)) 2074 2075 async def _dispatch_wire_only(cv: int, value: int, extra: dict[str, Any]) -> int: 2076 """Write to the wire without persisting to the CV store. 2077 2078 Used for indexed-CV setup writes (CV31=bank, CV32=index). 2079 Those bytes are physically written but must NOT show up in 2080 the roster's per-CV tracking — they're transient page- 2081 selector state, not user data. Omitting ``locoId`` from 2082 the payload tells the programmer service to skip the store 2083 update. 2084 """ 2085 payload: dict[str, Any] = { 2086 "cv": cv, "value": value, 2087 "mode": extra.get("mode", "service"), 2088 } 2089 addr = extra.get("address") 2090 if addr is not None: 2091 payload["address"] = addr 2092 payload["isLongAddress"] = bool(extra.get("long_address")) 2093 data = await _dispatch( 2094 handle, type="programmer", method="post", data=payload, 2095 ) 2096 return int(data.get("value", value)) 2097 2098 async def _dispatch_with_key( 2099 cv: int, value: int, store_key: str, extra: dict[str, Any], 2100 ) -> int: 2101 """Write to wire AND persist under ``store_key`` (used for 2102 indexed CVs whose store name is ``bank.index.cv``, not 2103 plain ``str(cv)``). We deliberately bypass the programmer 2104 service's built-in store update and do it here so we can 2105 control the key.""" 2106 wire = await _dispatch_wire_only(cv, value, extra) 2107 from jmri_server.services.programmer import _loco_cv_store # noqa: PLC0415 2108 _loco_cv_store(handle.ctx).set_cv( 2109 loco_id, store_key, value, extra.get("file_name"), 2110 ) 2111 return wire 2112 2113 async def _write(cv: int, value: int, item: str, extra: dict[str, Any]) -> int: 2114 var = _find_variable_for_write(loco_id, item) 2115 if var is None: 2116 # Composite fan-out sends ``cv=0`` as a sentinel meaning 2117 # "resolve the wire CV from ``item``". If we couldn't 2118 # resolve the item, that's a broken composite reference 2119 # or catalog mismatch — bail out with a clear error so 2120 # the batch reports the failure rather than silently 2121 # writing to CV 0. 2122 if cv == 0 and item: 2123 raise ValueError( 2124 f"composite sub-variable {item!r} not found in catalog", 2125 ) 2126 # No catalog metadata → treat as a raw single-CV write. 2127 return await _dispatch_one(cv, value, extra) 2128 # Snapshot current CV bytes from the store so read-modify-write 2129 # preserves bits owned by other variables on the same CV. 2130 # Keys are strings so indexed CVs (``16.1.275``) can coexist 2131 # with plain ones (``275``) without collision. 2132 from jmri_server.services.programmer import _loco_cv_store # noqa: PLC0415 2133 store = _loco_cv_store(handle.ctx) 2134 state = store.get(loco_id, extra.get("file_name")) 2135 cv_bytes = {str(k): int(v) for k, v in state.cv_values.items()} 2136 try: 2137 plan = variable_wire_write_plan(var, value, cv_bytes) 2138 except Exception: # noqa: BLE001 2139 # Unsupported kind (speed table, composite, ...) — 2140 # fall back to a raw write on the primary CV. 2141 return await _dispatch_one(cv, value, extra) 2142 if not plan: 2143 return int(value) # read-only / constant → no-op 2144 # Issue physical writes in order. Setup ops (CV31/CV32) 2145 # go through _dispatch_wire_only so they don't pollute 2146 # the roster; value ops go through _dispatch_with_key so 2147 # they land under the correct compound key for indexed 2148 # variables. 2149 last_committed = value 2150 for op in plan: 2151 if op.store_key is None: 2152 last_committed = await _dispatch_wire_only( 2153 op.wire_cv, op.byte_value, extra, 2154 ) 2155 else: 2156 last_committed = await _dispatch_with_key( 2157 op.wire_cv, op.byte_value, op.store_key, extra, 2158 ) 2159 return last_committed 2160 return _write 2161 2162 def _make_read_callback(loco_id: str) -> Any: 2163 """Build the read callback the queue calls per CV. 2164 2165 Delegates to the JSON ``programmer.get`` service — same wire as 2166 the single-CV Read button. The service stashes each read into 2167 the loco CV store, so after the batch completes the client can 2168 refresh the pane and see freshly-extracted values. 2169 """ 2170 async def _dispatch_read( 2171 cv: int, extra: dict[str, Any], *, store_key: str | None, 2172 ) -> int: 2173 """Wire read + optional store update under *store_key*. 2174 2175 When *store_key* is None (indexed CV31/CV32 setup writes 2176 aren't reads — this shouldn't be called), or when the read 2177 is for a simple CV that keys plainly as ``str(cv)``, we 2178 just include ``locoId`` in the payload and let the 2179 programmer service persist via its own ``record_read_result`` 2180 path. When the caller wants a compound store key, we omit 2181 ``locoId`` (service skips persistence) and set the value 2182 manually with the right key. 2183 """ 2184 payload: dict[str, Any] = { 2185 "cv": cv, 2186 "mode": extra.get("mode", "service"), 2187 } 2188 file_name = extra.get("file_name") 2189 if file_name: 2190 payload["fileName"] = file_name 2191 addr = extra.get("address") 2192 if addr is not None: 2193 payload["address"] = addr 2194 payload["isLongAddress"] = bool(extra.get("long_address")) 2195 # For plain reads we can let the service persist directly. 2196 if store_key is None or store_key == str(cv): 2197 payload["locoId"] = loco_id 2198 data = await _dispatch( 2199 handle, type="programmer", method="get", data=payload, 2200 ) 2201 return int(data.get("value", 0)) 2202 # Compound key path (indexed CV): dispatch without locoId, 2203 # then manually record under the compound key. 2204 data = await _dispatch( 2205 handle, type="programmer", method="get", data=payload, 2206 ) 2207 value = int(data.get("value", 0)) 2208 from jmri_server.services.programmer import _loco_cv_store # noqa: PLC0415 2209 _loco_cv_store(handle.ctx).record_read_result( 2210 loco_id, store_key, value, file_name, 2211 ) 2212 return value 2213 2214 async def _read(cv: int, item: str, extra: dict[str, Any]) -> int: 2215 var = _find_variable_for_write(loco_id, item) 2216 if var is None or not variable_is_indexed(var): 2217 # Plain read — service persists under ``str(cv)``. 2218 return await _dispatch_read(cv, extra, store_key=str(cv)) 2219 # Indexed read: CV31/CV32 setup (writes, no persistence), 2220 # then read the base CV and persist under compound key. 2221 assert var.split is not None 2222 # For a read op, we need to issue writes to CV31/CV32 too. 2223 # Those aren't in the "read wire plan" we can call 2224 # directly — write the setup manually using the 2225 # write-side dispatch helper. Fetch it via a lexical 2226 # closure — we're inside the same _make_read_callback 2227 # scope but the write helpers were defined in 2228 # _make_write_callback; reach through the queue by 2229 # calling the JSON write path directly here to avoid 2230 # crossing closures. 2231 async def _wire_write(target_cv: int, target_byte: int) -> None: 2232 setup_payload: dict[str, Any] = { 2233 "cv": target_cv, "value": target_byte, 2234 "mode": extra.get("mode", "service"), 2235 } 2236 addr = extra.get("address") 2237 if addr is not None: 2238 setup_payload["address"] = addr 2239 setup_payload["isLongAddress"] = bool( 2240 extra.get("long_address"), 2241 ) 2242 await _dispatch( 2243 handle, type="programmer", method="post", data=setup_payload, 2244 ) 2245 2246 await _wire_write(31, var.split.bank or 0) 2247 await _wire_write(32, var.split.index or 0) 2248 store_key = variable_store_key(var, cv) 2249 return await _dispatch_read(cv, extra, store_key=store_key) 2250 return _read 2251 2252 def _find_variable_for_write(loco_id: str, item: str) -> Any: 2253 """Look up the decoder variable named *item* for *loco_id*. 2254 2255 Returns ``None`` when the catalog isn't loaded, the loco has no 2256 decoder family/model, or the item name doesn't match anything 2257 in the definition. Callers should fall back to a raw CV write 2258 in that case. 2259 """ 2260 if not item: 2261 return None 2262 try: 2263 from jmri_server.services.roster import ( # noqa: PLC0415 2264 _roster_service_get_sync, 2265 ) 2266 entry = _roster_service_get_sync(handle.ctx, loco_id) 2267 except Exception: # noqa: BLE001 2268 return None 2269 if not entry: 2270 return None 2271 family = entry.get("decoderFamily") or "" 2272 model = entry.get("decoderModel") or "" 2273 if not family or not model: 2274 return None 2275 defn = _catalog().find(family, model) 2276 if defn is None: 2277 return None 2278 for v in defn.variables: 2279 if v.item == item: 2280 return v 2281 return None 2282 2283 # CV 29's Address Format bit — set means "use long address", clear 2284 # means "use short". Any variable whose mask overlaps this bit will 2285 # flip the decoder's address mode when written. 2286 _CV29_ADDRESS_FORMAT_BIT = 0x20 2287 2288 def _ops_mode_address_change_reason( 2289 write: "BatchWrite", loco_id: str, 2290 ) -> str | None: 2291 """Return a human-readable reason if *write* would change the 2292 decoder's operating address, or ``None`` if it's safe in ops mode. 2293 2294 Ops-mode POM writes reach the decoder via a DCC packet targeted 2295 at the current DCC address. When the write itself changes that 2296 address (long-address CVs 17/18, short-address CV1, or the CV29 2297 Address Format bit), the wire packet is fine, but *subsequent* 2298 writes in the same batch aim at the old address and go 2299 unanswered. Long-address specifically requires two atomic CV 2300 writes (CV17 + CV18); doing them via POM leaves the decoder in 2301 an inconsistent state between the two. 2302 2303 We refuse rather than warn — the safe alternative is to run the 2304 exact same edit on the programming track (service mode), which 2305 the user can do by flipping the picker at the top of the pane. 2306 """ 2307 var = _find_variable_for_write(loco_id, write.item) 2308 if var is not None: 2309 if var.kind is VariableKind.LONG_ADDRESS: 2310 return ( 2311 "Long Address writes touch CV17 and CV18 atomically; " 2312 "ops mode can't do that safely because the address " 2313 "changes between the two packets. Use service mode." 2314 ) 2315 if var.kind is VariableKind.SHORT_ADDRESS: 2316 return ( 2317 "Short Address writes change the decoder's address; " 2318 "subsequent ops-mode packets would target the old " 2319 "address. Use service mode." 2320 ) 2321 # Any masked variable on CV29 whose ``V`` region overlaps 2322 # bit 5 flips the address-format mode. That'd break follow- 2323 # up POM packets in the same batch. 2324 primary = variable_primary_cv(var) 2325 if ( 2326 primary == 29 2327 and var.bit_mask.mask_value & _CV29_ADDRESS_FORMAT_BIT 2328 ): 2329 return ( 2330 "Changing CV29's Address Format bit switches the " 2331 "decoder between short and long addressing; " 2332 "subsequent ops-mode packets would target the old " 2333 "address. Use service mode." 2334 ) 2335 return None 2336 2337 @app.post("/programmer/loco/{loco_id}/write-batch") 2338 async def programmer_loco_write_batch( 2339 request: Request, loco_id: str, 2340 payload: dict[str, Any] = Body(...), 2341 ) -> Any: 2342 """Kick off a batch of CV writes for *loco_id*. 2343 2344 Payload shape:: 2345 2346 { 2347 "writes": [{"cv": 1, "value": 3, "item": "Short Address"}, ...], 2348 "mode": "service" | "ops", 2349 "address": <int>, // ops mode only 2350 "isLongAddress": true|false, 2351 "fileName": "1.xml" // optional; JMRI roster fileName 2352 } 2353 2354 Returns 202 + the batch id on success, or 409 if a batch is 2355 already in flight on this loco. Refuses with 400 when the 2356 batch is ops-mode and any write would change the decoder's 2357 operating address mid-batch (long address, short address, CV29 2358 Address Format flip). 2359 """ 2360 raw_writes = payload.get("writes") or [] 2361 if not isinstance(raw_writes, list) or not raw_writes: 2362 raise HTTPException(400, "writes must be a non-empty list") 2363 writes: list[BatchWrite] = [] 2364 for w in raw_writes: 2365 try: 2366 writes.append(BatchWrite( 2367 cv=int(w["cv"]), 2368 value=int(w["value"]), 2369 item=str(w.get("item", "")), 2370 )) 2371 except (KeyError, TypeError, ValueError) as e: 2372 raise HTTPException(400, f"malformed write row: {e}") from None 2373 2374 mode = str(payload.get("mode", "service")) 2375 address = payload.get("address") 2376 long_address = bool(payload.get("isLongAddress", False)) 2377 file_name = payload.get("fileName") or None 2378 2379 # Ops-mode safety gate: refuse any write that would change the 2380 # loco's operating address mid-batch. Check up front so we 2381 # don't start a batch that'd fail halfway through (CV17 2382 # written, CV18 aimed at wrong address). 2383 if mode == "ops": 2384 for w in writes: 2385 reason = _ops_mode_address_change_reason(w, loco_id) 2386 if reason is not None: 2387 raise HTTPException( 2388 400, 2389 f"cannot write {w.item or f'CV {w.cv}'} in ops mode: " 2390 f"{reason}", 2391 ) 2392 2393 queue = get_or_create_queue( 2394 handle.ctx, loco_id, _make_write_callback(loco_id), 2395 read_callback=_make_read_callback(loco_id), 2396 ) 2397 try: 2398 batch = await queue.start( 2399 writes, 2400 mode=mode, 2401 address=int(address) if address is not None else None, 2402 long_address=long_address, 2403 file_name=file_name, 2404 ) 2405 except BatchInFlightError as e: 2406 raise HTTPException(409, str(e)) from None 2407 from fastapi.responses import JSONResponse # noqa: PLC0415 2408 return JSONResponse( 2409 {"batch_id": batch.batch_id}, status_code=202, 2410 ) 2411 2412 @app.post("/programmer/loco/{loco_id}/read-batch") 2413 async def programmer_loco_read_batch( 2414 request: Request, loco_id: str, 2415 payload: dict[str, Any] = Body(...), 2416 ) -> Any: 2417 """Kick off a batch of CV reads for *loco_id*. 2418 2419 Payload shape:: 2420 2421 { 2422 "reads": [{"cv": 1, "item": "Short Address"}, ...], 2423 "mode": "service" | "ops", 2424 "address": <int>, // ops mode only (rarely 2425 // useful; PowerPro can't read 2426 // ops-mode anyway) 2427 "fileName": "1.xml" 2428 } 2429 2430 Returns 202 + the batch id, or 409 if a batch is already in 2431 flight. Duplicate CV numbers in the payload are deduplicated 2432 here — reading the same CV twice in one batch is wasted time on 2433 a service-mode track (each read is 2–6 s), and the store already 2434 caches whichever value came back last. 2435 """ 2436 raw_reads = payload.get("reads") or [] 2437 if not isinstance(raw_reads, list) or not raw_reads: 2438 raise HTTPException(400, "reads must be a non-empty list") 2439 # Dedupe by CV, first-wins on the display label so the user 2440 # sees a sensible item name in the progress fragment. 2441 seen_cvs: set[int] = set() 2442 ops: list[BatchWrite] = [] 2443 for r in raw_reads: 2444 try: 2445 cv = int(r["cv"]) 2446 except (KeyError, TypeError, ValueError) as e: 2447 raise HTTPException(400, f"malformed read row: {e}") from None 2448 if cv in seen_cvs: 2449 continue 2450 seen_cvs.add(cv) 2451 ops.append(BatchWrite( 2452 cv=cv, kind="read", item=str(r.get("item", "")), 2453 )) 2454 2455 mode = str(payload.get("mode", "service")) 2456 address = payload.get("address") 2457 long_address = bool(payload.get("isLongAddress", False)) 2458 file_name = payload.get("fileName") or None 2459 2460 queue = get_or_create_queue( 2461 handle.ctx, loco_id, _make_write_callback(loco_id), 2462 read_callback=_make_read_callback(loco_id), 2463 ) 2464 try: 2465 batch = await queue.start( 2466 ops, 2467 mode=mode, 2468 address=int(address) if address is not None else None, 2469 long_address=long_address, 2470 file_name=file_name, 2471 ) 2472 except BatchInFlightError as e: 2473 raise HTTPException(409, str(e)) from None 2474 from fastapi.responses import JSONResponse # noqa: PLC0415 2475 return JSONResponse( 2476 {"batch_id": batch.batch_id}, status_code=202, 2477 ) 2478 2479 @app.get("/programmer/loco/{loco_id}/batch/status") 2480 async def programmer_loco_batch_status(loco_id: str) -> Any: 2481 """Poll the current batch's status for *loco_id*. 2482 2483 Returns the batch's serializable snapshot or 404 if there's no 2484 active or recently-completed batch on this loco. 2485 """ 2486 queues = getattr(handle.ctx, "_programming_queues", {}) 2487 queue = queues.get(loco_id) if isinstance(queues, dict) else None 2488 if queue is None or queue.current is None: 2489 raise HTTPException(404, "no active batch") 2490 return queue.current.to_status() 2491 2492 @app.post("/programmer/loco/{loco_id}/batch/cancel") 2493 async def programmer_loco_batch_cancel(loco_id: str) -> Any: 2494 """Request cancel on the current batch. The in-flight CV write 2495 finishes; subsequent queued writes are marked cancelled.""" 2496 queues = getattr(handle.ctx, "_programming_queues", {}) 2497 queue = queues.get(loco_id) if isinstance(queues, dict) else None 2498 if queue is None or queue.current is None: 2499 raise HTTPException(404, "no active batch") 2500 batch = queue.cancel() 2501 return batch.to_status() if batch else {"state": "no-op"} 2502 2503 @app.post("/programmer/loco/{loco_id}/batch/clear") 2504 async def programmer_loco_batch_clear(loco_id: str) -> Any: 2505 """Drop a terminal-state batch so the UI can start fresh.""" 2506 queues = getattr(handle.ctx, "_programming_queues", {}) 2507 queue = queues.get(loco_id) if isinstance(queues, dict) else None 2508 if queue is None: 2509 return {"cleared": False} 2510 try: 2511 queue.clear() 2512 except BatchInFlightError as e: 2513 raise HTTPException(409, str(e)) from None 2514 return {"cleared": True} 2515 2516 @app.post("/programmer/catalog/load", response_class=HTMLResponse) 2517 async def programmer_load_catalog(request: Request, directory: str = Form(...)) -> Any: 2518 """Kick off a catalog load in the background and return a 2519 polling widget. Actual load happens on a worker thread because 2520 the JMRI decoder tree has hundreds of files and parsing takes 2521 30-60 s on a Pi. See :func:`programmer_catalog_progress` for 2522 the polling endpoint the returned widget hits.""" 2523 import asyncio # noqa: PLC0415 2524 import pathlib as _pathlib # noqa: PLC0415 2525 2526 target = _pathlib.Path(directory).expanduser() 2527 if not target.exists(): 2528 return templates.TemplateResponse( 2529 request, "_programmer_catalog.html", 2530 {"error": f"directory does not exist: {directory}"}, 2531 status_code=400, 2532 ) 2533 if not target.is_dir(): 2534 return templates.TemplateResponse( 2535 request, "_programmer_catalog.html", 2536 {"error": f"not a directory: {directory}"}, 2537 status_code=400, 2538 ) 2539 2540 # Shared state the loader worker writes and the poll route reads. 2541 # A single load is in flight at a time; overlapping loads are 2542 # rejected to keep progress state coherent. 2543 progress = getattr(handle.ctx, "_catalog_load_progress", None) 2544 if progress and not progress.get("done") and not progress.get("error"): 2545 return templates.TemplateResponse( 2546 request, "_programmer_catalog.html", 2547 {"error": "a catalog load is already in progress"}, 2548 status_code=409, 2549 ) 2550 2551 # File-count precomputed so the widget can render a percentage. 2552 files = list(target.rglob("*.xml")) 2553 total = len(files) 2554 handle.ctx._catalog_load_progress = { # type: ignore[attr-defined] 2555 "loaded": 0, 2556 "total": total, 2557 "done": False, 2558 "error": None, 2559 "directory": str(target), 2560 } 2561 2562 def _worker() -> None: 2563 state = handle.ctx._catalog_load_progress # type: ignore[attr-defined] 2564 try: 2565 # Fast metadata scan — extracts only <family> and 2566 # <model> attrs per file. Variables/panes are parsed 2567 # lazily by DecoderCatalog.find() on first loco open. 2568 # See jmri_core.persistence.decoder_xml.iter_decoder_index. 2569 from jmri_core.persistence.decoder_xml import ( # noqa: PLC0415 2570 iter_decoder_index, 2571 ) 2572 catalog = DecoderCatalog() 2573 seen_files: set[pathlib.Path] = set() 2574 for entry in iter_decoder_index(target): 2575 catalog.add_index_entry(entry) 2576 # Progress ticks per unique source file so the 2577 # thermometer maps to file count, not entry count 2578 # (one file can emit many entries). 2579 if entry.source_file not in seen_files: 2580 seen_files.add(entry.source_file) 2581 state["loaded"] += 1 2582 # Sync progress up to the total in case files were 2583 # skipped (non-decoder XML). The pane fragment reads 2584 # ``loaded/total`` to compute the percentage. 2585 state["loaded"] = state["total"] 2586 # Auto-load sibling programmers/ (same layout as 2587 # DecoderCatalog.from_directory). 2588 programmers_dir = target.parent / "programmers" 2589 if programmers_dir.is_dir(): 2590 from jmri_core.persistence.programmer_xml import ( # noqa: PLC0415 2591 iter_programmer_configs, 2592 ) 2593 catalog.programmers = { 2594 pc.name: pc 2595 for pc in iter_programmer_configs(programmers_dir) 2596 } 2597 # NMRA manufacturer table sits at the ``xml/`` root as 2598 # ``decoderIndex.xml``. Load it so ``identify_by_cv`` 2599 # has the authoritative 179-entry mfgID → name mapping 2600 # instead of the small hardcoded stopgap. 2601 index_path = target.parent / "decoderIndex.xml" 2602 if index_path.is_file(): 2603 from jmri_decoderpro.catalog import _load_mfg_ids # noqa: PLC0415 2604 catalog.mfg_ids = _load_mfg_ids(index_path) 2605 handle.ctx._decoder_catalog = catalog # type: ignore[attr-defined] 2606 except Exception as e: # noqa: BLE001 2607 state["error"] = str(e) 2608 state["done"] = True 2609 2610 # asyncio.to_thread runs the sync loader on the default thread 2611 # pool. Fire-and-forget; the poll endpoint watches ``progress``. 2612 asyncio.get_event_loop().create_task(asyncio.to_thread(_worker)) 2613 2614 return templates.TemplateResponse( 2615 request, "_programmer_catalog_progress.html", 2616 {"total": total, "directory": str(target)}, 2617 ) 2618 2619 @app.get("/programmer/catalog/progress", response_class=HTMLResponse) 2620 async def programmer_catalog_progress(request: Request) -> Any: 2621 """HTMX polls this every 500 ms while a load is in flight. 2622 2623 Returns: 2624 * a progress fragment (percentage bar) while ``done`` is False; 2625 * the final "Catalog loaded" fragment when ``done`` is True; 2626 * an error fragment when the worker set ``error``. 2627 """ 2628 progress = getattr(handle.ctx, "_catalog_load_progress", None) 2629 if not progress: 2630 # No load in flight — return the current catalog badge so 2631 # the poll loop stops (HTMX unbinds when the response no 2632 # longer includes the progress element). 2633 catalog = _catalog() 2634 return templates.TemplateResponse( 2635 request, "_programmer_catalog.html", 2636 { 2637 "catalog_size": len(catalog), 2638 "manufacturers": catalog.manufacturers(), 2639 "programmer_count": len(catalog.programmers), 2640 }, 2641 ) 2642 if progress.get("error"): 2643 return templates.TemplateResponse( 2644 request, "_programmer_catalog.html", 2645 {"error": f"failed to load: {progress['error']}"}, 2646 status_code=400, 2647 ) 2648 if progress.get("done"): 2649 catalog = _catalog() 2650 return templates.TemplateResponse( 2651 request, "_programmer_catalog.html", 2652 { 2653 "catalog_size": len(catalog), 2654 "manufacturers": catalog.manufacturers(), 2655 "programmer_count": len(catalog.programmers), 2656 }, 2657 ) 2658 return templates.TemplateResponse( 2659 request, "_programmer_catalog_progress.html", 2660 { 2661 "total": progress["total"], 2662 "loaded": progress["loaded"], 2663 "directory": progress["directory"], 2664 }, 2665 ) 2666 2667 @app.get("/programmer/decoders", response_class=HTMLResponse) 2668 async def programmer_decoders( 2669 request: Request, mfg: str | None = None, family: str | None = None, 2670 ) -> Any: 2671 if family: 2672 models = _catalog().models(family) 2673 return templates.TemplateResponse( 2674 request, "_programmer_models.html", 2675 {"family": family, "models": models}, 2676 ) 2677 if mfg: 2678 families = _catalog().families(mfg) 2679 return templates.TemplateResponse( 2680 request, "_programmer_families.html", 2681 {"mfg": mfg, "families": families}, 2682 ) 2683 return HTMLResponse("") 2684 2685 @app.get("/programmer/variables", response_class=HTMLResponse) 2686 async def programmer_variables( 2687 request: Request, family: str, model: str, 2688 ) -> Any: 2689 defn = _catalog().find(family, model) 2690 if defn is None: 2691 return HTMLResponse("decoder not found", status_code=404) 2692 return templates.TemplateResponse( 2693 request, "_programmer_variables.html", 2694 {"definition": defn, "family": family, "model": model}, 2695 ) 2696 2697 @app.post("/programmer/identify", response_class=HTMLResponse) 2698 async def programmer_identify(request: Request) -> Any: 2699 """Read CV7 + CV8 via the service-mode programmer, then look up 2700 matching definitions in the loaded catalog. 2701 2702 Failure modes surfaced clearly rather than as a spinning 2703 request: 2704 2705 * Read errored / timed out — the underlying error message goes 2706 into the fragment. 2707 * Read succeeded but CV8 isn't a known NMRA mfg ID (most often 2708 because the decoder didn't ack the read, so the CS returned 2709 its NoAck sentinel byte) — user sees "no decoder / unknown 2710 manufacturer" instead of the catalog quietly iterating 2711 thousands of files. 2712 """ 2713 from jmri_decoderpro import UnknownDecoderIdentityError # noqa: PLC0415 2714 # NCE PowerPro service-mode reads are ~2-6 s each and 2715 # intermittently return the NoAck sentinel (0/48/255). 2716 # Auto-retry once on those to avoid making the user click 2717 # Identify twice for a transient wire hiccup. Any hard 2718 # HTTPException short-circuits to the error card as before. 2719 _WIRE_FAIL = {0, 48, 255} 2720 async def _read_cv_with_retry(cv: int) -> dict: 2721 first = await _dispatch( 2722 handle, type="programmer", method="get", 2723 data={"cv": cv, "mode": "service"}, 2724 ) 2725 if first.get("value") in _WIRE_FAIL: 2726 # Sentinel value → very likely a transient read failure. 2727 # Do one silent retry; if it succeeds, use that. 2728 try: 2729 second = await _dispatch( 2730 handle, type="programmer", method="get", 2731 data={"cv": cv, "mode": "service"}, 2732 ) 2733 except HTTPException: 2734 return first 2735 if second.get("value") not in _WIRE_FAIL: 2736 return second 2737 return first 2738 try: 2739 cv7 = await _read_cv_with_retry(7) 2740 cv8 = await _read_cv_with_retry(8) 2741 except HTTPException as e: 2742 return templates.TemplateResponse( 2743 request, "_programmer_identify_result.html", 2744 {"error": e.detail}, status_code=e.status_code, 2745 ) 2746 try: 2747 candidates = _catalog().identify_by_cv(cv7["value"], cv8["value"]) 2748 except UnknownDecoderIdentityError as e: 2749 return templates.TemplateResponse( 2750 request, "_programmer_identify_result.html", 2751 {"cv7": cv7["value"], "cv8": cv8["value"], "error": str(e)}, 2752 ) 2753 # Manufacturer-specific productID probe. JMRI's real identify 2754 # is per-mfg: SoundTraxx encodes productID across CVs 253/255/ 2755 # 256, other makers use other CVs. Follow the same protocol so 2756 # a Tsunami2 read collapses from ~54 families to a single 2757 # model (e.g. "TSU-1100 Baldwin Diesels", productID 2071). 2758 product_id: str | None = None 2759 product_id_error: str | None = None 2760 if candidates and "soundtraxx" in candidates[0].mfg.lower(): 2761 try: 2762 cv253 = await _dispatch( 2763 handle, type="programmer", method="get", 2764 data={"cv": 253, "mode": "service"}, 2765 ) 2766 cv255 = await _dispatch( 2767 handle, type="programmer", method="get", 2768 data={"cv": 255, "mode": "service"}, 2769 ) 2770 cv256 = await _dispatch( 2771 handle, type="programmer", method="get", 2772 data={"cv": 256, "mode": "service"}, 2773 ) 2774 # JMRI Java's IdentifyDecoder formula: 2775 # productID = CV256 | ((CV255 & 7) << 8) | (CV253 << 11) 2776 pid = ( 2777 (cv256["value"] & 0xFF) 2778 | ((cv255["value"] & 0x07) << 8) 2779 | ((cv253["value"] & 0xFF) << 11) 2780 ) 2781 product_id = str(pid) 2782 except HTTPException as e: 2783 product_id_error = e.detail 2784 # Filter by productID when we have one that matches at least 2785 # one candidate; if nothing matches, keep the family list so 2786 # the user still has something to pick from. 2787 if product_id is not None: 2788 narrowed = [ 2789 c for c in candidates 2790 if c.product_id and c.product_id == product_id 2791 ] 2792 if narrowed: 2793 candidates = narrowed 2794 # Group candidates by (mfg, family) so the picker doesn't render 2795 # one row per model — a SoundTraxx match on a fully-loaded 2796 # catalog can produce 700+ models. The user picks the family 2797 # here and then the specific model from the Model dropdown 2798 # above (which is filtered to the chosen family). 2799 from collections import defaultdict # noqa: PLC0415 2800 by_family: dict[tuple[str, str], dict[str, Any]] = {} 2801 for c in candidates: 2802 key = (c.mfg, c.family) 2803 slot = by_family.setdefault(key, { 2804 "mfg": c.mfg, "family": c.family, 2805 "model_count": 0, "models": [], 2806 }) 2807 slot["model_count"] += 1 2808 slot["models"].append(c.model) 2809 family_candidates = [by_family[k] for k in sorted(by_family)] 2810 return templates.TemplateResponse( 2811 request, "_programmer_identify_result.html", 2812 { 2813 "cv7": cv7["value"], "cv8": cv8["value"], 2814 "candidates": family_candidates, 2815 "total_models": len(candidates), 2816 "product_id": product_id, 2817 "product_id_error": product_id_error, 2818 }, 2819 ) 2820 2821 @app.post("/programmer/cv/read", response_class=HTMLResponse) 2822 async def programmer_read_cv( 2823 request: Request, cv: int = Form(...), mode: str = Form("service"), 2824 address: int | None = Form(None), 2825 ) -> Any: 2826 payload: dict[str, Any] = {"cv": cv, "mode": mode} 2827 if address is not None: 2828 payload["address"] = address 2829 try: 2830 data = await _dispatch(handle, type="programmer", method="get", data=payload) 2831 except HTTPException as e: 2832 return templates.TemplateResponse( 2833 request, "_programmer_cv_result.html", 2834 {"cv": cv, "error": e.detail}, 2835 status_code=e.status_code, 2836 ) 2837 return templates.TemplateResponse( 2838 request, "_programmer_cv_result.html", 2839 {"cv": cv, "value": data["value"], "mode": data["mode"]}, 2840 ) 2841 2842 @app.post("/programmer/cv/write", response_class=HTMLResponse) 2843 async def programmer_write_cv( 2844 request: Request, cv: int = Form(...), value: int = Form(...), 2845 mode: str = Form("service"), address: int | None = Form(None), 2846 ) -> Any: 2847 payload: dict[str, Any] = {"cv": cv, "value": value, "mode": mode} 2848 if address is not None: 2849 payload["address"] = address 2850 try: 2851 data = await _dispatch(handle, type="programmer", method="post", data=payload) 2852 except HTTPException as e: 2853 return templates.TemplateResponse( 2854 request, "_programmer_cv_result.html", 2855 {"cv": cv, "error": e.detail}, 2856 status_code=e.status_code, 2857 ) 2858 return templates.TemplateResponse( 2859 request, "_programmer_cv_result.html", 2860 {"cv": cv, "value": data["value"], "mode": data["mode"], "wrote": True}, 2861 ) 2862 2863 # ----- scripts ---------------------------------------------------------- 2864 2865 def _repl_session(): 2866 """Return the shared REPL session, lazy-creating on first hit. 2867 2868 The session lives on ``ctx._repl_session`` so it survives 2869 across every browser tab hitting the /scripts page — matching 2870 JMRI Java's Jython window semantics (one shared interpreter). 2871 Rebuilt from scratch on ``/scripts/repl/reset``. 2872 """ 2873 from jmri_scripting import ReplSession # noqa: PLC0415 2874 cached = getattr(handle.ctx, "_repl_session", None) 2875 if cached is not None: 2876 return cached 2877 # Piggy-back on the ScriptRunner to reuse its shim binding — 2878 # the ScriptService lazy-constructs the runner on first use; 2879 # do the same dance here so both share one JmriShim, one 2880 # dispatch-stream install, one connection. 2881 runner = getattr(handle.ctx, "_script_runner", None) 2882 if runner is None: 2883 from jmri_scripting import ScriptRunner # noqa: PLC0415 2884 import asyncio as _asyncio # noqa: PLC0415 2885 runner = ScriptRunner.for_connection( 2886 _asyncio.get_event_loop(), handle.ctx.connection, 2887 ) 2888 handle.ctx._script_runner = runner # type: ignore[attr-defined] 2889 session = ReplSession(runner.shim) 2890 handle.ctx._repl_session = session # type: ignore[attr-defined] 2891 return session 2892 2893 @app.post("/scripts/repl", response_class=HTMLResponse) 2894 async def scripts_repl( 2895 request: Request, source: str = Form(""), 2896 ) -> Any: # type: ignore[valid-type] 2897 """Submit *source* to the persistent interpreter, render 2898 the console fragment with the updated history. 2899 2900 Empty submissions are no-ops (rendering just refreshes 2901 the console) — matches the CPython REPL's "just hit 2902 Enter" behavior at the prompt.""" 2903 session = _repl_session() 2904 pending: str | None = None 2905 if source.strip(): 2906 # ``submit_async`` off-loads the exec to a worker 2907 # thread so shim calls that block on our event loop 2908 # don't deadlock — the loop stays free to advance 2909 # the shim's scheduled coroutines. 2910 result = await session.submit_async(source) 2911 if result.incomplete: 2912 pending = source 2913 return templates.TemplateResponse( 2914 request, "_repl_console.html", 2915 { 2916 "history": session.history, 2917 "pending_continuation": pending, 2918 }, 2919 ) 2920 2921 @app.post("/scripts/repl/reset", response_class=HTMLResponse) 2922 async def scripts_repl_reset(request: Request) -> Any: # type: ignore[valid-type] 2923 session = _repl_session() 2924 session.reset() 2925 return templates.TemplateResponse( 2926 request, "_repl_console.html", 2927 {"history": session.history, "pending_continuation": None}, 2928 ) 2929 2930 @app.get("/scripts", response_class=HTMLResponse) 2931 async def scripts_home(request: Request) -> Any: 2932 statuses = await _dispatch( 2933 handle, type="scripts", method="list", 2934 data={"library": True}, 2935 ) 2936 session = _repl_session() 2937 return templates.TemplateResponse( 2938 request, "scripts.html", 2939 _base_context(handle, "scripts") | { 2940 "statuses": statuses, 2941 "history": session.history, 2942 "pending_continuation": None, 2943 }, 2944 ) 2945 2946 @app.post("/scripts/run", response_class=HTMLResponse) 2947 async def scripts_run( 2948 request: Request, 2949 name: str = Form(...), 2950 source: str = Form(...), 2951 save: str | None = Form(None), 2952 ) -> Any: 2953 await _dispatch( 2954 handle, type="scripts", method="put", 2955 data={"name": name, "source": source, "save": bool(save)}, 2956 ) 2957 # Give the worker thread a moment to populate status. 2958 import asyncio as _asyncio # noqa: PLC0415 2959 await _asyncio.sleep(0.05) 2960 statuses = await _dispatch( 2961 handle, type="scripts", method="list", data={"library": True}, 2962 ) 2963 return templates.TemplateResponse( 2964 request, "_scripts_table.html", {"statuses": statuses}, 2965 ) 2966 2967 @app.post("/scripts/run-file/{name}", response_class=HTMLResponse) 2968 async def scripts_run_file(request: Request, name: str) -> Any: 2969 """Run an on-disk script by name. 2970 2971 Distinct from ``/scripts/run`` which takes fresh source 2972 from the textarea. This route re-reads the file from disk 2973 every call, so the debug loop (edit externally → click 2974 Run → see output) works. The runner replaces any 2975 same-named existing status entry so re-runs stack cleanly. 2976 """ 2977 try: 2978 await _dispatch( 2979 handle, type="scripts", method="post", 2980 data={"name": name, "action": "run_file"}, 2981 ) 2982 except HTTPException as exc: 2983 # Surface the 404 / permission error as a plain-text 2984 # banner in the table region so htmx doesn't blank the 2985 # UI on a stray click. 2986 statuses = await _dispatch( 2987 handle, type="scripts", method="list", 2988 data={"library": True}, 2989 ) 2990 return templates.TemplateResponse( 2991 request, "_scripts_table.html", 2992 {"statuses": statuses, "run_error": exc.detail}, 2993 ) 2994 # Same 50ms grace as /scripts/run so a fast-finishing 2995 # script has a chance to populate its status before the 2996 # table renders. 2997 import asyncio as _asyncio # noqa: PLC0415 2998 await _asyncio.sleep(0.05) 2999 statuses = await _dispatch( 3000 handle, type="scripts", method="list", data={"library": True}, 3001 ) 3002 return templates.TemplateResponse( 3003 request, "_scripts_table.html", {"statuses": statuses}, 3004 ) 3005 3006 @app.post("/scripts/stop/{name}", response_class=HTMLResponse) 3007 async def scripts_stop(request: Request, name: str) -> Any: 3008 await _dispatch( 3009 handle, type="scripts", method="post", 3010 data={"name": name, "action": "stop"}, 3011 ) 3012 statuses = await _dispatch( 3013 handle, type="scripts", method="list", data={"library": True}, 3014 ) 3015 return templates.TemplateResponse( 3016 request, "_scripts_table.html", {"statuses": statuses}, 3017 ) 3018 3019 @app.post("/scripts/delete/{name}", response_class=HTMLResponse) 3020 async def scripts_delete(request: Request, name: str) -> Any: 3021 """Stop the named script AND remove its on-disk file so the 3022 profile autoload doesn't bring it back.""" 3023 await _dispatch( 3024 handle, type="scripts", method="post", 3025 data={"name": name, "action": "delete"}, 3026 ) 3027 statuses = await _dispatch( 3028 handle, type="scripts", method="list", data={"library": True}, 3029 ) 3030 return templates.TemplateResponse( 3031 request, "_scripts_table.html", {"statuses": statuses}, 3032 ) 3033 3034 @app.get("/scripts/refresh", response_class=HTMLResponse) 3035 async def scripts_refresh(request: Request) -> Any: 3036 """Re-render the scripts table with the latest captured output. 3037 3038 Called from the per-row Refresh button so a long-running 3039 automaton's accumulated ``print()`` output surfaces without the 3040 user having to trigger a stop/start. 3041 """ 3042 statuses = await _dispatch( 3043 handle, type="scripts", method="list", data={"library": True}, 3044 ) 3045 return templates.TemplateResponse( 3046 request, "_scripts_table.html", {"statuses": statuses}, 3047 ) 3048 3049 # ----- profile CRUD ----------------------------------------------------- 3050 # Wraps the ``configProfile`` JSON topic. Post-mutation we redirect 3051 # back to ``/`` so the browser re-fetches the profile list — a 3052 # cleaner UX than a partial fragment when the "Load profile" form 3053 # (already on the page) needs to reflect the new list too. 3054 3055 @app.post("/profiles/create", response_class=HTMLResponse) 3056 async def profiles_create( 3057 request: Request, 3058 name: str = Form(...), 3059 ) -> Any: 3060 from fastapi.responses import RedirectResponse # noqa: PLC0415 3061 try: 3062 await _dispatch(handle, type="configProfile", method="put", 3063 data={"name": name}) 3064 # Rebuild profile_registry from disk so subsequent calls see 3065 # the new entry. The in-memory registry is what we already 3066 # mutated, so this is optional — but paranoid safety. 3067 except HTTPException as exc: 3068 # Render home with an error banner instead of a bare 4xx page. 3069 return await _home_with_error(request, exc.detail) 3070 return RedirectResponse(url="/", status_code=303) 3071 3072 @app.post("/profiles/{profile_id}/delete", response_class=HTMLResponse) 3073 async def profiles_delete(request: Request, profile_id: str) -> Any: 3074 from fastapi.responses import RedirectResponse # noqa: PLC0415 3075 try: 3076 await _dispatch(handle, type="configProfile", method="delete", 3077 data={"id": profile_id}) 3078 except HTTPException as exc: 3079 return await _home_with_error(request, exc.detail) 3080 return RedirectResponse(url="/", status_code=303) 3081 3082 async def _home_with_error(request: Request, message: str) -> Any: 3083 """Re-render the home page carrying an error banner. Used by 3084 the profile-CRUD routes when the JSON service rejects a 3085 request (duplicate name, active profile delete, etc.).""" 3086 power = await _dispatch(handle, type="power", method="get") 3087 fast_clock = await _dispatch(handle, type="time", method="get") 3088 registry = getattr(handle.ctx, "profile_registry", None) 3089 active = registry.active() if registry is not None else None 3090 log_tail_ctx = await _log_tail_context(handle) 3091 return templates.TemplateResponse( 3092 request, "home.html", 3093 _base_context(handle, "home") | { 3094 "power": power, 3095 "fast_clock": fast_clock, 3096 "connection": handle.ctx.connection, 3097 "command_station_on": bool( 3098 getattr(handle.ctx.connection, 3099 "is_command_station_on", False) 3100 ), 3101 "topics": _topics_with_help(handle), 3102 "profiles": [ 3103 {"id": p.id, "name": p.name, "path": p.path} 3104 for p in (registry.profiles if registry is not None else []) 3105 ], 3106 "active_profile_id": active.id if active else None, 3107 "settings_root": ( 3108 str(registry.settings_root) if registry is not None else None 3109 ), 3110 "profile_error": message, 3111 } | log_tail_ctx, 3112 ) 3113 3114 # ----- cross-profile bean transfer -------------------------------------- 3115 3116 def _resolve_profile_name(profile_id: str) -> str: 3117 """Look up the friendly name for a profile id via the 3118 registry. Falls back to the raw id when the registry hasn't 3119 been loaded (bare test setups).""" 3120 registry = getattr(handle.ctx, "profile_registry", None) 3121 if registry is None: 3122 return profile_id 3123 ref = registry.by_id(profile_id) 3124 return ref.name if ref is not None else profile_id 3125 3126 def _target_is_active(profile_id: str) -> bool: 3127 registry = getattr(handle.ctx, "profile_registry", None) 3128 if registry is None: 3129 return False 3130 active = registry.active() 3131 return active is not None and active.id == profile_id 3132 3133 @app.post("/profiles/transfer/preview", response_class=HTMLResponse) 3134 async def profiles_transfer_preview( 3135 request: Request, 3136 from_id: str = Form(...), 3137 to_id: str = Form(...), 3138 types: list[str] = Form(default=[]), 3139 ) -> Any: 3140 try: 3141 data = await _dispatch(handle, type="beanTransfer", method="post", data={ 3142 "action": "preview", 3143 "from": from_id, "to": to_id, "types": types, 3144 }) 3145 except HTTPException as exc: 3146 return templates.TemplateResponse( 3147 request, "_bean_transfer_preview.html", 3148 {"error": exc.detail}, 3149 ) 3150 return templates.TemplateResponse( 3151 request, "_bean_transfer_preview.html", 3152 { 3153 "counts": data.get("counts", {}), 3154 "will_overwrite": data.get("willOverwrite", {}), 3155 "from_id": from_id, 3156 "to_id": to_id, 3157 "from_name": _resolve_profile_name(from_id), 3158 "to_name": _resolve_profile_name(to_id), 3159 "to_is_active": _target_is_active(to_id), 3160 "types": types, 3161 }, 3162 ) 3163 3164 @app.post("/profiles/transfer/apply", response_class=HTMLResponse) 3165 async def profiles_transfer_apply( 3166 request: Request, 3167 from_id: str = Form(...), 3168 to_id: str = Form(...), 3169 types: list[str] = Form(default=[]), 3170 ) -> Any: 3171 try: 3172 data = await _dispatch(handle, type="beanTransfer", method="post", data={ 3173 "action": "apply", 3174 "from": from_id, "to": to_id, "types": types, 3175 }) 3176 except HTTPException as exc: 3177 return templates.TemplateResponse( 3178 request, "_bean_transfer_applied.html", 3179 {"error": exc.detail}, 3180 ) 3181 return templates.TemplateResponse( 3182 request, "_bean_transfer_applied.html", 3183 { 3184 "copied": data.get("copied", {}), 3185 "target_is_active": data.get("targetIsActive", False), 3186 }, 3187 ) 3188 3189 # ----- signal systems ---------------------------------------------------- 3190 3191 @app.get("/signalsystems", response_class=HTMLResponse) 3192 async def signal_systems_home(request: Request) -> Any: 3193 systems = await _dispatch( 3194 handle, type="signalSystem", method="list", 3195 ) 3196 return templates.TemplateResponse( 3197 request, "signalsystems.html", 3198 _base_context(handle, "signalsystems") | {"systems": systems}, 3199 ) 3200 3201 # ----- logs widget ------------------------------------------------------- 3202 3203 _LOG_TAIL_MAX = 20 3204 3205 async def _log_tail_context(_handle: "AppHandle") -> dict[str, Any]: 3206 """Build the recent-log widget's template context. 3207 3208 Fetches via the ``logs`` service so the widget and any script 3209 or mobile client see byte-for-byte the same tail — no chance 3210 of divergent formatting between the widget and the JSON API. 3211 """ 3212 try: 3213 snapshot = await _dispatch( 3214 handle, type="logs", method="get", 3215 data={"tail": _LOG_TAIL_MAX}, 3216 ) 3217 except HTTPException: 3218 snapshot = {"path": None, "tail": _LOG_TAIL_MAX, "lines": []} 3219 return { 3220 "log_path": snapshot.get("path"), 3221 "log_lines": snapshot.get("lines") or [], 3222 "log_tail_max": snapshot.get("tail", _LOG_TAIL_MAX), 3223 } 3224 3225 @app.get("/logs/refresh", response_class=HTMLResponse) 3226 async def logs_refresh(request: Request) -> Any: 3227 ctx = await _log_tail_context(handle) 3228 return templates.TemplateResponse( 3229 request, "_log_tail.html", ctx, 3230 ) 3231 3232 # ----- logixng ----------------------------------------------------------- 3233 3234 async def _logixng_list_fragment(request: Request) -> Any: 3235 programs = await _dispatch(handle, type="logixng", method="list") 3236 return templates.TemplateResponse( 3237 request, "_logixng_list.html", {"programs": programs}, 3238 ) 3239 3240 @app.get("/logixng", response_class=HTMLResponse) 3241 async def logixng_home(request: Request) -> Any: 3242 programs = await _dispatch(handle, type="logixng", method="list") 3243 return templates.TemplateResponse( 3244 request, "logixng.html", 3245 _base_context(handle, "logixng") | {"programs": programs}, 3246 ) 3247 3248 @app.get("/logixng/refresh", response_class=HTMLResponse) 3249 async def logixng_refresh(request: Request) -> Any: 3250 return await _logixng_list_fragment(request) 3251 3252 @app.post("/logixng/create", response_class=HTMLResponse) 3253 async def logixng_create( 3254 request: Request, 3255 name: str = Form(...), 3256 userName: str = Form(""), 3257 ) -> Any: 3258 try: 3259 await _dispatch(handle, type="logixng", method="put", data={ 3260 "name": name, "userName": userName, 3261 }) 3262 except HTTPException as exc: 3263 programs = await _dispatch( 3264 handle, type="logixng", method="list", 3265 ) 3266 return templates.TemplateResponse( 3267 request, "_logixng_list.html", 3268 {"programs": programs, "create_error": exc.detail}, 3269 status_code=200, 3270 ) 3271 return await _logixng_list_fragment(request) 3272 3273 @app.post("/logixng/import-xml", response_class=HTMLResponse) 3274 async def logixng_import_xml( 3275 request: Request, file: UploadFile, 3276 ) -> Any: 3277 """Convert a JMRI Java LogixNG XML file into our tree 3278 format and load every program into the running manager. 3279 3280 Bypasses the JSON put path so we can (a) skip re-encoding 3281 each program to a dict and back, and (b) collect 3282 importer warnings for surfacing to the operator. Each 3283 program is persisted so a server restart doesn't lose the 3284 import.""" 3285 from jmri_logixng import import_jmri_logixng_xml # noqa: PLC0415 3286 from jmri_server.services.logixng import _require_manager # noqa: PLC0415 3287 3288 raw = await file.read() 3289 if not raw: 3290 return templates.TemplateResponse( 3291 request, "_logixng_import_result.html", 3292 {"error": "empty file"}, 3293 status_code=400, 3294 ) 3295 try: 3296 result = import_jmri_logixng_xml(raw) 3297 except Exception as e: # noqa: BLE001 3298 return templates.TemplateResponse( 3299 request, "_logixng_import_result.html", 3300 {"error": f"failed to parse XML: {e}"}, 3301 status_code=400, 3302 ) 3303 3304 # Install each imported program under its JMRI system name 3305 # (the ``node_id`` we recorded during walk_programs). Persist 3306 # to the profile's logixng dir so the import survives a 3307 # restart — same behaviour as the JSON put path. 3308 try: 3309 mgr = _require_manager(handle.ctx) 3310 except HTTPException as exc: 3311 return templates.TemplateResponse( 3312 request, "_logixng_import_result.html", 3313 {"error": f"manager unavailable: {exc.detail}"}, 3314 status_code=503, 3315 ) 3316 logixng_dir = getattr(handle.ctx, "_logixng_dir", None) 3317 imported_names: list[str] = [] 3318 for program in result.programs: 3319 # Node stores the constructor's ``node_id`` on ``.id`` 3320 # (with a uuid-suffixed fallback when the caller 3321 # didn't supply one). Prefer that, fall back to 3322 # user_name for the rare case the imported program 3323 # had neither. 3324 raw_name = program.id or program.user_name or "imported" 3325 # Colons are legal in JMRI Java system names (``IQ:AUTO:0001``) 3326 # but not in Windows filenames — replace with dots so the 3327 # persisted YAML file is portable. On Linux/Pi the raw 3328 # name works fine; keeping the sanitization uniform 3329 # simplifies cross-platform testing. 3330 name = raw_name.replace(":", ".") 3331 await mgr.put(name, program) 3332 imported_names.append(name) 3333 if logixng_dir is not None: 3334 mgr.save_one(logixng_dir, name) 3335 3336 # Load programs list for the out-of-band swap. 3337 programs = await _dispatch( 3338 handle, request, type="logixng", method="list", 3339 ) 3340 return templates.TemplateResponse( 3341 request, "_logixng_import_result.html", 3342 { 3343 "imported": imported_names, 3344 "warnings": result.warnings, 3345 "programs": programs, 3346 }, 3347 ) 3348 3349 @app.post("/logixng/{name}/delete", response_class=HTMLResponse) 3350 async def logixng_delete(request: Request, name: str) -> Any: 3351 try: 3352 await _dispatch( 3353 handle, type="logixng", method="delete", data={"name": name}, 3354 ) 3355 except HTTPException: 3356 pass 3357 return await _logixng_list_fragment(request) 3358 3359 @app.post("/logixng/{name}/enable", response_class=HTMLResponse) 3360 async def logixng_enable(request: Request, name: str) -> Any: 3361 await _dispatch(handle, type="logixng", method="post", data={ 3362 "name": name, "command": "enable", 3363 }) 3364 return await _logixng_list_fragment(request) 3365 3366 @app.post("/logixng/{name}/disable", response_class=HTMLResponse) 3367 async def logixng_disable(request: Request, name: str) -> Any: 3368 await _dispatch(handle, type="logixng", method="post", data={ 3369 "name": name, "command": "disable", 3370 }) 3371 return await _logixng_list_fragment(request) 3372 3373 @app.post("/logixng/{name}/run", response_class=HTMLResponse) 3374 async def logixng_run(request: Request, name: str) -> Any: 3375 try: 3376 await _dispatch(handle, type="logixng", method="post", data={ 3377 "name": name, "command": "run", 3378 }) 3379 except HTTPException as exc: 3380 programs = await _dispatch( 3381 handle, type="logixng", method="list", 3382 ) 3383 return templates.TemplateResponse( 3384 request, "_logixng_list.html", 3385 {"programs": programs, "run_error": exc.detail}, 3386 status_code=200, 3387 ) 3388 return await _logixng_list_fragment(request) 3389 3390 @app.post("/logixng/{name}/run-conditional", response_class=HTMLResponse) 3391 async def logixng_run_conditional( 3392 request: Request, name: str, 3393 conditional_id: str = Form(...), 3394 ) -> Any: 3395 try: 3396 await _dispatch(handle, type="logixng", method="post", data={ 3397 "name": name, "command": "run_conditional", 3398 "conditional_id": conditional_id, 3399 }) 3400 except HTTPException as exc: 3401 programs = await _dispatch( 3402 handle, type="logixng", method="list", 3403 ) 3404 return templates.TemplateResponse( 3405 request, "_logixng_list.html", 3406 {"programs": programs, "run_error": exc.detail}, 3407 status_code=200, 3408 ) 3409 return await _logixng_list_fragment(request) 3410 3411 # ----- consists ---------------------------------------------------------- 3412 3413 async def _consist_table_fragment(request: Request) -> Any: 3414 consists = await _dispatch(handle, type="consist", method="list") 3415 return templates.TemplateResponse( 3416 request, "_consists_table.html", {"consists": consists}, 3417 ) 3418 3419 @app.get("/consists", response_class=HTMLResponse) 3420 async def consists_home(request: Request) -> Any: 3421 consists = await _dispatch(handle, type="consist", method="list") 3422 return templates.TemplateResponse( 3423 request, "consists.html", 3424 _base_context(handle, "consists") | {"consists": consists}, 3425 ) 3426 3427 @app.get("/consists/refresh", response_class=HTMLResponse) 3428 async def consists_refresh(request: Request) -> Any: 3429 return await _consist_table_fragment(request) 3430 3431 @app.post("/consists/create", response_class=HTMLResponse) 3432 async def consists_create( 3433 request: Request, 3434 address: int = Form(...), 3435 kind: str = Form("command_station"), 3436 comment: str = Form(""), 3437 ) -> Any: 3438 # Consist addresses are short 1..127 by spec (CV19); the service 3439 # enforces range + roster-collision. Any error there comes back 3440 # as an HTTPException — render it into the notice so htmx sees a 3441 # 200 with the message rather than swapping in the FastAPI 3442 # default JSON error page. 3443 try: 3444 await _dispatch(handle, type="consist", method="put", data={ 3445 "address": address, "kind": kind, "comment": comment, 3446 }) 3447 except HTTPException as exc: 3448 return templates.TemplateResponse( 3449 request, "_consists_table.html", 3450 { 3451 "consists": await _dispatch( 3452 handle, type="consist", method="list" 3453 ), 3454 "create_error": exc.detail, 3455 }, 3456 status_code=200, 3457 ) 3458 return await _consist_table_fragment(request) 3459 3460 @app.post("/consists/{address}/delete", response_class=HTMLResponse) 3461 async def consists_delete(request: Request, address: int) -> Any: 3462 try: 3463 await _dispatch( 3464 handle, type="consist", method="delete", 3465 data={"address": address}, 3466 ) 3467 except HTTPException: 3468 pass # 404 — table refresh below will just show whatever's left. 3469 return await _consist_table_fragment(request) 3470 3471 @app.post("/consists/{address}/add-member", response_class=HTMLResponse) 3472 async def consists_add_member( 3473 request: Request, address: int, 3474 member_address: int = Form(...), 3475 long_address: str | None = Form(None), 3476 reversed: str | None = Form(None), 3477 ) -> Any: 3478 # Form has a "Reversed" checkbox — unchecked (the default) means 3479 # the loco is oriented forward in the consist. 3480 await _dispatch(handle, type="consist", method="post", data={ 3481 "address": address, "action": "add_member", 3482 "member": { 3483 "address": member_address, 3484 "long_address": bool(long_address), 3485 "direction_normal": not bool(reversed), 3486 }, 3487 }) 3488 return await _consist_table_fragment(request) 3489 3490 @app.post("/consists/{address}/remove-member", response_class=HTMLResponse) 3491 async def consists_remove_member( 3492 request: Request, address: int, 3493 member_address: int = Form(...), 3494 ) -> Any: 3495 try: 3496 await _dispatch(handle, type="consist", method="post", data={ 3497 "address": address, "action": "remove_member", 3498 "member_address": member_address, 3499 }) 3500 except HTTPException: 3501 pass 3502 return await _consist_table_fragment(request) 3503 3504 @app.post("/consists/{address}/set-speed", response_class=HTMLResponse) 3505 async def consists_set_speed( 3506 request: Request, address: int, speed: float = Form(...), 3507 ) -> Any: 3508 await _dispatch(handle, type="consist", method="post", data={ 3509 "address": address, "action": "set_speed", "speed": speed, 3510 }) 3511 return await _consist_table_fragment(request) 3512 3513 @app.post("/consists/{address}/direction", response_class=HTMLResponse) 3514 async def consists_set_direction( 3515 request: Request, address: int, direction: str = Form(...), 3516 ) -> Any: 3517 await _dispatch(handle, type="consist", method="post", data={ 3518 "address": address, "action": "set_direction", "direction": direction, 3519 }) 3520 return await _consist_table_fragment(request) 3521 3522 @app.post("/consists/{address}/estop", response_class=HTMLResponse) 3523 async def consists_estop(request: Request, address: int) -> Any: 3524 await _dispatch(handle, type="consist", method="post", data={ 3525 "address": address, "action": "emergency_stop", 3526 }) 3527 return await _consist_table_fragment(request) 3528 3529 @app.post("/panel/upload", response_class=HTMLResponse) 3530 async def panel_upload(request: Request, file: UploadFile) -> Any: 3531 from jmri_core.persistence import load_panel_bytes # noqa: PLC0415 3532 3533 data = await file.read() 3534 try: 3535 panel_file = load_panel_bytes(data) 3536 except Exception as e: # noqa: BLE001 3537 return templates.TemplateResponse( 3538 request, "_panel_summary.html", 3539 {"error": f"failed to parse: {e}"}, 3540 status_code=400, 3541 ) 3542 summary = { 3543 "filename": file.filename, 3544 "root_tag": panel_file.root_tag, 3545 "jmri_version": panel_file.jmri_version, 3546 "panels": [p.get("name") or "(unnamed)" for p in panel_file.panels()], 3547 "layout_editors": [ 3548 le.get("name") or "(unnamed)" for le in panel_file.layout_editors() 3549 ], 3550 "top_level_child_count": len(list(panel_file.root)), 3551 } 3552 return templates.TemplateResponse( 3553 request, "_panel_summary.html", {"summary": summary}, 3554 )
def
install_template_routes( app: fastapi.applications.FastAPI, handle: jmri_server.app.AppHandle, templates: starlette.templating.Jinja2Templates) -> None:
100def install_template_routes( 101 app: FastAPI, handle: AppHandle, templates: Jinja2Templates 102) -> None: 103 # ---- auth: /login + /logout ---------------------------------------- 104 # These routes only do something meaningful when 105 # ``JMRI_AUTH_ENABLED=1``. With auth off, ``ctx.auth`` is None and 106 # /login shows a friendly "auth is disabled" message rather than 107 # a form that couldn't do anything anyway. 108 109 from jmri_server.auth.middleware import SESSION_COOKIE # noqa: PLC0415 110 111 def _csrf_token(request: Request) -> str: 112 """Get-or-set a CSRF token in the ``JMRICSRF`` cookie. 113 114 Double-submit-cookie CSRF pattern: the same random token is 115 embedded in the form AND in a cookie; on POST the server 116 checks they match. A cross-site attacker can't read the 117 cookie (SameSite=Lax) so can't forge a matching form field. 118 """ 119 existing = request.cookies.get("JMRICSRF") 120 if existing: 121 return existing 122 import secrets # noqa: PLC0415 123 return secrets.token_urlsafe(24) 124 125 def _verify_csrf(request: Request, submitted: str) -> bool: 126 cookie = request.cookies.get("JMRICSRF", "") 127 if not cookie or not submitted: 128 return False 129 import hmac # noqa: PLC0415 130 return hmac.compare_digest(cookie, submitted) 131 132 @app.get("/login", response_class=HTMLResponse) 133 async def login_page( 134 request: Request, next: str = "/", 135 ) -> Any: # type: ignore[valid-type] 136 auth = getattr(handle.ctx, "auth", None) 137 error = None 138 if auth is None: 139 error = "Auth is disabled on this server (JMRI_AUTH_ENABLED is unset). Every request runs as admin." 140 token = _csrf_token(request) 141 response = templates.TemplateResponse( 142 request, "login.html", 143 _base_context(handle, "login") | { 144 "csrf_token": token, 145 "next": next, 146 "error": error, 147 }, 148 ) 149 # SameSite=Lax so cross-site GETs don't leak the token; the 150 # form POST is same-origin so it lines up. 151 response.set_cookie( 152 "JMRICSRF", token, samesite="lax", 153 httponly=False, max_age=3600, 154 ) 155 return response 156 157 @app.post("/login", response_class=HTMLResponse) 158 async def login_submit( 159 request: Request, 160 username: str = Form(...), 161 password: str = Form(...), 162 next: str = Form("/"), 163 csrf: str = Form("", alias="_csrf"), 164 ) -> Any: # type: ignore[valid-type] 165 from fastapi.responses import RedirectResponse # noqa: PLC0415 166 auth = getattr(handle.ctx, "auth", None) 167 if auth is None: 168 # Auth off — no login needed; just bounce to home. 169 return RedirectResponse(url=next or "/", status_code=303) 170 if not _verify_csrf(request, csrf): 171 token = _csrf_token(request) 172 response = templates.TemplateResponse( 173 request, "login.html", 174 _base_context(handle, "login") | { 175 "csrf_token": token, 176 "next": next, 177 "error": "Session expired — try again.", 178 }, 179 status_code=400, 180 ) 181 response.set_cookie( 182 "JMRICSRF", token, samesite="lax", 183 httponly=False, max_age=3600, 184 ) 185 return response 186 result = auth.login_password(username, password) 187 if result is None: 188 token = _csrf_token(request) 189 response = templates.TemplateResponse( 190 request, "login.html", 191 _base_context(handle, "login") | { 192 "csrf_token": token, 193 "next": next, 194 "error": "Invalid username or password.", 195 }, 196 status_code=401, 197 ) 198 response.set_cookie( 199 "JMRICSRF", token, samesite="lax", 200 httponly=False, max_age=3600, 201 ) 202 return response 203 response = RedirectResponse(url=next or "/", status_code=303) 204 cookie_value = auth.mint_session_cookie(result.user) 205 # httponly so JS can't touch it (defends against XSS 206 # session theft). secure=False for now since we run on 207 # plain HTTP LANs; flip to True when we grow HTTPS. 208 response.set_cookie( 209 SESSION_COOKIE, cookie_value, 210 httponly=True, samesite="lax", 211 max_age=7 * 24 * 3600, 212 ) 213 return response 214 215 @app.post("/logout") 216 async def logout(request: Request) -> Any: # type: ignore[valid-type] 217 from fastapi.responses import RedirectResponse # noqa: PLC0415 218 response = RedirectResponse(url="/login", status_code=303) 219 response.delete_cookie(SESSION_COOKIE) 220 return response 221 222 # ---- /users admin + self-service ----------------------------------- 223 224 async def _render_users( 225 request: Request, 226 *, 227 notice: str | None = None, 228 new_token: str | None = None, 229 new_token_user: str | None = None, 230 ) -> Any: 231 """Fetch the user list (as admin, if the caller is admin) and 232 render the /users template. Non-admin callers see an empty 233 list — the template hides the admin sections in that case.""" 234 from jmri_server.auth.models import Role # noqa: PLC0415 235 role = getattr(request.state, "role", Role.ADMIN) 236 is_admin = role is Role.ADMIN 237 users: list[dict[str, Any]] = [] 238 if is_admin: 239 try: 240 users = await _dispatch( 241 handle, request, type="configUser", method="list", 242 ) 243 except HTTPException: 244 users = [] 245 token = _csrf_token(request) 246 response = templates.TemplateResponse( 247 request, "users.html", 248 _base_context(handle, "users") | { 249 "users": users, 250 "is_admin": is_admin, 251 "csrf_token": token, 252 "notice": notice, 253 "new_token": new_token, 254 "new_token_user": new_token_user, 255 }, 256 ) 257 response.set_cookie( 258 "JMRICSRF", token, samesite="lax", 259 httponly=False, max_age=3600, 260 ) 261 return response 262 263 @app.get("/users", response_class=HTMLResponse) 264 async def users_home(request: Request) -> Any: # type: ignore[valid-type] 265 return await _render_users(request) 266 267 @app.post("/users/create", response_class=HTMLResponse) 268 async def users_create( 269 request: Request, 270 username: str = Form(...), 271 password: str = Form(...), 272 role: str = Form("operator"), 273 csrf: str = Form("", alias="_csrf"), 274 ) -> Any: # type: ignore[valid-type] 275 if not _verify_csrf(request, csrf): 276 return await _render_users(request, notice="Session expired — try again.") 277 try: 278 await _dispatch( 279 handle, request, type="configUser", method="put", 280 data={"username": username, "password": password, "role": role}, 281 ) 282 except HTTPException as exc: 283 return await _render_users(request, notice=f"Create failed: {exc.detail}") 284 return await _render_users(request, notice=f"Created user {username}.") 285 286 # ---- self-service routes register FIRST so the /users/{username} 287 # ---- parameterized routes below don't shadow /users/self/* with 288 # ---- username="self". Starlette matches routes in registration 289 # ---- order; the more specific paths must come first. 290 291 @app.post("/users/self/change-password", response_class=HTMLResponse) 292 async def users_self_change_password_early( 293 request: Request, 294 new_password: str = Form(...), 295 csrf: str = Form("", alias="_csrf"), 296 ) -> Any: # type: ignore[valid-type] 297 if not _verify_csrf(request, csrf): 298 return await _render_users(request, notice="Session expired.") 299 user = getattr(request.state, "user", None) 300 if user is None: 301 return await _render_users( 302 request, notice="Not logged in — password change unavailable.", 303 ) 304 try: 305 await _dispatch( 306 handle, request, type="configUser", method="post", 307 data={ 308 "action": "change_password", 309 "username": user.username, 310 "new_password": new_password, 311 }, 312 ) 313 except HTTPException as exc: 314 return await _render_users( 315 request, notice=f"Password change failed: {exc.detail}", 316 ) 317 return await _render_users( 318 request, notice="Password updated.", 319 ) 320 321 @app.post("/users/self/regenerate-token", response_class=HTMLResponse) 322 async def users_self_regen_token_early( 323 request: Request, 324 csrf: str = Form("", alias="_csrf"), 325 ) -> Any: # type: ignore[valid-type] 326 if not _verify_csrf(request, csrf): 327 return await _render_users(request, notice="Session expired.") 328 user = getattr(request.state, "user", None) 329 if user is None: 330 return await _render_users( 331 request, notice="Not logged in — token regenerate unavailable.", 332 ) 333 try: 334 resp = await _dispatch( 335 handle, request, type="configUser", method="post", 336 data={ 337 "action": "regenerate_token", 338 "username": user.username, 339 }, 340 ) 341 except HTTPException as exc: 342 return await _render_users( 343 request, notice=f"Regenerate failed: {exc.detail}", 344 ) 345 return await _render_users( 346 request, 347 new_token=resp["token"], 348 new_token_user=user.username, 349 ) 350 351 @app.post("/users/{username}/delete", response_class=HTMLResponse) 352 async def users_delete( 353 request: Request, 354 username: str, 355 csrf: str = Form("", alias="_csrf"), 356 ) -> Any: # type: ignore[valid-type] 357 if not _verify_csrf(request, csrf): 358 return await _render_users(request, notice="Session expired.") 359 try: 360 await _dispatch( 361 handle, request, type="configUser", method="delete", 362 data={"username": username}, 363 ) 364 except HTTPException as exc: 365 return await _render_users(request, notice=f"Delete failed: {exc.detail}") 366 return await _render_users(request, notice=f"Deleted user {username}.") 367 368 @app.post("/users/{username}/set-role", response_class=HTMLResponse) 369 async def users_set_role( 370 request: Request, 371 username: str, 372 role: str = Form(...), 373 csrf: str = Form("", alias="_csrf"), 374 ) -> Any: # type: ignore[valid-type] 375 if not _verify_csrf(request, csrf): 376 return await _render_users(request, notice="Session expired.") 377 try: 378 await _dispatch( 379 handle, request, type="configUser", method="put", 380 data={"username": username, "role": role}, 381 ) 382 except HTTPException as exc: 383 return await _render_users( 384 request, notice=f"Role change failed: {exc.detail}", 385 ) 386 return await _render_users( 387 request, notice=f"{username} → {role}", 388 ) 389 390 @app.post("/users/{username}/regenerate-token", response_class=HTMLResponse) 391 async def users_regen_token( 392 request: Request, 393 username: str, 394 csrf: str = Form("", alias="_csrf"), 395 ) -> Any: # type: ignore[valid-type] 396 if not _verify_csrf(request, csrf): 397 return await _render_users(request, notice="Session expired.") 398 try: 399 resp = await _dispatch( 400 handle, request, type="configUser", method="post", 401 data={"action": "regenerate_token", "username": username}, 402 ) 403 except HTTPException as exc: 404 return await _render_users( 405 request, notice=f"Regenerate failed: {exc.detail}", 406 ) 407 return await _render_users( 408 request, 409 new_token=resp["token"], 410 new_token_user=username, 411 ) 412 413 # The decoder catalog is used by both the roster page (to gate the 414 # Program button) and the programmer routes further down. Kept on 415 # ``handle.ctx._decoder_catalog`` so tests / other subsystems can 416 # populate the catalog directly. Routes fetch it via ``_decoder_catalog`` 417 # on every request so a runtime swap is honoured. 418 from jmri_decoderpro import DecoderCatalog # noqa: PLC0415 419 420 if not hasattr(handle.ctx, "_decoder_catalog"): 421 handle.ctx._decoder_catalog = DecoderCatalog() # type: ignore[attr-defined] 422 423 def _catalog() -> DecoderCatalog: 424 return handle.ctx._decoder_catalog # type: ignore[attr-defined,no-any-return] 425 426 @app.get("/", response_class=HTMLResponse) 427 async def home(request: Request) -> Any: 428 # Diagnostic timing — turned on with ``JMRI_TIME_HOME=1``. Logs 429 # per-dispatch and total render time at WARNING so it shows up 430 # in ``uvicorn.log`` without cranking log verbosity globally. 431 # Off by default so production users see no extra chatter. 432 import time as _time # noqa: PLC0415 433 from jmri_core.env import time_home as _time_home # noqa: PLC0415 434 _diag = _time_home() 435 _t0 = _time.monotonic() if _diag else 0.0 436 power = await _dispatch(handle, type="power", method="get") 437 _t1 = _time.monotonic() if _diag else 0.0 438 fast_clock = await _dispatch(handle, type="time", method="get") 439 _t2 = _time.monotonic() if _diag else 0.0 440 if _diag: 441 import logging as _logging # noqa: PLC0415 442 _logging.getLogger("jmri_web.timing").warning( 443 "GET / dispatch power=%.0fms time=%.0fms", 444 (_t1 - _t0) * 1000, (_t2 - _t1) * 1000, 445 ) 446 registry = getattr(handle.ctx, "profile_registry", None) 447 profiles: list[dict[str, Any]] = [] 448 active_id: str | None = None 449 settings_root_str: str | None = None 450 if registry is not None: 451 active = registry.active() 452 active_id = active.id if active else None 453 profiles = [ 454 {"id": p.id, "name": p.name, "path": p.path} 455 for p in registry.profiles 456 ] 457 settings_root_str = str(registry.settings_root) 458 _t3 = _time.monotonic() if _diag else 0.0 459 log_tail_ctx = await _log_tail_context(handle) 460 resp = templates.TemplateResponse( 461 request, "home.html", 462 _base_context(handle, "home") | { 463 "power": power, 464 "fast_clock": fast_clock, 465 "connection": handle.ctx.connection, 466 "command_station_on": bool( 467 getattr(handle.ctx.connection, 468 "is_command_station_on", False) 469 ), 470 "topics": _topics_with_help(handle), 471 "profiles": profiles, 472 "active_profile_id": active_id, 473 "settings_root": settings_root_str, 474 } | log_tail_ctx, 475 ) 476 if _diag: 477 _t4 = _time.monotonic() 478 import logging as _logging # noqa: PLC0415 479 _logging.getLogger("jmri_web.timing").warning( 480 "GET / render ctx=%.0fms tmpl=%.0fms TOTAL=%.0fms", 481 (_t3 - _t2) * 1000, (_t4 - _t3) * 1000, 482 (_t4 - _t0) * 1000, 483 ) 484 return resp 485 486 @app.post("/power/set", response_class=HTMLResponse) 487 async def set_power(request: Request, state: int = Form(...)) -> Any: 488 data = await _dispatch(handle, type="power", method="post", data={"state": state}) 489 return templates.TemplateResponse( 490 request, "_power_panel.html", {"power": data}, 491 ) 492 493 @app.post("/profiles/load", response_class=HTMLResponse) 494 async def load_profile(request: Request, profile_id: str = Form(...)) -> Any: 495 """Switch the active profile. Clears in-memory beans, re-points 496 ``ConfigStore`` and the roster store, repopulates from the new 497 profile's ``PanelPro.xml``. Hardware connection is untouched.""" 498 from jmri_server.app import switch_active_profile # noqa: PLC0415 499 # Diagnostic timing — opt in with ``JMRI_TIME_HOME=1``. Logs 500 # per-phase durations at WARNING so they surface in uvicorn.log 501 # without dialing up global log verbosity. Off by default. 502 import logging as _logging # noqa: PLC0415 503 import time as _time # noqa: PLC0415 504 from jmri_core.env import time_home as _time_home # noqa: PLC0415 505 _diag = _time_home() 506 _log = _logging.getLogger("jmri_web.timing") 507 _t0 = _time.monotonic() if _diag else 0.0 508 try: 509 await switch_active_profile(handle, profile_id) 510 except (KeyError, RuntimeError) as e: 511 raise HTTPException(status_code=400, detail=str(e)) from None 512 _t1 = _time.monotonic() if _diag else 0.0 513 # Re-render the home page with the new active profile. 514 power = await _dispatch(handle, type="power", method="get") 515 _t2 = _time.monotonic() if _diag else 0.0 516 fast_clock = await _dispatch(handle, type="time", method="get") 517 _t3 = _time.monotonic() if _diag else 0.0 518 registry = getattr(handle.ctx, "profile_registry", None) 519 active = registry.active() if registry is not None else None 520 if _diag: 521 _log.warning( 522 "POST /profiles/load switch=%.0fms power=%.0fms time=%.0fms", 523 (_t1 - _t0) * 1000, (_t2 - _t1) * 1000, (_t3 - _t2) * 1000, 524 ) 525 return templates.TemplateResponse( 526 request, "home.html", 527 _base_context(handle, "home") | { 528 "power": power, 529 "fast_clock": fast_clock, 530 "connection": handle.ctx.connection, 531 "command_station_on": bool( 532 getattr(handle.ctx.connection, 533 "is_command_station_on", False) 534 ), 535 "topics": _topics_with_help(handle), 536 "profiles": [ 537 {"id": p.id, "name": p.name, "path": p.path} 538 for p in (registry.profiles if registry is not None else []) 539 ], 540 "active_profile_id": active.id if active else None, 541 "load_notice": f"Loaded profile: {active.name}" if active else None, 542 }, 543 ) 544 545 # ----- throttle --------------------------------------------------------- 546 547 @app.get("/throttle", response_class=HTMLResponse) 548 async def throttle( 549 request: Request, 550 address: int | None = None, 551 long: int | None = None, 552 ) -> Any: 553 """Throttle page. Without query args, returns to the most recently 554 acquired throttle (single-user alpha; ``ctx._last_throttle`` stores 555 ``(address, long_address)``). Falls back to address 3 if no history. 556 """ 557 last = getattr(handle.ctx, "_last_throttle", None) 558 if address is None: 559 if isinstance(last, tuple) and len(last) == 2: 560 address, last_long = int(last[0]), bool(last[1]) 561 if long is None: 562 long = 1 if last_long else 0 563 else: 564 address = 3 565 # NMRA default: addresses >= 128 are long unless explicitly overridden. 566 long_address = (address >= 128) if long is None else bool(long) 567 data = await _dispatch( 568 handle, type="throttle", method="put", 569 data={ 570 "address": address, 571 "name": str(address), 572 "isLongAddress": long_address, 573 }, 574 ) 575 # Remember this acquisition so the next visit lands here. 576 handle.ctx._last_throttle = (address, long_address) # type: ignore[attr-defined] 577 # Surface roster entries so the page can build a picker. 578 roster = await _dispatch(handle, type="roster", method="list") 579 return templates.TemplateResponse( 580 request, "throttle.html", 581 _base_context(handle, "throttle") | { 582 "address": address, 583 "long_address": long_address, 584 "throttle": data, 585 "roster": roster, 586 }, 587 ) 588 589 @app.post("/throttle/{address}/state", response_class=HTMLResponse) 590 async def throttle_set_state( 591 request: Request, address: int, 592 speed: float = Form(None), 593 forward: int = Form(None), 594 ) -> Any: 595 payload: dict[str, Any] = {"name": str(address)} 596 if speed is not None: 597 payload["speed"] = speed 598 if forward is not None: 599 payload["forward"] = bool(forward) 600 data = await _dispatch(handle, type="throttle", method="post", data=payload) 601 return templates.TemplateResponse( 602 request, "_throttle_status.html", {"throttle": data}, 603 ) 604 605 @app.post("/throttle/{address}/function/{index}", response_class=HTMLResponse) 606 async def throttle_set_function( 607 request: Request, address: int, index: int, on: int = Form(...), 608 ) -> Any: 609 payload = {"name": str(address), f"F{index}": bool(on)} 610 data = await _dispatch(handle, type="throttle", method="post", data=payload) 611 return templates.TemplateResponse( 612 request, "_throttle_status.html", {"throttle": data}, 613 ) 614 615 @app.post("/throttle/{address}/release", response_class=HTMLResponse) 616 async def throttle_release(request: Request, address: int) -> Any: 617 await _dispatch( 618 handle, type="throttle", method="delete", data={"name": str(address)}, 619 ) 620 return templates.TemplateResponse( 621 request, "_throttle_status.html", 622 {"throttle": {"name": str(address), "released": True}}, 623 ) 624 625 # ----- turnouts --------------------------------------------------------- 626 627 @app.get("/turnouts", response_class=HTMLResponse) 628 async def turnouts(request: Request) -> Any: 629 items = await _dispatch(handle, type="turnout", method="list") 630 return templates.TemplateResponse( 631 request, "turnouts.html", 632 _base_context(handle, "turnouts") | {"turnouts": items}, 633 ) 634 635 @app.post("/turnouts/create", response_class=HTMLResponse) 636 async def create_turnout(request: Request, name: str = Form(...)) -> Any: 637 await _dispatch(handle, type="turnout", method="put", data={"name": name}) 638 items = await _dispatch(handle, type="turnout", method="list") 639 return templates.TemplateResponse( 640 request, "_turnouts_table.html", {"turnouts": items}, 641 ) 642 643 @app.post("/turnouts/{name}/state", response_class=HTMLResponse) 644 async def set_turnout_state(request: Request, name: str, state: int = Form(...)) -> Any: 645 data = await _dispatch( 646 handle, type="turnout", method="post", 647 data={"name": name, "state": state}, 648 ) 649 return templates.TemplateResponse( 650 request, "_turnout_row.html", {"t": data}, 651 ) 652 653 @app.get("/turnouts/{name}/row", response_class=HTMLResponse) 654 async def turnout_row(request: Request, name: str) -> Any: 655 """Read-only row partial. Used by the Cancel button to revert an edit.""" 656 data = await _dispatch( 657 handle, type="turnout", method="get", data={"name": name}, 658 ) 659 return templates.TemplateResponse(request, "_turnout_row.html", {"t": data}) 660 661 @app.get("/turnouts/{name}/edit", response_class=HTMLResponse) 662 async def turnout_edit_row(request: Request, name: str) -> Any: 663 data = await _dispatch( 664 handle, type="turnout", method="get", data={"name": name}, 665 ) 666 return templates.TemplateResponse( 667 request, "_turnout_row_edit.html", {"t": data}, 668 ) 669 670 @app.post("/turnouts/{name}/update", response_class=HTMLResponse) 671 async def turnout_update( 672 request: Request, 673 name: str, 674 userName: str = Form(""), 675 inverted: int = Form(0), 676 ) -> Any: 677 payload: dict[str, Any] = { 678 "name": name, 679 "userName": userName or None, 680 "inverted": bool(inverted), 681 } 682 data = await _dispatch(handle, type="turnout", method="post", data=payload) 683 return templates.TemplateResponse(request, "_turnout_row.html", {"t": data}) 684 685 @app.delete("/turnouts/{name}", response_class=HTMLResponse) 686 async def turnout_delete(request: Request, name: str) -> Any: 687 await _dispatch(handle, type="turnout", method="delete", data={"name": name}) 688 # HTMX swap target is the row; returning empty deletes it from the DOM. 689 return HTMLResponse("") 690 691 # ----- sensors ---------------------------------------------------------- 692 693 @app.get("/sensors", response_class=HTMLResponse) 694 async def sensors(request: Request) -> Any: 695 items = await _dispatch(handle, type="sensor", method="list") 696 return templates.TemplateResponse( 697 request, "sensors.html", 698 _base_context(handle, "sensors") | {"sensors": items}, 699 ) 700 701 @app.post("/sensors/create", response_class=HTMLResponse) 702 async def create_sensor(request: Request, name: str = Form(...)) -> Any: 703 await _dispatch(handle, type="sensor", method="put", data={"name": name}) 704 items = await _dispatch(handle, type="sensor", method="list") 705 return templates.TemplateResponse( 706 request, "_sensors_table.html", {"sensors": items}, 707 ) 708 709 @app.post("/sensors/{name}/state", response_class=HTMLResponse) 710 async def set_sensor_state(request: Request, name: str, state: int = Form(...)) -> Any: 711 """For the internal-connection simulator: force a sensor active/inactive.""" 712 data = await _dispatch( 713 handle, type="sensor", method="post", 714 data={"name": name, "state": state}, 715 ) 716 return templates.TemplateResponse( 717 request, "_sensor_row.html", {"s": data}, 718 ) 719 720 @app.get("/sensors/{name}/row", response_class=HTMLResponse) 721 async def sensor_row(request: Request, name: str) -> Any: 722 data = await _dispatch( 723 handle, type="sensor", method="get", data={"name": name}, 724 ) 725 return templates.TemplateResponse(request, "_sensor_row.html", {"s": data}) 726 727 @app.get("/sensors/{name}/edit", response_class=HTMLResponse) 728 async def sensor_edit_row(request: Request, name: str) -> Any: 729 data = await _dispatch( 730 handle, type="sensor", method="get", data={"name": name}, 731 ) 732 return templates.TemplateResponse( 733 request, "_sensor_row_edit.html", {"s": data}, 734 ) 735 736 @app.post("/sensors/{name}/update", response_class=HTMLResponse) 737 async def sensor_update( 738 request: Request, 739 name: str, 740 userName: str = Form(""), 741 inverted: int = Form(0), 742 ) -> Any: 743 payload: dict[str, Any] = { 744 "name": name, 745 "userName": userName or None, 746 "inverted": bool(inverted), 747 } 748 data = await _dispatch(handle, type="sensor", method="post", data=payload) 749 return templates.TemplateResponse(request, "_sensor_row.html", {"s": data}) 750 751 @app.delete("/sensors/{name}", response_class=HTMLResponse) 752 async def sensor_delete(request: Request, name: str) -> Any: 753 await _dispatch(handle, type="sensor", method="delete", data={"name": name}) 754 return HTMLResponse("") 755 756 # ----- lights ----------------------------------------------------------- 757 758 @app.get("/lights", response_class=HTMLResponse) 759 async def lights(request: Request) -> Any: 760 items = await _dispatch(handle, type="light", method="list") 761 return templates.TemplateResponse( 762 request, "lights.html", 763 _base_context(handle, "lights") | {"lights": items}, 764 ) 765 766 @app.post("/lights/create", response_class=HTMLResponse) 767 async def create_light(request: Request, name: str = Form(...)) -> Any: 768 await _dispatch(handle, type="light", method="put", data={"name": name}) 769 items = await _dispatch(handle, type="light", method="list") 770 return templates.TemplateResponse( 771 request, "_lights_table.html", {"lights": items}, 772 ) 773 774 @app.post("/lights/{name}/state", response_class=HTMLResponse) 775 async def set_light_state(request: Request, name: str, state: int = Form(...)) -> Any: 776 data = await _dispatch( 777 handle, type="light", method="post", 778 data={"name": name, "state": state}, 779 ) 780 return templates.TemplateResponse( 781 request, "_light_row.html", {"l": data}, 782 ) 783 784 @app.get("/lights/{name}/row", response_class=HTMLResponse) 785 async def light_row(request: Request, name: str) -> Any: 786 data = await _dispatch( 787 handle, type="light", method="get", data={"name": name}, 788 ) 789 return templates.TemplateResponse(request, "_light_row.html", {"l": data}) 790 791 @app.get("/lights/{name}/edit", response_class=HTMLResponse) 792 async def light_edit_row(request: Request, name: str) -> Any: 793 data = await _dispatch( 794 handle, type="light", method="get", data={"name": name}, 795 ) 796 return templates.TemplateResponse( 797 request, "_light_row_edit.html", {"l": data}, 798 ) 799 800 @app.post("/lights/{name}/update", response_class=HTMLResponse) 801 async def light_update( 802 request: Request, 803 name: str, 804 userName: str = Form(""), 805 ) -> Any: 806 payload: dict[str, Any] = { 807 "name": name, 808 "userName": userName or None, 809 } 810 data = await _dispatch(handle, type="light", method="post", data=payload) 811 return templates.TemplateResponse(request, "_light_row.html", {"l": data}) 812 813 @app.delete("/lights/{name}", response_class=HTMLResponse) 814 async def light_delete(request: Request, name: str) -> Any: 815 await _dispatch(handle, type="light", method="delete", data={"name": name}) 816 return HTMLResponse("") 817 818 # ----- roster ----------------------------------------------------------- 819 820 async def _roster_view(request: Request, format: str | None = None) -> Any: 821 """Engine Driver fetches the roster as XML at ``/roster?format=xml``; 822 the web UI fetches the same data as HTML. Branch on the query.""" 823 if format == "xml": 824 # Serve the on-disk roster.xml verbatim if it exists; that's 825 # what JMRI's RosterServlet would emit. Engine Driver parses 826 # the ``<roster-config>`` shape directly. 827 profile = getattr(handle.ctx, "_roster_profile", None) 828 if profile is not None and profile.roster_index.exists(): 829 return Response( 830 content=profile.roster_index.read_bytes(), 831 media_type="application/xml", 832 ) 833 # No roster.xml yet — synthesize an empty one. 834 return Response( 835 content=b'<?xml version="1.0" encoding="UTF-8"?>\n' 836 b'<roster-config><roster/></roster-config>\n', 837 media_type="application/xml", 838 ) 839 items = await _dispatch(handle, type="roster", method="list") 840 return templates.TemplateResponse( 841 request, "roster.html", 842 _base_context(handle, "roster") | { 843 "roster": items, 844 "catalog_loaded": len(_catalog()) > 0, 845 # For the decoder-picker dropdown. Empty when catalog 846 # isn't loaded — the template falls back to a hint. 847 "manufacturers": _catalog().manufacturers(), 848 }, 849 ) 850 851 # Both the canonical and trailing-slash forms — Engine Driver issues 852 # ``GET /roster/?format=xml`` and won't follow the FastAPI 307 redirect 853 # that the default routing emits for a slash mismatch. 854 @app.get("/roster", response_class=HTMLResponse) 855 async def roster(request: Request, format: str | None = None) -> Any: 856 return await _roster_view(request, format) 857 858 @app.get("/roster/", response_class=HTMLResponse, include_in_schema=False) 859 async def roster_slashed(request: Request, format: str | None = None) -> Any: 860 return await _roster_view(request, format) 861 862 # ----- Engine Driver / WiThrottle side-channel HTTP endpoints ---------- 863 864 @app.get("/prefs/{app_name}/{file_name}", include_in_schema=False) 865 async def withrottle_prefs(app_name: str, file_name: str) -> Response: 866 """Engine Driver pulls/pushes its own preferences from the server. 867 868 JMRI's WiThrottle server stores these at ``<profile>/preferences/ 869 engine_driver/auto_preferences.ed`` and serves them back. For the 870 alpha we return an empty 200 instead of 404 so Engine Driver 871 doesn't panic — its later "save" will not be persisted, but the 872 connection stays healthy. 873 """ 874 profile = getattr(handle.ctx, "_roster_profile", None) 875 if profile is not None: 876 path = profile.preferences_dir / app_name / file_name 877 if path.exists(): 878 return Response( 879 content=path.read_bytes(), 880 media_type="application/octet-stream", 881 ) 882 return Response(content=b"", media_type="application/octet-stream") 883 884 @app.put("/prefs/{app_name}/{file_name}", include_in_schema=False) 885 async def withrottle_prefs_put( 886 request: Request, app_name: str, file_name: str, 887 ) -> Response: 888 """Accept (and persist) Engine Driver's saved preferences.""" 889 profile = getattr(handle.ctx, "_roster_profile", None) 890 body = await request.body() 891 if profile is not None: 892 target_dir = profile.preferences_dir / app_name 893 target_dir.mkdir(parents=True, exist_ok=True) 894 (target_dir / file_name).write_bytes(body) 895 return Response(status_code=204) 896 897 @app.post("/roster/create", response_class=HTMLResponse) 898 async def roster_create( 899 request: Request, 900 id: str = Form(...), 901 dccAddress: str = Form("3"), 902 longAddress: int = Form(0), 903 decoderFamily: str = Form(""), 904 decoderModel: str = Form(""), 905 roadName: str = Form(""), 906 roadNumber: str = Form(""), 907 mfg: str = Form(""), 908 model: str = Form(""), 909 owner: str = Form(""), 910 maxSpeedPct: int = Form(100), 911 ) -> Any: 912 await _dispatch(handle, type="roster", method="put", data={ 913 "id": id, 914 "dccAddress": dccAddress, 915 "longAddress": bool(longAddress), 916 "decoderFamily": decoderFamily, 917 "decoderModel": decoderModel, 918 "roadName": roadName, 919 "roadNumber": roadNumber, 920 "mfg": mfg, 921 "model": model, 922 "owner": owner, 923 "maxSpeedPct": maxSpeedPct, 924 }) 925 items = await _dispatch(handle, type="roster", method="list") 926 return templates.TemplateResponse( 927 request, "_roster_table.html", 928 {"roster": items, "catalog_loaded": len(_catalog()) > 0}, 929 ) 930 931 @app.get("/roster/{entry_id}/row", response_class=HTMLResponse) 932 async def roster_row(request: Request, entry_id: str) -> Any: 933 data = await _dispatch( 934 handle, type="roster", method="get", data={"id": entry_id}, 935 ) 936 return templates.TemplateResponse( 937 request, "_roster_row.html", 938 {"e": data, "catalog_loaded": len(_catalog()) > 0}, 939 ) 940 941 @app.get("/roster/{entry_id}/edit", response_class=HTMLResponse) 942 async def roster_edit_row(request: Request, entry_id: str) -> Any: 943 data = await _dispatch( 944 handle, type="roster", method="get", data={"id": entry_id}, 945 ) 946 return templates.TemplateResponse( 947 request, "_roster_row_edit.html", {"e": data}, 948 ) 949 950 @app.post("/roster/{entry_id}/update", response_class=HTMLResponse) 951 async def roster_update( 952 request: Request, 953 entry_id: str, 954 dccAddress: str = Form(""), 955 longAddress: int = Form(0), 956 decoderFamily: str = Form(""), 957 decoderModel: str = Form(""), 958 roadName: str = Form(""), 959 roadNumber: str = Form(""), 960 mfg: str = Form(""), 961 model: str = Form(""), 962 owner: str = Form(""), 963 maxSpeedPct: int = Form(100), 964 ) -> Any: 965 data = await _dispatch(handle, type="roster", method="post", data={ 966 "id": entry_id, 967 "dccAddress": dccAddress, 968 "longAddress": bool(longAddress), 969 "decoderFamily": decoderFamily, 970 "decoderModel": decoderModel, 971 "roadName": roadName, 972 "roadNumber": roadNumber, 973 "mfg": mfg, 974 "model": model, 975 "owner": owner, 976 "maxSpeedPct": maxSpeedPct, 977 }) 978 return templates.TemplateResponse( 979 request, "_roster_row.html", 980 {"e": data, "catalog_loaded": len(_catalog()) > 0}, 981 ) 982 983 @app.delete("/roster/{entry_id}", response_class=HTMLResponse) 984 async def roster_delete(request: Request, entry_id: str) -> Any: 985 await _dispatch(handle, type="roster", method="delete", data={"id": entry_id}) 986 return HTMLResponse("") 987 988 # ----- panel viewer ----------------------------------------------------- 989 990 @app.get("/panel", response_class=HTMLResponse) 991 async def panel(request: Request) -> Any: 992 return templates.TemplateResponse( 993 request, "panel.html", 994 _base_context(handle, "panel") | {"panel_summary": None}, 995 ) 996 997 # ----- decoder programmer ----------------------------------------------- 998 999 from jmri_core.persistence.decoder_xml import ( # noqa: PLC0415 1000 DecoderVariable, 1001 UnsupportedVariableKind, 1002 VariableKind, 1003 find_composite_current_choice, 1004 variable_apply_to_state, 1005 variable_cv_list, 1006 variable_extract_from_state, 1007 variable_extract_from_store, 1008 variable_is_indexed, 1009 variable_primary_cv, 1010 variable_store_key, 1011 variable_wire_read_plan, 1012 variable_wire_write_plan, 1013 ) 1014 from jmri_decoderpro import ( # noqa: PLC0415 1015 compose_panes, 1016 find_pane_by_slug, 1017 ) 1018 1019 def _programmer_context() -> dict[str, Any]: 1020 """The header/badge state for both the main /programmer page and 1021 the tabbed per-loco view. 1022 1023 ``decoder_xml_default`` pre-fills the catalog-load input from 1024 ``JMRI_DECODER_XML_ROOT`` when set; unset → empty string, 1025 matching the pre-Tier-3 behaviour where the operator typed 1026 the path each time. 1027 1028 ``load_in_progress`` (+ ``loaded`` / ``total`` / ``directory``) 1029 surface when either the ``JMRI_DECODER_XML_ROOT`` autoload or 1030 a manual ``/programmer/catalog/load`` is running. The catalog 1031 fragment renders the thermometer instead of the "Catalog 1032 empty" hint in that state, so a user who lands on 1033 ``/programmer`` mid-autoload doesn't see a stale message 1034 prompting them to re-click Load. 1035 """ 1036 from jmri_core.env import decoder_xml_root # noqa: PLC0415 1037 default_root = decoder_xml_root() 1038 progress = getattr(handle.ctx, "_catalog_load_progress", None) 1039 load_in_progress = bool( 1040 progress and not progress.get("done") and not progress.get("error"), 1041 ) 1042 ctx: dict[str, Any] = { 1043 "catalog_size": len(_catalog()), 1044 "manufacturers": _catalog().manufacturers(), 1045 "programmer_count": len(_catalog().programmers), 1046 "decoder_xml_default": str(default_root) if default_root else "", 1047 "load_in_progress": load_in_progress, 1048 } 1049 if load_in_progress: 1050 ctx["loaded"] = progress.get("loaded", 0) 1051 ctx["total"] = progress.get("total", 0) 1052 ctx["directory"] = progress.get("directory", "") 1053 return ctx 1054 1055 @app.get("/programmer", response_class=HTMLResponse) 1056 async def programmer_home(request: Request) -> Any: 1057 try: 1058 avail = await _dispatch(handle, type="programmer", method="list") 1059 except HTTPException: 1060 avail = [] 1061 return templates.TemplateResponse( 1062 request, "programmer.html", 1063 _base_context(handle, "programmer") 1064 | _programmer_context() 1065 | {"availability": avail}, 1066 ) 1067 1068 # ----- per-locomotive programmer (tabbed, DecoderPro-style) ------------- 1069 1070 def _pick_programmer_config(): 1071 """Return the programmer definition used for tab composition. 1072 1073 Preference (matches JMRI's default): ``Comprehensive`` > any 1074 other loaded programmer > None. 1075 """ 1076 if not _catalog().programmers: 1077 return None 1078 return ( 1079 _catalog().programmers.get("Comprehensive") 1080 or next(iter(_catalog().programmers.values()), None) 1081 ) 1082 1083 def _build_variable_row( 1084 var, cv_bytes: dict[str, int], current_value: int | None, 1085 *, definition=None, 1086 ) -> dict[str, Any]: 1087 """Turn a :class:`DecoderVariable` into the dict the pane template 1088 renders. Enum/composite choices are unpacked to plain dicts so 1089 Jinja can iterate them without knowing pydantic. 1090 1091 For composite variables, we also compute: 1092 1093 * ``current_choice`` — which preset's settings match the 1094 current sub-variable state (None if no match). 1095 * ``composite_choices[i].settings`` — the sub-variable writes 1096 each preset expands into. Templates carry this as 1097 ``data-settings`` JSON so client-side batch collection can 1098 fan a composite pick out into per-sub-variable BatchWrite 1099 entries without a server round-trip per choice. 1100 """ 1101 cv_list = variable_cv_list(var) 1102 cv_display = ".".join(str(c) for c in cv_list) if cv_list else var.cv 1103 primary_cv = variable_primary_cv(var) 1104 # For composite: settings map to sub-variable writes. 1105 composite_current_choice: str | None = None 1106 composite_choices: list[dict[str, Any]] = [] 1107 if var.kind is VariableKind.COMPOSITE and definition is not None: 1108 by_item = {v.item: v for v in definition.variables} 1109 composite_current_choice = find_composite_current_choice( 1110 var, by_item, cv_bytes, 1111 ) 1112 for cc in var.composite_choices: 1113 composite_choices.append({ 1114 "choice": cc.choice, 1115 "settings": [ 1116 {"item": s.label, "value": s.value} 1117 for s in cc.settings 1118 ], 1119 }) 1120 else: 1121 composite_choices = [ 1122 {"choice": c.choice, "settings": []} 1123 for c in var.composite_choices 1124 ] 1125 return { 1126 "item": var.item, 1127 "label": var.label, 1128 "tooltip": var.tooltip, 1129 "cv": primary_cv, 1130 "cv_display": cv_display, 1131 "kind": var.kind.value, 1132 "read_only": var.read_only, 1133 "write_only": var.write_only, 1134 "current_value": current_value, 1135 "current_choice": composite_current_choice, 1136 "wire_max": (1 << var.bit_mask.width) - 1 if var.kind in ( 1137 VariableKind.DECIMAL, VariableKind.HEXADECIMAL, VariableKind.ENUM, 1138 ) else 65535, 1139 "enum_choices": [ 1140 {"value": c.value, "choice": c.choice} 1141 for c in var.enum_choices 1142 ], 1143 "composite_choices": composite_choices, 1144 "constant_value": var.constant_value, 1145 } 1146 1147 def _resolve_cv_bytes( 1148 variables, cv_values: dict, 1149 ) -> dict[str, int]: 1150 """Build a string-keyed ``{store_key: byte}`` snapshot for the loco. 1151 1152 Layers, low → high precedence: 1153 1154 1. Per-variable defaults from the catalog: for each variable 1155 that declares a ``default`` (already-*masked* value), we 1156 apply it via :func:`variable_apply_to_state` so bits from 1157 multiple variables share the same CV byte correctly (e.g. 1158 CV 29's Address Format bit 5, Speed Steps bit 1, Power 1159 Source Conv bit 2 all compose into one default byte). 1160 Indexed variables get their default byte written under the 1161 compound key ``bank.index.cv``. 1162 2. Roster ``<values>`` from ``cv_values`` (already str-keyed) 1163 — a stored byte overrides whatever the default layer 1164 produced. 1165 1166 Result is fed to :func:`variable_extract_from_store` which 1167 looks up plain-int-string keys for non-indexed variables and 1168 compound keys for indexed ones. 1169 """ 1170 result: dict[str, int] = {} 1171 for var in variables: 1172 if var.default is None: 1173 continue 1174 if var.kind in ( 1175 VariableKind.SPEED_TABLE, 1176 VariableKind.COMPOSITE, 1177 VariableKind.SPLIT_DATETIME, 1178 VariableKind.CONSTANT, 1179 ): 1180 # These kinds don't participate in the mask-composition 1181 # model. Speed table has one CV per step (defaults are 1182 # per-step and handled by the speed-table row builder). 1183 # Composite and split-datetime aren't wire-value coded. 1184 # Constants don't live in the byte snapshot at all. 1185 continue 1186 # Fetch the current byte under the variable's store key 1187 # (needed for masked RMW). Then apply the default and put 1188 # the result back under the same key. 1189 if variable_is_indexed(var): 1190 # Indexed: apply per byte, keyed by compound name. 1191 # Multi-byte indexed splits get one key per byte. 1192 assert var.split is not None 1193 base = var.split.primary_cv 1194 for i in range(var.split.byte_count): 1195 key = variable_store_key(var, base + i) 1196 # Best-effort byte-slice of a multi-byte human default. 1197 shift = 8 * (var.split.byte_count - 1 - i) 1198 result[key] = (int(var.default) >> shift) & 0xFF 1199 continue 1200 try: 1201 # Non-indexed: use the legacy int-keyed helper against 1202 # an int-keyed view of ``result`` for backward-compat. 1203 int_view = { 1204 int(k): v for k, v in result.items() if k.isdigit() 1205 } 1206 updates = variable_apply_to_state(var, var.default, int_view) 1207 except Exception: # noqa: BLE001 1208 continue 1209 for cv_int, byte in updates.items(): 1210 result[str(cv_int)] = byte 1211 # Overlay roster/stored values — they win over defaults. 1212 for k, v in cv_values.items(): 1213 try: 1214 result[str(k)] = int(v) 1215 except (TypeError, ValueError): 1216 continue 1217 return result 1218 1219 def _build_speed_table_rows( 1220 var, cv_values: dict, 1221 ) -> list[dict[str, Any]]: 1222 """Expand a ``<speedTable>`` variable into one row per step. 1223 1224 NMRA layout: ``entries`` consecutive CVs starting at the 1225 variable's primary CV (typically CV67 → step 1, CV68 → step 2, 1226 …, CV94 → step 28). Each row is an independent 0..255 decimal 1227 input; dirty tracking + batch writes treat them as separate 1228 writes so the queue can retry a single failed step without 1229 re-sending the whole curve. 1230 """ 1231 base_cv = variable_primary_cv(var) 1232 entries = max(int(var.speed_table_entries or 28), 1) 1233 rows: list[dict[str, Any]] = [] 1234 for step_idx in range(entries): 1235 cv = base_cv + step_idx 1236 current = cv_values.get(str(cv)) 1237 try: 1238 current_int = int(current) if current is not None else None 1239 except (TypeError, ValueError): 1240 current_int = None 1241 # Each step becomes its own pseudo-variable so downstream 1242 # dirty tracking uses a per-step ``item`` id. 1243 step_label = f"Speed Step {step_idx + 1}" 1244 step_item = f"{var.item}[{step_idx + 1}]" 1245 rows.append({ 1246 "item": step_item, 1247 "label": step_label, 1248 "tooltip": var.tooltip, 1249 "cv": cv, 1250 "cv_display": str(cv), 1251 "kind": "decimal", 1252 "read_only": var.read_only, 1253 "write_only": var.write_only, 1254 "current_value": current_int, 1255 "wire_max": 255, 1256 "enum_choices": [], 1257 "composite_choices": [], 1258 "constant_value": None, 1259 }) 1260 return rows 1261 1262 _FNMAP_ITEM_RE = re.compile( 1263 r"^(?P<fn>F[^ ]+) controls output (?P<out>\d+)$", 1264 ) 1265 1266 def _fnmap_fn_sort_key(fn: str) -> tuple[int, int, str]: 1267 """JMRI ordering: STOP, DRIVE, FWD, REV, FL(f), FL(r), 1268 F0..F28, then anything else (sensors S0.., etc.) trailing. 1269 """ 1270 specials = { 1271 "STOP": (0, 0), "DRIVE": (0, 1), 1272 "FWD": (0, 2), "REV": (0, 3), 1273 "FL": (0, 4), "FL(f)": (0, 5), "FL(r)": (0, 6), 1274 "F0F": (0, 7), "F0R": (0, 8), 1275 } 1276 if fn in specials: 1277 band, idx = specials[fn] 1278 return (band, idx, "") 1279 m = re.match(r"^F(\d+)$", fn) 1280 if m: 1281 return (1, int(m.group(1)), "") 1282 m = re.match(r"^S(\d+)$", fn) 1283 if m: 1284 return (3, int(m.group(1)), "") 1285 return (2, 0, fn) 1286 1287 def _layout_tree_has_fnmapping(tree: dict) -> bool: 1288 """Walk *tree* looking for any ``fnmapping`` leaf node.""" 1289 if tree.get("type") == "fnmapping": 1290 return True 1291 for child in tree.get("children", ()): 1292 if _layout_tree_has_fnmapping(child): 1293 return True 1294 return False 1295 1296 def _build_fnmap_matrix( 1297 definition, cv_bytes: dict[str, int], 1298 row_by_item: dict[str, dict[str, Any]], 1299 *, variables=None, 1300 ) -> dict[str, Any]: 1301 """Collect all ``Fx controls output y`` variables into a matrix. 1302 1303 Mirrors JMRI's ``FnMapPanel`` layout: 1304 1305 * Rows are functions in natural order — ``FL(f)``, ``FL(r)``, 1306 ``F0``, ``F1``…``F28``, plus any special names the decoder 1307 uses (STOP/DRIVE/FWD/REV — rare, but supported). 1308 * Columns are physical outputs. Only outputs referenced by 1309 at least one variable become columns; outputs a decoder 1310 doesn't wire are suppressed so wide decoders (15+ outputs) 1311 don't get pointless empty columns. 1312 * Each column carries a two-line header — the numeric 1313 ``Output N`` and the decoder-declared output label (e.g. 1314 "Airhorn", "Bell", "FX3"). The label comes from 1315 ``<output name="N" label="…"/>`` on the model definition; 1316 JMRI's authoring convention is to use ``|`` as a line break 1317 inside the label, so we split on it for the sub-lines. 1318 1319 Returns ``{"fn_labels": […], "columns": [{"num": N, 1320 "label_lines": […]}], "cells": {(fn, out): row}}``. 1321 """ 1322 fn_labels: list[str] = [] 1323 outputs: set[int] = set() 1324 cells: dict[tuple[str, int], dict[str, Any]] = {} 1325 seen_fn: set[str] = set() 1326 if variables is None: 1327 variables = definition.variables if definition else () 1328 for var in variables: 1329 m = _FNMAP_ITEM_RE.match(var.item) 1330 if not m: 1331 continue 1332 fn = m.group("fn") 1333 try: 1334 out = int(m.group("out")) 1335 except ValueError: 1336 continue 1337 row = row_by_item.get(var.item) 1338 if row is None: 1339 try: 1340 current_int = variable_extract_from_store(var, cv_bytes) 1341 except UnsupportedVariableKind: 1342 current_int = None 1343 row = _build_variable_row( 1344 var, cv_bytes, current_int, definition=definition, 1345 ) 1346 cells[(fn, out)] = row 1347 outputs.add(out) 1348 if fn not in seen_fn: 1349 seen_fn.add(fn) 1350 fn_labels.append(fn) 1351 fn_labels.sort(key=_fnmap_fn_sort_key) 1352 1353 # Column header labels come from the decoder's <output> elements. 1354 # These live either on the model (JMRI's official spec) or on 1355 # the family — SoundTraxx Tsunami2 and Digitrax put them at 1356 # family level so the labels apply to every model. Model wins 1357 # when both define the same output; family fills the gaps. 1358 output_label_by_num: dict[int, str] = {} 1359 if definition is not None: 1360 output_sources = [] 1361 if definition.family is not None: 1362 output_sources.append(definition.family.outputs) 1363 if definition.model is not None: 1364 output_sources.append(definition.model.outputs) 1365 for source in output_sources: 1366 for o in source: 1367 try: 1368 n = int(o.name) 1369 except (TypeError, ValueError): 1370 continue 1371 if o.label: 1372 output_label_by_num[n] = o.label 1373 columns: list[dict[str, Any]] = [] 1374 for out in sorted(outputs): 1375 raw = output_label_by_num.get(out, "") 1376 # ``|`` is JMRI's authoring convention for line breaks in 1377 # output labels ("Airhorn| ", "Dynamic|Brakes"). 1378 lines = [ 1379 seg.strip() for seg in raw.split("|") if seg.strip() 1380 ] if raw else [] 1381 columns.append({"num": out, "label_lines": lines}) 1382 return { 1383 "fn_labels": fn_labels, 1384 "columns": columns, 1385 "cells": cells, 1386 } 1387 1388 def _build_pane_view( 1389 composed_pane, definition, cv_values: dict, 1390 ) -> dict[str, Any]: 1391 """Assemble the render context for one pane's HTML fragment. 1392 1393 Baseline value shown per row = roster ``<values>`` if the CV is 1394 set there, otherwise the decoder catalog's default. This merged 1395 baseline is what dirty tracking compares against; the JS on the 1396 page marks a row dirty when the user's input diverges from it. 1397 1398 SPEED_TABLE variables (``<speedTable entries="N"/>``) expand 1399 into N rows via :func:`_build_speed_table_rows` — one row per 1400 step CV so the batch queue can write each step independently. 1401 """ 1402 if composed_pane.is_synthetic: 1403 return {} 1404 # Filter variables against the current model's productID and 1405 # any ``<qualifier>`` conditions on other variables. 1406 # 1407 # * productID gate: ``<variable include="pid1,pid2,…">`` / 1408 # ``exclude="…"`` limits the variable to specific decoder 1409 # revisions. 1410 # * Qualifier gate: ``<qualifier>`` triples reference another 1411 # variable's current value (``<variableref>X</variableref> 1412 # <relation>ge</relation><value>7</value>``). All qualifiers 1413 # on a variable must pass (AND) for it to be visible. 1414 # 1415 # SoundTraxx Tsunami2 uses both heavily: HEP Mode / HEP Mode2 / 1416 # Turbine Start-Stop share CV 16.1.303 and take turns being 1417 # applicable depending on the loco model (productID) and prime 1418 # mover selection (``Sound Option 5``). Without the two gates 1419 # all three end up piled into the same grid cell. 1420 current_pid = ( 1421 definition.model.product_id 1422 if definition is not None and definition.model is not None 1423 else "" 1424 ) 1425 def _pid_applies(var) -> bool: 1426 if var.include_product_ids and current_pid not in var.include_product_ids: 1427 return False 1428 if var.exclude_product_ids and current_pid in var.exclude_product_ids: 1429 return False 1430 return True 1431 pid_vars = tuple( 1432 v for v in (definition.variables if definition else ()) 1433 if _pid_applies(v) 1434 ) 1435 # Build a preliminary by-item map from productID-applicable 1436 # variables and a preliminary cv_bytes so qualifiers can look 1437 # up their referenced variables' current values. 1438 pid_by_item = {v.item: v for v in pid_vars} 1439 pid_cv_bytes = _resolve_cv_bytes(pid_vars, cv_values) 1440 def _qualifiers_apply(var) -> bool: 1441 for q in var.qualifiers: 1442 ref = pid_by_item.get(q["variableref"]) 1443 if ref is None: 1444 # Referenced variable not present after productID 1445 # filtering — JMRI treats a missing ref as "gate 1446 # doesn't apply", so let the qualifier pass. Being 1447 # permissive here mirrors JMRI's tolerant behaviour 1448 # for decoder files that share qualifier snippets 1449 # across models. 1450 continue 1451 try: 1452 current = variable_extract_from_store(ref, pid_cv_bytes) 1453 except UnsupportedVariableKind: 1454 continue 1455 if current is None: 1456 continue 1457 rel = q["relation"] 1458 if rel == "exists": 1459 continue 1460 try: 1461 target = int(q["value"]) 1462 except (TypeError, ValueError): 1463 continue 1464 if rel == "eq" and current != target: return False 1465 if rel == "ne" and current == target: return False 1466 if rel == "gt" and current <= target: return False 1467 if rel == "ge" and current < target: return False 1468 if rel == "lt" and current >= target: return False 1469 if rel == "le" and current > target: return False 1470 return True 1471 filtered_vars = tuple(v for v in pid_vars if _qualifiers_apply(v)) 1472 by_item = {v.item: v for v in filtered_vars} 1473 rows: list[dict[str, Any]] = [] 1474 # ``<speedtable/>`` widget: JMRI's Speed Table pane uses this 1475 # instead of per-step <display> items. Find the decoder's 1476 # SPEED_TABLE variable and expand it here (before iterating 1477 # items_available) so the pane renders even when no explicit 1478 # <display item="Speed Table"/> reference exists. Track which 1479 # items we've expanded to avoid double-rendering if the pane 1480 # also lists them via <display>. 1481 expanded_items: set[str] = set() 1482 pane_decl = composed_pane.pane 1483 # Two triggers for speed-table expansion: 1484 # (a) ``<speedtable/>`` widget detected by the parser, OR 1485 # (b) the composed pane's name/slug identifies it as the Speed 1486 # Table tab — a name-based fallback so a real JMRI install 1487 # that ships an unexpected widget variant still expands 1488 # the decoder's SPEED_TABLE variable when the user opens 1489 # the "Speed Table" tab. 1490 pane_is_speed_table = ( 1491 (pane_decl is not None and pane_decl.has_speed_table) 1492 or composed_pane.slug == "speed-table" 1493 or composed_pane.name.strip().lower() == "speed table" 1494 ) 1495 if pane_is_speed_table: 1496 found_speed_table_var = False 1497 for v in (definition.variables if definition else ()): 1498 if v.kind == VariableKind.SPEED_TABLE: 1499 rows.extend(_build_speed_table_rows(v, cv_values)) 1500 expanded_items.add(v.item) 1501 found_speed_table_var = True 1502 if not found_speed_table_var: 1503 # No SPEED_TABLE variable in the catalog — many decoder 1504 # XMLs omit it, relying on the NMRA convention that CV 1505 # 67..CV 94 are the 28-step table (bytes 1..28). Emit a 1506 # synthetic 28-row expansion so the user can still write 1507 # each step. Use the class:`DecoderVariable` model with 1508 # the anchor CV and default entries so 1509 # :func:`_build_speed_table_rows` handles it. 1510 synthetic = DecoderVariable.model_validate({ 1511 "CV": "67", 1512 "item": "Speed Table", 1513 "kind": VariableKind.SPEED_TABLE, 1514 "speed_table_entries": 28, 1515 }) 1516 rows.extend(_build_speed_table_rows(synthetic, cv_values)) 1517 # Build an int-keyed CV byte snapshot, with defaults filling in 1518 # for CVs that haven't been read/written yet. For bit-mask 1519 # variables (Address Format, Speed Steps, Power Source Conv — 1520 # all sharing CV 29), we must call ``variable_extract_from_state`` 1521 # so the row's "current" shows the *masked* value, not the raw 1522 # byte. Otherwise every enum on a shared CV compares its wire 1523 # value (0/1/2/…) against the whole byte (e.g. 2) and dirty- 1524 # tracks a false positive; and switching panes shows the same 1525 # raw byte across all variables, looking wildly inconsistent. 1526 cv_bytes = _resolve_cv_bytes(filtered_vars, cv_values) 1527 for item_name in composed_pane.items_available: 1528 if item_name in expanded_items: 1529 continue 1530 var = by_item.get(item_name) 1531 if var is None: 1532 continue 1533 if var.kind == VariableKind.SPEED_TABLE: 1534 rows.extend(_build_speed_table_rows(var, cv_values)) 1535 continue 1536 try: 1537 current_int = variable_extract_from_store(var, cv_bytes) 1538 except UnsupportedVariableKind: 1539 current_int = None 1540 rows.append(_build_variable_row( 1541 var, cv_bytes, current_int, definition=definition, 1542 )) 1543 # On the Speed Table pane we synthesize CV 67..CV 94 rows before 1544 # iterating items_available, so the natural row order interleaves 1545 # oddly (67..94 first, then whatever items_available lists like 1546 # CV 65 Kickstart / CV 66 Forward Trim). Sort by CV number here 1547 # so the pane reads top-to-bottom in CV order. Only applies to 1548 # the speed-table pane — other panes rely on JMRI's explicit 1549 # display ordering. 1550 if pane_is_speed_table: 1551 def _cv_sort_key(row: dict[str, Any]) -> int: 1552 try: 1553 return int(row.get("cv") or 0) 1554 except (TypeError, ValueError): 1555 return 0 1556 rows.sort(key=_cv_sort_key) 1557 1558 # Raw CV table (CVs pane) — enumerate every CV the decoder 1559 # defines, merged with whatever's been read/written into the 1560 # store. Each row is displayed with its stored value (if any) 1561 # and gives the user a Read/Write handle on it. Matches JMRI 1562 # Java's CVs tab which lists everything up front, not just 1563 # what's been touched. 1564 raw_cv_rows: list[dict[str, Any]] = [] 1565 if pane_decl is not None and pane_decl.has_cvtable: 1566 catalog_names: set[str] = set() 1567 for var in (definition.variables if definition else ()): 1568 if var.kind in ( 1569 VariableKind.CONSTANT, VariableKind.SPEED_TABLE, 1570 VariableKind.COMPOSITE, VariableKind.SPLIT_DATETIME, 1571 ): 1572 continue 1573 if variable_is_indexed(var): 1574 assert var.split is not None 1575 base = var.split.primary_cv 1576 for i in range(var.split.byte_count): 1577 catalog_names.add(variable_store_key(var, base + i)) 1578 else: 1579 for cv in variable_cv_list(var): 1580 catalog_names.add(str(cv)) 1581 # Also include anything in the store that isn't in the 1582 # catalog (e.g. legacy roster entries, raw CVs the user 1583 # added via the "Add CV" input in a prior session). 1584 all_names = catalog_names | set(cv_values.keys()) 1585 def _raw_sort_key(name: str) -> tuple[int, int, int, str]: 1586 # Sort plain CVs first (by number), then indexed by 1587 # (bank, index, base). Falls through to string compare 1588 # for anything that doesn't parse. 1589 parts = name.split(".") 1590 try: 1591 nums = [int(p) for p in parts] 1592 except ValueError: 1593 return (2, 0, 0, name) 1594 if len(nums) == 1: 1595 return (0, nums[0], 0, "") 1596 if len(nums) == 3: 1597 return (1, nums[0], nums[1] * 1000 + nums[2], "") 1598 return (1, 0, 0, name) 1599 for name in sorted(all_names, key=_raw_sort_key): 1600 v = cv_values.get(name) 1601 try: 1602 v_int = int(v) if v is not None else None 1603 except (TypeError, ValueError): 1604 v_int = None 1605 raw_cv_rows.append({ 1606 "name": name, 1607 "value": v_int, 1608 "hex": f"{v_int:02X}" if v_int is not None else "", 1609 }) 1610 # Rich layout tree — the JMRI-authored column/row/label/ 1611 # separator structure. The template renders it recursively via 1612 # the ``layout_node`` macro. Each ``<display>`` leaf resolves 1613 # to a variable row (looked up by item), which the macro emits 1614 # inline with its label + input + Read button. When the pane 1615 # doesn't have a raw XML element (synthetic tests, or older 1616 # panes), the fallback tree in ``pane_layout_tree`` gives a 1617 # flat column of the pane's ``items`` list — same widgets as 1618 # the legacy table view, just wrapped in divs. 1619 from jmri_core.persistence.decoder_xml import pane_layout_tree # noqa: PLC0415 1620 layout_tree = None 1621 row_by_item: dict[str, dict[str, Any]] = {} 1622 if composed_pane.pane is not None: 1623 layout_tree = pane_layout_tree(composed_pane.pane) 1624 row_by_item = {r["item"]: r for r in rows if r.get("item")} 1625 1626 # ----- Function Map matrix ----- 1627 # JMRI's ``<fnmapping/>`` marker is an app-generated grid of 1628 # checkboxes keyed by every ``Fx controls output y`` variable 1629 # the decoder defines. The pane XML doesn't list these 1630 # individually (that's the point of the marker), so we collect 1631 # them here from ``definition.variables`` and shape them into 1632 # a matrix keyed by function label + output number. When the 1633 # pane has no fnmapping marker we skip the whole thing. 1634 fnmap_matrix: dict[str, Any] | None = None 1635 if layout_tree is not None and _layout_tree_has_fnmapping(layout_tree): 1636 fnmap_matrix = _build_fnmap_matrix( 1637 definition, cv_bytes, row_by_item, variables=filtered_vars, 1638 ) 1639 return { 1640 "variable_rows": rows, 1641 "raw_cv_rows": raw_cv_rows, 1642 "layout_tree": layout_tree, 1643 "row_by_item": row_by_item, 1644 "is_speed_table_pane": pane_is_speed_table, 1645 "fnmap_matrix": fnmap_matrix, 1646 } 1647 1648 @app.get("/programmer/loco/{loco_id}", response_class=HTMLResponse) 1649 async def programmer_loco( 1650 request: Request, loco_id: str, tab: str | None = None, 1651 ) -> Any: 1652 """Per-loco tabbed CV programmer. 1653 1654 Roster entry drives the tab set: the loco's decoder family/model 1655 picks a :class:`DecoderDefinition` from the catalog, then 1656 ``compose_panes`` unions the standard programmer panes (from 1657 Comprehensive.xml if loaded) with any decoder-declared extras. 1658 """ 1659 try: 1660 loco = await _dispatch( 1661 handle, type="roster", method="get", data={"id": loco_id}, 1662 ) 1663 except HTTPException as e: 1664 return HTMLResponse( 1665 f"no such roster entry: {loco_id}", status_code=e.status_code, 1666 ) 1667 1668 family = loco.get("decoderFamily") or "" 1669 model = loco.get("decoderModel") or "" 1670 definition = ( 1671 _catalog().find(family, model) if (family and model) else None 1672 ) 1673 programmer_cfg = _pick_programmer_config() 1674 1675 variables = definition.variables if definition else () 1676 decoder_panes = definition.panes if definition else () 1677 tab_panes = compose_panes(list(variables), programmer_cfg, list(decoder_panes)) 1678 1679 # Per-loco CV state. If the store sees CVs in both the JMRI- 1680 # canonical roster/<id>.xml <values> block AND our legacy 1681 # rosters/<id>.xml, ask the user which is authoritative before 1682 # rendering panes with potentially-wrong values. 1683 file_name = loco.get("fileName") or None 1684 from jmri_server.services.programmer import _loco_cv_store # noqa: PLC0415 1685 from jmri_server.loco_cv_store import AmbiguousCvSourceError # noqa: PLC0415 1686 store = _loco_cv_store(handle.ctx) 1687 try: 1688 state = store.get(loco_id, file_name) 1689 except AmbiguousCvSourceError as ambig: 1690 return templates.TemplateResponse( 1691 request, "programmer_cv_source_choose.html", 1692 _base_context(handle, "programmer") 1693 | _programmer_context() 1694 | { 1695 "loco": loco, 1696 "loco_id": loco_id, 1697 "roster_path": str(ambig.roster_path), 1698 "legacy_path": str(ambig.legacy_path), 1699 }, 1700 ) 1701 1702 # Default active tab = Roster Entry. Query-param override lets 1703 # deep links land on a specific tab. 1704 active_slug = tab or (tab_panes[0].slug if tab_panes else "roster-entry") 1705 active_pane = find_pane_by_slug(tab_panes, active_slug) or ( 1706 tab_panes[0] if tab_panes else None 1707 ) 1708 if active_pane is not None: 1709 active_slug = active_pane.slug 1710 1711 pane_ctx = _build_pane_view(active_pane, definition, state.cv_values) if active_pane else {} 1712 1713 return templates.TemplateResponse( 1714 request, "programmer_loco.html", 1715 _base_context(handle, "programmer") 1716 | _programmer_context() 1717 | { 1718 "loco": loco, 1719 "loco_id": loco_id, 1720 "definition": definition, 1721 "family": family, 1722 "model": model, 1723 "cv_values": state.cv_values, 1724 "tab_panes": tab_panes, 1725 "active_slug": active_slug, 1726 "pane": active_pane, 1727 **pane_ctx, 1728 }, 1729 ) 1730 1731 @app.post( 1732 "/programmer/loco/{loco_id}/resolve-source", 1733 response_class=HTMLResponse, 1734 ) 1735 async def programmer_loco_resolve_source( 1736 request: Request, loco_id: str, source: str = Form(...), 1737 ) -> Any: 1738 """Persist the user's pick for which CV-state source is 1739 authoritative on this loco, then bounce back to the programmer 1740 view so the panes render with the chosen source.""" 1741 from jmri_server.loco_cv_store import CvSource # noqa: PLC0415 1742 from jmri_server.services.programmer import _loco_cv_store # noqa: PLC0415 1743 try: 1744 picked = CvSource(source) 1745 except ValueError: 1746 raise HTTPException(400, f"unknown source: {source}") from None 1747 _loco_cv_store(handle.ctx).set_preferred_source(loco_id, picked) 1748 # 303 See Other → re-run the GET so panes render. 1749 return HTMLResponse( 1750 "", status_code=303, 1751 headers={"Location": f"/programmer/loco/{loco_id}"}, 1752 ) 1753 1754 @app.get( 1755 "/programmer/loco/{loco_id}/pane/{pane_slug}", 1756 response_class=HTMLResponse, 1757 ) 1758 async def programmer_loco_pane( 1759 request: Request, loco_id: str, pane_slug: str, 1760 ) -> Any: 1761 """Return one pane's HTML fragment for HTMX tab-swap.""" 1762 try: 1763 loco = await _dispatch( 1764 handle, type="roster", method="get", data={"id": loco_id}, 1765 ) 1766 except HTTPException as e: 1767 return HTMLResponse(str(e.detail), status_code=e.status_code) 1768 1769 family = loco.get("decoderFamily") or "" 1770 model = loco.get("decoderModel") or "" 1771 definition = ( 1772 _catalog().find(family, model) if (family and model) else None 1773 ) 1774 programmer_cfg = _pick_programmer_config() 1775 1776 variables = definition.variables if definition else () 1777 decoder_panes = definition.panes if definition else () 1778 tab_panes = compose_panes(list(variables), programmer_cfg, list(decoder_panes)) 1779 active_pane = find_pane_by_slug(tab_panes, pane_slug) 1780 if active_pane is None: 1781 return HTMLResponse("no such pane", status_code=404) 1782 1783 file_name = loco.get("fileName") or None 1784 from jmri_server.services.programmer import _loco_cv_store # noqa: PLC0415 1785 store = _loco_cv_store(handle.ctx) 1786 state = store.get(loco_id, file_name) 1787 1788 pane_ctx = _build_pane_view(active_pane, definition, state.cv_values) 1789 return templates.TemplateResponse( 1790 request, "_programmer_pane_content.html", 1791 { 1792 "loco": loco, 1793 "loco_id": loco_id, 1794 "family": family, 1795 "model": model, 1796 "active_slug": pane_slug, 1797 "pane": active_pane, 1798 "cv_values": state.cv_values, 1799 **pane_ctx, 1800 }, 1801 ) 1802 1803 @app.post("/programmer/loco/{loco_id}/read") 1804 async def programmer_loco_read( 1805 request: Request, loco_id: str, 1806 payload: dict[str, Any] = Body(...), 1807 ) -> Any: 1808 """JSON-shaped single-CV read. 1809 1810 When the payload includes the row's ``item`` and the catalog 1811 knows a variable for that name, the response also carries the 1812 variable's *extracted* masked value ("humanValue") — the JS 1813 needs that to update the row's ``data-current`` correctly. 1814 Otherwise a masked-bit read (Address Format, Speed Steps, etc.) 1815 would put the raw CV byte into ``data-current`` and every 1816 subsequent dirty-check on that row would compare the byte 1817 against the enum's masked-value option and flip false-dirty. 1818 """ 1819 cv = payload.get("cv") 1820 if not isinstance(cv, int): 1821 raise HTTPException(400, "read payload requires integer 'cv'") 1822 item = str(payload.get("item") or "") 1823 dispatch_payload: dict[str, Any] = { 1824 "cv": cv, 1825 "mode": str(payload.get("mode", "service")), 1826 "locoId": loco_id, 1827 } 1828 file_name = payload.get("fileName") or payload.get("file_name") 1829 if file_name: 1830 dispatch_payload["fileName"] = file_name 1831 address = payload.get("address") 1832 if address is not None: 1833 dispatch_payload["address"] = int(address) 1834 dispatch_payload["isLongAddress"] = bool(payload.get("isLongAddress")) 1835 try: 1836 data = await _dispatch( 1837 handle, type="programmer", method="get", data=dispatch_payload, 1838 ) 1839 except HTTPException as e: 1840 raise HTTPException(e.status_code, e.detail) from None 1841 wire_value = int(data.get("value", 0)) 1842 human_value: int | None = None 1843 var = _find_variable_for_write(loco_id, item) if item else None 1844 if var is not None: 1845 # Snapshot the post-write CV bytes (the read already pushed 1846 # this CV into the store) and extract per-variable so the 1847 # client can set ``data-current`` to the masked value. 1848 from jmri_server.services.programmer import _loco_cv_store # noqa: PLC0415 1849 store = _loco_cv_store(handle.ctx) 1850 state = store.get(loco_id, file_name) 1851 # String-keyed snapshot supports indexed CVs (``16.1.275``) 1852 # coexisting with plain CVs (``275``) — the extract helper 1853 # picks the right key per variable. 1854 cv_bytes = {str(k): int(v) for k, v in state.cv_values.items()} 1855 try: 1856 human_value = variable_extract_from_store(var, cv_bytes) 1857 except UnsupportedVariableKind: 1858 human_value = None 1859 return { 1860 "cv": cv, 1861 "value": wire_value, 1862 "humanValue": human_value if human_value is not None else wire_value, 1863 "mode": str(data.get("mode", "service_direct")), 1864 } 1865 1866 @app.post("/programmer/loco/{loco_id}/write", response_class=HTMLResponse) 1867 async def programmer_loco_write( 1868 request: Request, loco_id: str, 1869 cv: int = Form(...), value: int = Form(...), 1870 mode: str = Form("service"), address: int | None = Form(None), 1871 file_name: str = Form(""), 1872 ) -> Any: 1873 payload: dict[str, Any] = { 1874 "cv": cv, "value": value, "mode": mode, "locoId": loco_id, 1875 } 1876 if file_name: 1877 payload["fileName"] = file_name 1878 if address is not None: 1879 payload["address"] = address 1880 try: 1881 data = await _dispatch(handle, type="programmer", method="post", data=payload) 1882 except HTTPException as e: 1883 return templates.TemplateResponse( 1884 request, "_programmer_cv_result.html", 1885 {"cv": cv, "error": e.detail}, status_code=e.status_code, 1886 ) 1887 return templates.TemplateResponse( 1888 request, "_programmer_cv_result.html", 1889 {"cv": cv, "value": data["value"], "mode": data["mode"], "loco_id": loco_id}, 1890 ) 1891 1892 # ----- raw CV access on the CVs tab --------------------------------- 1893 1894 from jmri_core.persistence.decoder_xml import SplitCv # noqa: PLC0415 1895 1896 def _parse_raw_cv_name(name: str) -> tuple[int, SplitCv | None]: 1897 """Split a raw-CV table entry into ``(wire_cv, split_or_None)``. 1898 1899 ``"275"`` → (275, None) — plain CV, no indexed setup. 1900 ``"16.1.275"`` → (275, SplitCv(bank=16, index=1, primary_cv=275)) 1901 — compound CV, caller must prepend CV31/CV32 setup writes. 1902 1903 Raises :class:`ValueError` on unparseable input so the caller 1904 can 400 out with a clear message. 1905 """ 1906 name = name.strip() 1907 if not name: 1908 raise ValueError("CV name must not be empty") 1909 if "." in name: 1910 split = SplitCv.parse(name) 1911 if split.primary_cv <= 0 or split.bank is None: 1912 raise ValueError(f"malformed indexed CV name: {name!r}") 1913 return split.primary_cv, split 1914 try: 1915 return int(name), None 1916 except ValueError as e: 1917 raise ValueError(f"CV name must be int or bank.index.base, got {name!r}") from e 1918 1919 @app.post("/programmer/loco/{loco_id}/cv/read") 1920 async def programmer_loco_raw_cv_read( 1921 loco_id: str, payload: dict[str, Any] = Body(...), 1922 ) -> Any: 1923 """Read one CV on the CVs tab. Accepts either plain (``275``) 1924 or compound (``16.1.275``) names. 1925 1926 For compound CVs, dispatches CV31 = bank and CV32 = index setup 1927 writes before the read, and stashes the result in the roster 1928 under the compound key so subsequent renders show it. 1929 """ 1930 cv_name = str(payload.get("cvName") or "") 1931 try: 1932 wire_cv, split = _parse_raw_cv_name(cv_name) 1933 except ValueError as e: 1934 raise HTTPException(400, str(e)) from None 1935 mode = str(payload.get("mode", "service")) 1936 address = payload.get("address") 1937 file_name = payload.get("fileName") 1938 1939 common: dict[str, Any] = {"mode": mode} 1940 if address is not None: 1941 common["address"] = int(address) 1942 common["isLongAddress"] = bool(payload.get("isLongAddress")) 1943 if file_name: 1944 common["fileName"] = file_name 1945 1946 # Compound: prepend CV31/CV32 setup as service writes (no store). 1947 if split is not None: 1948 for setup_cv, setup_byte in ((31, split.bank), (32, split.index or 0)): 1949 await _dispatch( 1950 handle, type="programmer", method="post", 1951 data={**common, "cv": setup_cv, "value": int(setup_byte or 0)}, 1952 ) 1953 # Read the base CV; skip service-side persistence, we'll set the 1954 # store manually with the correct key below. 1955 try: 1956 data = await _dispatch( 1957 handle, type="programmer", method="get", 1958 data={**common, "cv": wire_cv}, 1959 ) 1960 except HTTPException as e: 1961 raise HTTPException(e.status_code, e.detail) from None 1962 value = int(data.get("value", 0)) 1963 from jmri_server.services.programmer import _loco_cv_store # noqa: PLC0415 1964 _loco_cv_store(handle.ctx).record_read_result( 1965 loco_id, cv_name, value, 1966 file_name if isinstance(file_name, str) else None, 1967 ) 1968 return {"cvName": cv_name, "value": value} 1969 1970 @app.post("/programmer/loco/{loco_id}/cv/write") 1971 async def programmer_loco_raw_cv_write( 1972 loco_id: str, payload: dict[str, Any] = Body(...), 1973 ) -> Any: 1974 """Write one CV on the CVs tab. Plain or compound; same shape 1975 as the read endpoint but with a ``value`` byte. 1976 1977 Compound writes prepend CV31/CV32 setup writes and persist the 1978 payload under the compound key (matching JMRI Java's roster 1979 format ``<CVvalue name="16.1.275" value="X"/>``).""" 1980 cv_name = str(payload.get("cvName") or "") 1981 try: 1982 wire_cv, split = _parse_raw_cv_name(cv_name) 1983 except ValueError as e: 1984 raise HTTPException(400, str(e)) from None 1985 try: 1986 value = int(payload["value"]) 1987 except (KeyError, TypeError, ValueError): 1988 raise HTTPException(400, "value must be an int 0..255") from None 1989 if not 0 <= value <= 255: 1990 raise HTTPException(400, "value out of range 0..255") 1991 1992 mode = str(payload.get("mode", "service")) 1993 address = payload.get("address") 1994 file_name = payload.get("fileName") 1995 common: dict[str, Any] = {"mode": mode} 1996 if address is not None: 1997 common["address"] = int(address) 1998 common["isLongAddress"] = bool(payload.get("isLongAddress")) 1999 if file_name: 2000 common["fileName"] = file_name 2001 2002 if split is not None: 2003 for setup_cv, setup_byte in ((31, split.bank), (32, split.index or 0)): 2004 await _dispatch( 2005 handle, type="programmer", method="post", 2006 data={**common, "cv": setup_cv, "value": int(setup_byte or 0)}, 2007 ) 2008 # Base-CV write. Omit ``locoId`` so the service's built-in store 2009 # update doesn't fire under the plain int key — we do it below 2010 # with the compound (or plain) name. 2011 try: 2012 data = await _dispatch( 2013 handle, type="programmer", method="post", 2014 data={**common, "cv": wire_cv, "value": value}, 2015 ) 2016 except HTTPException as e: 2017 raise HTTPException(e.status_code, e.detail) from None 2018 from jmri_server.services.programmer import _loco_cv_store # noqa: PLC0415 2019 _loco_cv_store(handle.ctx).set_cv( 2020 loco_id, cv_name, value, 2021 file_name if isinstance(file_name, str) else None, 2022 ) 2023 return {"cvName": cv_name, "value": int(data.get("value", value))} 2024 2025 # ----- programmer step 4: dirty-tracked batch writes ---------------- 2026 2027 from jmri_server.programming_queue import ( # noqa: PLC0415 2028 BatchInFlightError, 2029 BatchWrite, 2030 get_or_create_queue, 2031 ) 2032 2033 def _make_write_callback(loco_id: str) -> Any: 2034 """Build the callback the queue calls per CV. 2035 2036 Two responsibilities on top of just dispatching to 2037 ``programmer.post``: 2038 2039 * **Read-modify-write for masked variables.** The client sends 2040 the variable's *human* value (e.g. Address Format=1 for long). 2041 The wire needs the full CV byte with the other variables' 2042 bits preserved. We look up the variable by ``item`` name in 2043 the decoder catalog, compute the full byte via 2044 :func:`variable_apply_to_state` against the current CV 2045 snapshot, and issue writes for the resulting CV bytes. 2046 * **Split-CV / long-address writes.** ``variable_apply_to_state`` 2047 returns one entry per touched CV; long-address writes two 2048 bytes (CV 17 + CV 18) — we dispatch each so the wire sees a 2049 real per-CV WRITE_DIR_CV_CMD. 2050 2051 A write that can't be resolved to a catalog variable (no 2052 catalog loaded, item name missing) falls back to writing the 2053 raw value to the raw CV — matches the pre-fix behaviour so 2054 single-CV panes / raw-CV writes still work. 2055 """ 2056 async def _dispatch_one( 2057 cv: int, value: int, extra: dict[str, Any], 2058 ) -> int: 2059 payload: dict[str, Any] = { 2060 "cv": cv, "value": value, 2061 "mode": extra.get("mode", "service"), 2062 "locoId": loco_id, 2063 } 2064 file_name = extra.get("file_name") 2065 if file_name: 2066 payload["fileName"] = file_name 2067 addr = extra.get("address") 2068 if addr is not None: 2069 payload["address"] = addr 2070 payload["isLongAddress"] = bool(extra.get("long_address")) 2071 data = await _dispatch( 2072 handle, type="programmer", method="post", data=payload, 2073 ) 2074 return int(data.get("value", value)) 2075 2076 async def _dispatch_wire_only(cv: int, value: int, extra: dict[str, Any]) -> int: 2077 """Write to the wire without persisting to the CV store. 2078 2079 Used for indexed-CV setup writes (CV31=bank, CV32=index). 2080 Those bytes are physically written but must NOT show up in 2081 the roster's per-CV tracking — they're transient page- 2082 selector state, not user data. Omitting ``locoId`` from 2083 the payload tells the programmer service to skip the store 2084 update. 2085 """ 2086 payload: dict[str, Any] = { 2087 "cv": cv, "value": value, 2088 "mode": extra.get("mode", "service"), 2089 } 2090 addr = extra.get("address") 2091 if addr is not None: 2092 payload["address"] = addr 2093 payload["isLongAddress"] = bool(extra.get("long_address")) 2094 data = await _dispatch( 2095 handle, type="programmer", method="post", data=payload, 2096 ) 2097 return int(data.get("value", value)) 2098 2099 async def _dispatch_with_key( 2100 cv: int, value: int, store_key: str, extra: dict[str, Any], 2101 ) -> int: 2102 """Write to wire AND persist under ``store_key`` (used for 2103 indexed CVs whose store name is ``bank.index.cv``, not 2104 plain ``str(cv)``). We deliberately bypass the programmer 2105 service's built-in store update and do it here so we can 2106 control the key.""" 2107 wire = await _dispatch_wire_only(cv, value, extra) 2108 from jmri_server.services.programmer import _loco_cv_store # noqa: PLC0415 2109 _loco_cv_store(handle.ctx).set_cv( 2110 loco_id, store_key, value, extra.get("file_name"), 2111 ) 2112 return wire 2113 2114 async def _write(cv: int, value: int, item: str, extra: dict[str, Any]) -> int: 2115 var = _find_variable_for_write(loco_id, item) 2116 if var is None: 2117 # Composite fan-out sends ``cv=0`` as a sentinel meaning 2118 # "resolve the wire CV from ``item``". If we couldn't 2119 # resolve the item, that's a broken composite reference 2120 # or catalog mismatch — bail out with a clear error so 2121 # the batch reports the failure rather than silently 2122 # writing to CV 0. 2123 if cv == 0 and item: 2124 raise ValueError( 2125 f"composite sub-variable {item!r} not found in catalog", 2126 ) 2127 # No catalog metadata → treat as a raw single-CV write. 2128 return await _dispatch_one(cv, value, extra) 2129 # Snapshot current CV bytes from the store so read-modify-write 2130 # preserves bits owned by other variables on the same CV. 2131 # Keys are strings so indexed CVs (``16.1.275``) can coexist 2132 # with plain ones (``275``) without collision. 2133 from jmri_server.services.programmer import _loco_cv_store # noqa: PLC0415 2134 store = _loco_cv_store(handle.ctx) 2135 state = store.get(loco_id, extra.get("file_name")) 2136 cv_bytes = {str(k): int(v) for k, v in state.cv_values.items()} 2137 try: 2138 plan = variable_wire_write_plan(var, value, cv_bytes) 2139 except Exception: # noqa: BLE001 2140 # Unsupported kind (speed table, composite, ...) — 2141 # fall back to a raw write on the primary CV. 2142 return await _dispatch_one(cv, value, extra) 2143 if not plan: 2144 return int(value) # read-only / constant → no-op 2145 # Issue physical writes in order. Setup ops (CV31/CV32) 2146 # go through _dispatch_wire_only so they don't pollute 2147 # the roster; value ops go through _dispatch_with_key so 2148 # they land under the correct compound key for indexed 2149 # variables. 2150 last_committed = value 2151 for op in plan: 2152 if op.store_key is None: 2153 last_committed = await _dispatch_wire_only( 2154 op.wire_cv, op.byte_value, extra, 2155 ) 2156 else: 2157 last_committed = await _dispatch_with_key( 2158 op.wire_cv, op.byte_value, op.store_key, extra, 2159 ) 2160 return last_committed 2161 return _write 2162 2163 def _make_read_callback(loco_id: str) -> Any: 2164 """Build the read callback the queue calls per CV. 2165 2166 Delegates to the JSON ``programmer.get`` service — same wire as 2167 the single-CV Read button. The service stashes each read into 2168 the loco CV store, so after the batch completes the client can 2169 refresh the pane and see freshly-extracted values. 2170 """ 2171 async def _dispatch_read( 2172 cv: int, extra: dict[str, Any], *, store_key: str | None, 2173 ) -> int: 2174 """Wire read + optional store update under *store_key*. 2175 2176 When *store_key* is None (indexed CV31/CV32 setup writes 2177 aren't reads — this shouldn't be called), or when the read 2178 is for a simple CV that keys plainly as ``str(cv)``, we 2179 just include ``locoId`` in the payload and let the 2180 programmer service persist via its own ``record_read_result`` 2181 path. When the caller wants a compound store key, we omit 2182 ``locoId`` (service skips persistence) and set the value 2183 manually with the right key. 2184 """ 2185 payload: dict[str, Any] = { 2186 "cv": cv, 2187 "mode": extra.get("mode", "service"), 2188 } 2189 file_name = extra.get("file_name") 2190 if file_name: 2191 payload["fileName"] = file_name 2192 addr = extra.get("address") 2193 if addr is not None: 2194 payload["address"] = addr 2195 payload["isLongAddress"] = bool(extra.get("long_address")) 2196 # For plain reads we can let the service persist directly. 2197 if store_key is None or store_key == str(cv): 2198 payload["locoId"] = loco_id 2199 data = await _dispatch( 2200 handle, type="programmer", method="get", data=payload, 2201 ) 2202 return int(data.get("value", 0)) 2203 # Compound key path (indexed CV): dispatch without locoId, 2204 # then manually record under the compound key. 2205 data = await _dispatch( 2206 handle, type="programmer", method="get", data=payload, 2207 ) 2208 value = int(data.get("value", 0)) 2209 from jmri_server.services.programmer import _loco_cv_store # noqa: PLC0415 2210 _loco_cv_store(handle.ctx).record_read_result( 2211 loco_id, store_key, value, file_name, 2212 ) 2213 return value 2214 2215 async def _read(cv: int, item: str, extra: dict[str, Any]) -> int: 2216 var = _find_variable_for_write(loco_id, item) 2217 if var is None or not variable_is_indexed(var): 2218 # Plain read — service persists under ``str(cv)``. 2219 return await _dispatch_read(cv, extra, store_key=str(cv)) 2220 # Indexed read: CV31/CV32 setup (writes, no persistence), 2221 # then read the base CV and persist under compound key. 2222 assert var.split is not None 2223 # For a read op, we need to issue writes to CV31/CV32 too. 2224 # Those aren't in the "read wire plan" we can call 2225 # directly — write the setup manually using the 2226 # write-side dispatch helper. Fetch it via a lexical 2227 # closure — we're inside the same _make_read_callback 2228 # scope but the write helpers were defined in 2229 # _make_write_callback; reach through the queue by 2230 # calling the JSON write path directly here to avoid 2231 # crossing closures. 2232 async def _wire_write(target_cv: int, target_byte: int) -> None: 2233 setup_payload: dict[str, Any] = { 2234 "cv": target_cv, "value": target_byte, 2235 "mode": extra.get("mode", "service"), 2236 } 2237 addr = extra.get("address") 2238 if addr is not None: 2239 setup_payload["address"] = addr 2240 setup_payload["isLongAddress"] = bool( 2241 extra.get("long_address"), 2242 ) 2243 await _dispatch( 2244 handle, type="programmer", method="post", data=setup_payload, 2245 ) 2246 2247 await _wire_write(31, var.split.bank or 0) 2248 await _wire_write(32, var.split.index or 0) 2249 store_key = variable_store_key(var, cv) 2250 return await _dispatch_read(cv, extra, store_key=store_key) 2251 return _read 2252 2253 def _find_variable_for_write(loco_id: str, item: str) -> Any: 2254 """Look up the decoder variable named *item* for *loco_id*. 2255 2256 Returns ``None`` when the catalog isn't loaded, the loco has no 2257 decoder family/model, or the item name doesn't match anything 2258 in the definition. Callers should fall back to a raw CV write 2259 in that case. 2260 """ 2261 if not item: 2262 return None 2263 try: 2264 from jmri_server.services.roster import ( # noqa: PLC0415 2265 _roster_service_get_sync, 2266 ) 2267 entry = _roster_service_get_sync(handle.ctx, loco_id) 2268 except Exception: # noqa: BLE001 2269 return None 2270 if not entry: 2271 return None 2272 family = entry.get("decoderFamily") or "" 2273 model = entry.get("decoderModel") or "" 2274 if not family or not model: 2275 return None 2276 defn = _catalog().find(family, model) 2277 if defn is None: 2278 return None 2279 for v in defn.variables: 2280 if v.item == item: 2281 return v 2282 return None 2283 2284 # CV 29's Address Format bit — set means "use long address", clear 2285 # means "use short". Any variable whose mask overlaps this bit will 2286 # flip the decoder's address mode when written. 2287 _CV29_ADDRESS_FORMAT_BIT = 0x20 2288 2289 def _ops_mode_address_change_reason( 2290 write: "BatchWrite", loco_id: str, 2291 ) -> str | None: 2292 """Return a human-readable reason if *write* would change the 2293 decoder's operating address, or ``None`` if it's safe in ops mode. 2294 2295 Ops-mode POM writes reach the decoder via a DCC packet targeted 2296 at the current DCC address. When the write itself changes that 2297 address (long-address CVs 17/18, short-address CV1, or the CV29 2298 Address Format bit), the wire packet is fine, but *subsequent* 2299 writes in the same batch aim at the old address and go 2300 unanswered. Long-address specifically requires two atomic CV 2301 writes (CV17 + CV18); doing them via POM leaves the decoder in 2302 an inconsistent state between the two. 2303 2304 We refuse rather than warn — the safe alternative is to run the 2305 exact same edit on the programming track (service mode), which 2306 the user can do by flipping the picker at the top of the pane. 2307 """ 2308 var = _find_variable_for_write(loco_id, write.item) 2309 if var is not None: 2310 if var.kind is VariableKind.LONG_ADDRESS: 2311 return ( 2312 "Long Address writes touch CV17 and CV18 atomically; " 2313 "ops mode can't do that safely because the address " 2314 "changes between the two packets. Use service mode." 2315 ) 2316 if var.kind is VariableKind.SHORT_ADDRESS: 2317 return ( 2318 "Short Address writes change the decoder's address; " 2319 "subsequent ops-mode packets would target the old " 2320 "address. Use service mode." 2321 ) 2322 # Any masked variable on CV29 whose ``V`` region overlaps 2323 # bit 5 flips the address-format mode. That'd break follow- 2324 # up POM packets in the same batch. 2325 primary = variable_primary_cv(var) 2326 if ( 2327 primary == 29 2328 and var.bit_mask.mask_value & _CV29_ADDRESS_FORMAT_BIT 2329 ): 2330 return ( 2331 "Changing CV29's Address Format bit switches the " 2332 "decoder between short and long addressing; " 2333 "subsequent ops-mode packets would target the old " 2334 "address. Use service mode." 2335 ) 2336 return None 2337 2338 @app.post("/programmer/loco/{loco_id}/write-batch") 2339 async def programmer_loco_write_batch( 2340 request: Request, loco_id: str, 2341 payload: dict[str, Any] = Body(...), 2342 ) -> Any: 2343 """Kick off a batch of CV writes for *loco_id*. 2344 2345 Payload shape:: 2346 2347 { 2348 "writes": [{"cv": 1, "value": 3, "item": "Short Address"}, ...], 2349 "mode": "service" | "ops", 2350 "address": <int>, // ops mode only 2351 "isLongAddress": true|false, 2352 "fileName": "1.xml" // optional; JMRI roster fileName 2353 } 2354 2355 Returns 202 + the batch id on success, or 409 if a batch is 2356 already in flight on this loco. Refuses with 400 when the 2357 batch is ops-mode and any write would change the decoder's 2358 operating address mid-batch (long address, short address, CV29 2359 Address Format flip). 2360 """ 2361 raw_writes = payload.get("writes") or [] 2362 if not isinstance(raw_writes, list) or not raw_writes: 2363 raise HTTPException(400, "writes must be a non-empty list") 2364 writes: list[BatchWrite] = [] 2365 for w in raw_writes: 2366 try: 2367 writes.append(BatchWrite( 2368 cv=int(w["cv"]), 2369 value=int(w["value"]), 2370 item=str(w.get("item", "")), 2371 )) 2372 except (KeyError, TypeError, ValueError) as e: 2373 raise HTTPException(400, f"malformed write row: {e}") from None 2374 2375 mode = str(payload.get("mode", "service")) 2376 address = payload.get("address") 2377 long_address = bool(payload.get("isLongAddress", False)) 2378 file_name = payload.get("fileName") or None 2379 2380 # Ops-mode safety gate: refuse any write that would change the 2381 # loco's operating address mid-batch. Check up front so we 2382 # don't start a batch that'd fail halfway through (CV17 2383 # written, CV18 aimed at wrong address). 2384 if mode == "ops": 2385 for w in writes: 2386 reason = _ops_mode_address_change_reason(w, loco_id) 2387 if reason is not None: 2388 raise HTTPException( 2389 400, 2390 f"cannot write {w.item or f'CV {w.cv}'} in ops mode: " 2391 f"{reason}", 2392 ) 2393 2394 queue = get_or_create_queue( 2395 handle.ctx, loco_id, _make_write_callback(loco_id), 2396 read_callback=_make_read_callback(loco_id), 2397 ) 2398 try: 2399 batch = await queue.start( 2400 writes, 2401 mode=mode, 2402 address=int(address) if address is not None else None, 2403 long_address=long_address, 2404 file_name=file_name, 2405 ) 2406 except BatchInFlightError as e: 2407 raise HTTPException(409, str(e)) from None 2408 from fastapi.responses import JSONResponse # noqa: PLC0415 2409 return JSONResponse( 2410 {"batch_id": batch.batch_id}, status_code=202, 2411 ) 2412 2413 @app.post("/programmer/loco/{loco_id}/read-batch") 2414 async def programmer_loco_read_batch( 2415 request: Request, loco_id: str, 2416 payload: dict[str, Any] = Body(...), 2417 ) -> Any: 2418 """Kick off a batch of CV reads for *loco_id*. 2419 2420 Payload shape:: 2421 2422 { 2423 "reads": [{"cv": 1, "item": "Short Address"}, ...], 2424 "mode": "service" | "ops", 2425 "address": <int>, // ops mode only (rarely 2426 // useful; PowerPro can't read 2427 // ops-mode anyway) 2428 "fileName": "1.xml" 2429 } 2430 2431 Returns 202 + the batch id, or 409 if a batch is already in 2432 flight. Duplicate CV numbers in the payload are deduplicated 2433 here — reading the same CV twice in one batch is wasted time on 2434 a service-mode track (each read is 2–6 s), and the store already 2435 caches whichever value came back last. 2436 """ 2437 raw_reads = payload.get("reads") or [] 2438 if not isinstance(raw_reads, list) or not raw_reads: 2439 raise HTTPException(400, "reads must be a non-empty list") 2440 # Dedupe by CV, first-wins on the display label so the user 2441 # sees a sensible item name in the progress fragment. 2442 seen_cvs: set[int] = set() 2443 ops: list[BatchWrite] = [] 2444 for r in raw_reads: 2445 try: 2446 cv = int(r["cv"]) 2447 except (KeyError, TypeError, ValueError) as e: 2448 raise HTTPException(400, f"malformed read row: {e}") from None 2449 if cv in seen_cvs: 2450 continue 2451 seen_cvs.add(cv) 2452 ops.append(BatchWrite( 2453 cv=cv, kind="read", item=str(r.get("item", "")), 2454 )) 2455 2456 mode = str(payload.get("mode", "service")) 2457 address = payload.get("address") 2458 long_address = bool(payload.get("isLongAddress", False)) 2459 file_name = payload.get("fileName") or None 2460 2461 queue = get_or_create_queue( 2462 handle.ctx, loco_id, _make_write_callback(loco_id), 2463 read_callback=_make_read_callback(loco_id), 2464 ) 2465 try: 2466 batch = await queue.start( 2467 ops, 2468 mode=mode, 2469 address=int(address) if address is not None else None, 2470 long_address=long_address, 2471 file_name=file_name, 2472 ) 2473 except BatchInFlightError as e: 2474 raise HTTPException(409, str(e)) from None 2475 from fastapi.responses import JSONResponse # noqa: PLC0415 2476 return JSONResponse( 2477 {"batch_id": batch.batch_id}, status_code=202, 2478 ) 2479 2480 @app.get("/programmer/loco/{loco_id}/batch/status") 2481 async def programmer_loco_batch_status(loco_id: str) -> Any: 2482 """Poll the current batch's status for *loco_id*. 2483 2484 Returns the batch's serializable snapshot or 404 if there's no 2485 active or recently-completed batch on this loco. 2486 """ 2487 queues = getattr(handle.ctx, "_programming_queues", {}) 2488 queue = queues.get(loco_id) if isinstance(queues, dict) else None 2489 if queue is None or queue.current is None: 2490 raise HTTPException(404, "no active batch") 2491 return queue.current.to_status() 2492 2493 @app.post("/programmer/loco/{loco_id}/batch/cancel") 2494 async def programmer_loco_batch_cancel(loco_id: str) -> Any: 2495 """Request cancel on the current batch. The in-flight CV write 2496 finishes; subsequent queued writes are marked cancelled.""" 2497 queues = getattr(handle.ctx, "_programming_queues", {}) 2498 queue = queues.get(loco_id) if isinstance(queues, dict) else None 2499 if queue is None or queue.current is None: 2500 raise HTTPException(404, "no active batch") 2501 batch = queue.cancel() 2502 return batch.to_status() if batch else {"state": "no-op"} 2503 2504 @app.post("/programmer/loco/{loco_id}/batch/clear") 2505 async def programmer_loco_batch_clear(loco_id: str) -> Any: 2506 """Drop a terminal-state batch so the UI can start fresh.""" 2507 queues = getattr(handle.ctx, "_programming_queues", {}) 2508 queue = queues.get(loco_id) if isinstance(queues, dict) else None 2509 if queue is None: 2510 return {"cleared": False} 2511 try: 2512 queue.clear() 2513 except BatchInFlightError as e: 2514 raise HTTPException(409, str(e)) from None 2515 return {"cleared": True} 2516 2517 @app.post("/programmer/catalog/load", response_class=HTMLResponse) 2518 async def programmer_load_catalog(request: Request, directory: str = Form(...)) -> Any: 2519 """Kick off a catalog load in the background and return a 2520 polling widget. Actual load happens on a worker thread because 2521 the JMRI decoder tree has hundreds of files and parsing takes 2522 30-60 s on a Pi. See :func:`programmer_catalog_progress` for 2523 the polling endpoint the returned widget hits.""" 2524 import asyncio # noqa: PLC0415 2525 import pathlib as _pathlib # noqa: PLC0415 2526 2527 target = _pathlib.Path(directory).expanduser() 2528 if not target.exists(): 2529 return templates.TemplateResponse( 2530 request, "_programmer_catalog.html", 2531 {"error": f"directory does not exist: {directory}"}, 2532 status_code=400, 2533 ) 2534 if not target.is_dir(): 2535 return templates.TemplateResponse( 2536 request, "_programmer_catalog.html", 2537 {"error": f"not a directory: {directory}"}, 2538 status_code=400, 2539 ) 2540 2541 # Shared state the loader worker writes and the poll route reads. 2542 # A single load is in flight at a time; overlapping loads are 2543 # rejected to keep progress state coherent. 2544 progress = getattr(handle.ctx, "_catalog_load_progress", None) 2545 if progress and not progress.get("done") and not progress.get("error"): 2546 return templates.TemplateResponse( 2547 request, "_programmer_catalog.html", 2548 {"error": "a catalog load is already in progress"}, 2549 status_code=409, 2550 ) 2551 2552 # File-count precomputed so the widget can render a percentage. 2553 files = list(target.rglob("*.xml")) 2554 total = len(files) 2555 handle.ctx._catalog_load_progress = { # type: ignore[attr-defined] 2556 "loaded": 0, 2557 "total": total, 2558 "done": False, 2559 "error": None, 2560 "directory": str(target), 2561 } 2562 2563 def _worker() -> None: 2564 state = handle.ctx._catalog_load_progress # type: ignore[attr-defined] 2565 try: 2566 # Fast metadata scan — extracts only <family> and 2567 # <model> attrs per file. Variables/panes are parsed 2568 # lazily by DecoderCatalog.find() on first loco open. 2569 # See jmri_core.persistence.decoder_xml.iter_decoder_index. 2570 from jmri_core.persistence.decoder_xml import ( # noqa: PLC0415 2571 iter_decoder_index, 2572 ) 2573 catalog = DecoderCatalog() 2574 seen_files: set[pathlib.Path] = set() 2575 for entry in iter_decoder_index(target): 2576 catalog.add_index_entry(entry) 2577 # Progress ticks per unique source file so the 2578 # thermometer maps to file count, not entry count 2579 # (one file can emit many entries). 2580 if entry.source_file not in seen_files: 2581 seen_files.add(entry.source_file) 2582 state["loaded"] += 1 2583 # Sync progress up to the total in case files were 2584 # skipped (non-decoder XML). The pane fragment reads 2585 # ``loaded/total`` to compute the percentage. 2586 state["loaded"] = state["total"] 2587 # Auto-load sibling programmers/ (same layout as 2588 # DecoderCatalog.from_directory). 2589 programmers_dir = target.parent / "programmers" 2590 if programmers_dir.is_dir(): 2591 from jmri_core.persistence.programmer_xml import ( # noqa: PLC0415 2592 iter_programmer_configs, 2593 ) 2594 catalog.programmers = { 2595 pc.name: pc 2596 for pc in iter_programmer_configs(programmers_dir) 2597 } 2598 # NMRA manufacturer table sits at the ``xml/`` root as 2599 # ``decoderIndex.xml``. Load it so ``identify_by_cv`` 2600 # has the authoritative 179-entry mfgID → name mapping 2601 # instead of the small hardcoded stopgap. 2602 index_path = target.parent / "decoderIndex.xml" 2603 if index_path.is_file(): 2604 from jmri_decoderpro.catalog import _load_mfg_ids # noqa: PLC0415 2605 catalog.mfg_ids = _load_mfg_ids(index_path) 2606 handle.ctx._decoder_catalog = catalog # type: ignore[attr-defined] 2607 except Exception as e: # noqa: BLE001 2608 state["error"] = str(e) 2609 state["done"] = True 2610 2611 # asyncio.to_thread runs the sync loader on the default thread 2612 # pool. Fire-and-forget; the poll endpoint watches ``progress``. 2613 asyncio.get_event_loop().create_task(asyncio.to_thread(_worker)) 2614 2615 return templates.TemplateResponse( 2616 request, "_programmer_catalog_progress.html", 2617 {"total": total, "directory": str(target)}, 2618 ) 2619 2620 @app.get("/programmer/catalog/progress", response_class=HTMLResponse) 2621 async def programmer_catalog_progress(request: Request) -> Any: 2622 """HTMX polls this every 500 ms while a load is in flight. 2623 2624 Returns: 2625 * a progress fragment (percentage bar) while ``done`` is False; 2626 * the final "Catalog loaded" fragment when ``done`` is True; 2627 * an error fragment when the worker set ``error``. 2628 """ 2629 progress = getattr(handle.ctx, "_catalog_load_progress", None) 2630 if not progress: 2631 # No load in flight — return the current catalog badge so 2632 # the poll loop stops (HTMX unbinds when the response no 2633 # longer includes the progress element). 2634 catalog = _catalog() 2635 return templates.TemplateResponse( 2636 request, "_programmer_catalog.html", 2637 { 2638 "catalog_size": len(catalog), 2639 "manufacturers": catalog.manufacturers(), 2640 "programmer_count": len(catalog.programmers), 2641 }, 2642 ) 2643 if progress.get("error"): 2644 return templates.TemplateResponse( 2645 request, "_programmer_catalog.html", 2646 {"error": f"failed to load: {progress['error']}"}, 2647 status_code=400, 2648 ) 2649 if progress.get("done"): 2650 catalog = _catalog() 2651 return templates.TemplateResponse( 2652 request, "_programmer_catalog.html", 2653 { 2654 "catalog_size": len(catalog), 2655 "manufacturers": catalog.manufacturers(), 2656 "programmer_count": len(catalog.programmers), 2657 }, 2658 ) 2659 return templates.TemplateResponse( 2660 request, "_programmer_catalog_progress.html", 2661 { 2662 "total": progress["total"], 2663 "loaded": progress["loaded"], 2664 "directory": progress["directory"], 2665 }, 2666 ) 2667 2668 @app.get("/programmer/decoders", response_class=HTMLResponse) 2669 async def programmer_decoders( 2670 request: Request, mfg: str | None = None, family: str | None = None, 2671 ) -> Any: 2672 if family: 2673 models = _catalog().models(family) 2674 return templates.TemplateResponse( 2675 request, "_programmer_models.html", 2676 {"family": family, "models": models}, 2677 ) 2678 if mfg: 2679 families = _catalog().families(mfg) 2680 return templates.TemplateResponse( 2681 request, "_programmer_families.html", 2682 {"mfg": mfg, "families": families}, 2683 ) 2684 return HTMLResponse("") 2685 2686 @app.get("/programmer/variables", response_class=HTMLResponse) 2687 async def programmer_variables( 2688 request: Request, family: str, model: str, 2689 ) -> Any: 2690 defn = _catalog().find(family, model) 2691 if defn is None: 2692 return HTMLResponse("decoder not found", status_code=404) 2693 return templates.TemplateResponse( 2694 request, "_programmer_variables.html", 2695 {"definition": defn, "family": family, "model": model}, 2696 ) 2697 2698 @app.post("/programmer/identify", response_class=HTMLResponse) 2699 async def programmer_identify(request: Request) -> Any: 2700 """Read CV7 + CV8 via the service-mode programmer, then look up 2701 matching definitions in the loaded catalog. 2702 2703 Failure modes surfaced clearly rather than as a spinning 2704 request: 2705 2706 * Read errored / timed out — the underlying error message goes 2707 into the fragment. 2708 * Read succeeded but CV8 isn't a known NMRA mfg ID (most often 2709 because the decoder didn't ack the read, so the CS returned 2710 its NoAck sentinel byte) — user sees "no decoder / unknown 2711 manufacturer" instead of the catalog quietly iterating 2712 thousands of files. 2713 """ 2714 from jmri_decoderpro import UnknownDecoderIdentityError # noqa: PLC0415 2715 # NCE PowerPro service-mode reads are ~2-6 s each and 2716 # intermittently return the NoAck sentinel (0/48/255). 2717 # Auto-retry once on those to avoid making the user click 2718 # Identify twice for a transient wire hiccup. Any hard 2719 # HTTPException short-circuits to the error card as before. 2720 _WIRE_FAIL = {0, 48, 255} 2721 async def _read_cv_with_retry(cv: int) -> dict: 2722 first = await _dispatch( 2723 handle, type="programmer", method="get", 2724 data={"cv": cv, "mode": "service"}, 2725 ) 2726 if first.get("value") in _WIRE_FAIL: 2727 # Sentinel value → very likely a transient read failure. 2728 # Do one silent retry; if it succeeds, use that. 2729 try: 2730 second = await _dispatch( 2731 handle, type="programmer", method="get", 2732 data={"cv": cv, "mode": "service"}, 2733 ) 2734 except HTTPException: 2735 return first 2736 if second.get("value") not in _WIRE_FAIL: 2737 return second 2738 return first 2739 try: 2740 cv7 = await _read_cv_with_retry(7) 2741 cv8 = await _read_cv_with_retry(8) 2742 except HTTPException as e: 2743 return templates.TemplateResponse( 2744 request, "_programmer_identify_result.html", 2745 {"error": e.detail}, status_code=e.status_code, 2746 ) 2747 try: 2748 candidates = _catalog().identify_by_cv(cv7["value"], cv8["value"]) 2749 except UnknownDecoderIdentityError as e: 2750 return templates.TemplateResponse( 2751 request, "_programmer_identify_result.html", 2752 {"cv7": cv7["value"], "cv8": cv8["value"], "error": str(e)}, 2753 ) 2754 # Manufacturer-specific productID probe. JMRI's real identify 2755 # is per-mfg: SoundTraxx encodes productID across CVs 253/255/ 2756 # 256, other makers use other CVs. Follow the same protocol so 2757 # a Tsunami2 read collapses from ~54 families to a single 2758 # model (e.g. "TSU-1100 Baldwin Diesels", productID 2071). 2759 product_id: str | None = None 2760 product_id_error: str | None = None 2761 if candidates and "soundtraxx" in candidates[0].mfg.lower(): 2762 try: 2763 cv253 = await _dispatch( 2764 handle, type="programmer", method="get", 2765 data={"cv": 253, "mode": "service"}, 2766 ) 2767 cv255 = await _dispatch( 2768 handle, type="programmer", method="get", 2769 data={"cv": 255, "mode": "service"}, 2770 ) 2771 cv256 = await _dispatch( 2772 handle, type="programmer", method="get", 2773 data={"cv": 256, "mode": "service"}, 2774 ) 2775 # JMRI Java's IdentifyDecoder formula: 2776 # productID = CV256 | ((CV255 & 7) << 8) | (CV253 << 11) 2777 pid = ( 2778 (cv256["value"] & 0xFF) 2779 | ((cv255["value"] & 0x07) << 8) 2780 | ((cv253["value"] & 0xFF) << 11) 2781 ) 2782 product_id = str(pid) 2783 except HTTPException as e: 2784 product_id_error = e.detail 2785 # Filter by productID when we have one that matches at least 2786 # one candidate; if nothing matches, keep the family list so 2787 # the user still has something to pick from. 2788 if product_id is not None: 2789 narrowed = [ 2790 c for c in candidates 2791 if c.product_id and c.product_id == product_id 2792 ] 2793 if narrowed: 2794 candidates = narrowed 2795 # Group candidates by (mfg, family) so the picker doesn't render 2796 # one row per model — a SoundTraxx match on a fully-loaded 2797 # catalog can produce 700+ models. The user picks the family 2798 # here and then the specific model from the Model dropdown 2799 # above (which is filtered to the chosen family). 2800 from collections import defaultdict # noqa: PLC0415 2801 by_family: dict[tuple[str, str], dict[str, Any]] = {} 2802 for c in candidates: 2803 key = (c.mfg, c.family) 2804 slot = by_family.setdefault(key, { 2805 "mfg": c.mfg, "family": c.family, 2806 "model_count": 0, "models": [], 2807 }) 2808 slot["model_count"] += 1 2809 slot["models"].append(c.model) 2810 family_candidates = [by_family[k] for k in sorted(by_family)] 2811 return templates.TemplateResponse( 2812 request, "_programmer_identify_result.html", 2813 { 2814 "cv7": cv7["value"], "cv8": cv8["value"], 2815 "candidates": family_candidates, 2816 "total_models": len(candidates), 2817 "product_id": product_id, 2818 "product_id_error": product_id_error, 2819 }, 2820 ) 2821 2822 @app.post("/programmer/cv/read", response_class=HTMLResponse) 2823 async def programmer_read_cv( 2824 request: Request, cv: int = Form(...), mode: str = Form("service"), 2825 address: int | None = Form(None), 2826 ) -> Any: 2827 payload: dict[str, Any] = {"cv": cv, "mode": mode} 2828 if address is not None: 2829 payload["address"] = address 2830 try: 2831 data = await _dispatch(handle, type="programmer", method="get", data=payload) 2832 except HTTPException as e: 2833 return templates.TemplateResponse( 2834 request, "_programmer_cv_result.html", 2835 {"cv": cv, "error": e.detail}, 2836 status_code=e.status_code, 2837 ) 2838 return templates.TemplateResponse( 2839 request, "_programmer_cv_result.html", 2840 {"cv": cv, "value": data["value"], "mode": data["mode"]}, 2841 ) 2842 2843 @app.post("/programmer/cv/write", response_class=HTMLResponse) 2844 async def programmer_write_cv( 2845 request: Request, cv: int = Form(...), value: int = Form(...), 2846 mode: str = Form("service"), address: int | None = Form(None), 2847 ) -> Any: 2848 payload: dict[str, Any] = {"cv": cv, "value": value, "mode": mode} 2849 if address is not None: 2850 payload["address"] = address 2851 try: 2852 data = await _dispatch(handle, type="programmer", method="post", data=payload) 2853 except HTTPException as e: 2854 return templates.TemplateResponse( 2855 request, "_programmer_cv_result.html", 2856 {"cv": cv, "error": e.detail}, 2857 status_code=e.status_code, 2858 ) 2859 return templates.TemplateResponse( 2860 request, "_programmer_cv_result.html", 2861 {"cv": cv, "value": data["value"], "mode": data["mode"], "wrote": True}, 2862 ) 2863 2864 # ----- scripts ---------------------------------------------------------- 2865 2866 def _repl_session(): 2867 """Return the shared REPL session, lazy-creating on first hit. 2868 2869 The session lives on ``ctx._repl_session`` so it survives 2870 across every browser tab hitting the /scripts page — matching 2871 JMRI Java's Jython window semantics (one shared interpreter). 2872 Rebuilt from scratch on ``/scripts/repl/reset``. 2873 """ 2874 from jmri_scripting import ReplSession # noqa: PLC0415 2875 cached = getattr(handle.ctx, "_repl_session", None) 2876 if cached is not None: 2877 return cached 2878 # Piggy-back on the ScriptRunner to reuse its shim binding — 2879 # the ScriptService lazy-constructs the runner on first use; 2880 # do the same dance here so both share one JmriShim, one 2881 # dispatch-stream install, one connection. 2882 runner = getattr(handle.ctx, "_script_runner", None) 2883 if runner is None: 2884 from jmri_scripting import ScriptRunner # noqa: PLC0415 2885 import asyncio as _asyncio # noqa: PLC0415 2886 runner = ScriptRunner.for_connection( 2887 _asyncio.get_event_loop(), handle.ctx.connection, 2888 ) 2889 handle.ctx._script_runner = runner # type: ignore[attr-defined] 2890 session = ReplSession(runner.shim) 2891 handle.ctx._repl_session = session # type: ignore[attr-defined] 2892 return session 2893 2894 @app.post("/scripts/repl", response_class=HTMLResponse) 2895 async def scripts_repl( 2896 request: Request, source: str = Form(""), 2897 ) -> Any: # type: ignore[valid-type] 2898 """Submit *source* to the persistent interpreter, render 2899 the console fragment with the updated history. 2900 2901 Empty submissions are no-ops (rendering just refreshes 2902 the console) — matches the CPython REPL's "just hit 2903 Enter" behavior at the prompt.""" 2904 session = _repl_session() 2905 pending: str | None = None 2906 if source.strip(): 2907 # ``submit_async`` off-loads the exec to a worker 2908 # thread so shim calls that block on our event loop 2909 # don't deadlock — the loop stays free to advance 2910 # the shim's scheduled coroutines. 2911 result = await session.submit_async(source) 2912 if result.incomplete: 2913 pending = source 2914 return templates.TemplateResponse( 2915 request, "_repl_console.html", 2916 { 2917 "history": session.history, 2918 "pending_continuation": pending, 2919 }, 2920 ) 2921 2922 @app.post("/scripts/repl/reset", response_class=HTMLResponse) 2923 async def scripts_repl_reset(request: Request) -> Any: # type: ignore[valid-type] 2924 session = _repl_session() 2925 session.reset() 2926 return templates.TemplateResponse( 2927 request, "_repl_console.html", 2928 {"history": session.history, "pending_continuation": None}, 2929 ) 2930 2931 @app.get("/scripts", response_class=HTMLResponse) 2932 async def scripts_home(request: Request) -> Any: 2933 statuses = await _dispatch( 2934 handle, type="scripts", method="list", 2935 data={"library": True}, 2936 ) 2937 session = _repl_session() 2938 return templates.TemplateResponse( 2939 request, "scripts.html", 2940 _base_context(handle, "scripts") | { 2941 "statuses": statuses, 2942 "history": session.history, 2943 "pending_continuation": None, 2944 }, 2945 ) 2946 2947 @app.post("/scripts/run", response_class=HTMLResponse) 2948 async def scripts_run( 2949 request: Request, 2950 name: str = Form(...), 2951 source: str = Form(...), 2952 save: str | None = Form(None), 2953 ) -> Any: 2954 await _dispatch( 2955 handle, type="scripts", method="put", 2956 data={"name": name, "source": source, "save": bool(save)}, 2957 ) 2958 # Give the worker thread a moment to populate status. 2959 import asyncio as _asyncio # noqa: PLC0415 2960 await _asyncio.sleep(0.05) 2961 statuses = await _dispatch( 2962 handle, type="scripts", method="list", data={"library": True}, 2963 ) 2964 return templates.TemplateResponse( 2965 request, "_scripts_table.html", {"statuses": statuses}, 2966 ) 2967 2968 @app.post("/scripts/run-file/{name}", response_class=HTMLResponse) 2969 async def scripts_run_file(request: Request, name: str) -> Any: 2970 """Run an on-disk script by name. 2971 2972 Distinct from ``/scripts/run`` which takes fresh source 2973 from the textarea. This route re-reads the file from disk 2974 every call, so the debug loop (edit externally → click 2975 Run → see output) works. The runner replaces any 2976 same-named existing status entry so re-runs stack cleanly. 2977 """ 2978 try: 2979 await _dispatch( 2980 handle, type="scripts", method="post", 2981 data={"name": name, "action": "run_file"}, 2982 ) 2983 except HTTPException as exc: 2984 # Surface the 404 / permission error as a plain-text 2985 # banner in the table region so htmx doesn't blank the 2986 # UI on a stray click. 2987 statuses = await _dispatch( 2988 handle, type="scripts", method="list", 2989 data={"library": True}, 2990 ) 2991 return templates.TemplateResponse( 2992 request, "_scripts_table.html", 2993 {"statuses": statuses, "run_error": exc.detail}, 2994 ) 2995 # Same 50ms grace as /scripts/run so a fast-finishing 2996 # script has a chance to populate its status before the 2997 # table renders. 2998 import asyncio as _asyncio # noqa: PLC0415 2999 await _asyncio.sleep(0.05) 3000 statuses = await _dispatch( 3001 handle, type="scripts", method="list", data={"library": True}, 3002 ) 3003 return templates.TemplateResponse( 3004 request, "_scripts_table.html", {"statuses": statuses}, 3005 ) 3006 3007 @app.post("/scripts/stop/{name}", response_class=HTMLResponse) 3008 async def scripts_stop(request: Request, name: str) -> Any: 3009 await _dispatch( 3010 handle, type="scripts", method="post", 3011 data={"name": name, "action": "stop"}, 3012 ) 3013 statuses = await _dispatch( 3014 handle, type="scripts", method="list", data={"library": True}, 3015 ) 3016 return templates.TemplateResponse( 3017 request, "_scripts_table.html", {"statuses": statuses}, 3018 ) 3019 3020 @app.post("/scripts/delete/{name}", response_class=HTMLResponse) 3021 async def scripts_delete(request: Request, name: str) -> Any: 3022 """Stop the named script AND remove its on-disk file so the 3023 profile autoload doesn't bring it back.""" 3024 await _dispatch( 3025 handle, type="scripts", method="post", 3026 data={"name": name, "action": "delete"}, 3027 ) 3028 statuses = await _dispatch( 3029 handle, type="scripts", method="list", data={"library": True}, 3030 ) 3031 return templates.TemplateResponse( 3032 request, "_scripts_table.html", {"statuses": statuses}, 3033 ) 3034 3035 @app.get("/scripts/refresh", response_class=HTMLResponse) 3036 async def scripts_refresh(request: Request) -> Any: 3037 """Re-render the scripts table with the latest captured output. 3038 3039 Called from the per-row Refresh button so a long-running 3040 automaton's accumulated ``print()`` output surfaces without the 3041 user having to trigger a stop/start. 3042 """ 3043 statuses = await _dispatch( 3044 handle, type="scripts", method="list", data={"library": True}, 3045 ) 3046 return templates.TemplateResponse( 3047 request, "_scripts_table.html", {"statuses": statuses}, 3048 ) 3049 3050 # ----- profile CRUD ----------------------------------------------------- 3051 # Wraps the ``configProfile`` JSON topic. Post-mutation we redirect 3052 # back to ``/`` so the browser re-fetches the profile list — a 3053 # cleaner UX than a partial fragment when the "Load profile" form 3054 # (already on the page) needs to reflect the new list too. 3055 3056 @app.post("/profiles/create", response_class=HTMLResponse) 3057 async def profiles_create( 3058 request: Request, 3059 name: str = Form(...), 3060 ) -> Any: 3061 from fastapi.responses import RedirectResponse # noqa: PLC0415 3062 try: 3063 await _dispatch(handle, type="configProfile", method="put", 3064 data={"name": name}) 3065 # Rebuild profile_registry from disk so subsequent calls see 3066 # the new entry. The in-memory registry is what we already 3067 # mutated, so this is optional — but paranoid safety. 3068 except HTTPException as exc: 3069 # Render home with an error banner instead of a bare 4xx page. 3070 return await _home_with_error(request, exc.detail) 3071 return RedirectResponse(url="/", status_code=303) 3072 3073 @app.post("/profiles/{profile_id}/delete", response_class=HTMLResponse) 3074 async def profiles_delete(request: Request, profile_id: str) -> Any: 3075 from fastapi.responses import RedirectResponse # noqa: PLC0415 3076 try: 3077 await _dispatch(handle, type="configProfile", method="delete", 3078 data={"id": profile_id}) 3079 except HTTPException as exc: 3080 return await _home_with_error(request, exc.detail) 3081 return RedirectResponse(url="/", status_code=303) 3082 3083 async def _home_with_error(request: Request, message: str) -> Any: 3084 """Re-render the home page carrying an error banner. Used by 3085 the profile-CRUD routes when the JSON service rejects a 3086 request (duplicate name, active profile delete, etc.).""" 3087 power = await _dispatch(handle, type="power", method="get") 3088 fast_clock = await _dispatch(handle, type="time", method="get") 3089 registry = getattr(handle.ctx, "profile_registry", None) 3090 active = registry.active() if registry is not None else None 3091 log_tail_ctx = await _log_tail_context(handle) 3092 return templates.TemplateResponse( 3093 request, "home.html", 3094 _base_context(handle, "home") | { 3095 "power": power, 3096 "fast_clock": fast_clock, 3097 "connection": handle.ctx.connection, 3098 "command_station_on": bool( 3099 getattr(handle.ctx.connection, 3100 "is_command_station_on", False) 3101 ), 3102 "topics": _topics_with_help(handle), 3103 "profiles": [ 3104 {"id": p.id, "name": p.name, "path": p.path} 3105 for p in (registry.profiles if registry is not None else []) 3106 ], 3107 "active_profile_id": active.id if active else None, 3108 "settings_root": ( 3109 str(registry.settings_root) if registry is not None else None 3110 ), 3111 "profile_error": message, 3112 } | log_tail_ctx, 3113 ) 3114 3115 # ----- cross-profile bean transfer -------------------------------------- 3116 3117 def _resolve_profile_name(profile_id: str) -> str: 3118 """Look up the friendly name for a profile id via the 3119 registry. Falls back to the raw id when the registry hasn't 3120 been loaded (bare test setups).""" 3121 registry = getattr(handle.ctx, "profile_registry", None) 3122 if registry is None: 3123 return profile_id 3124 ref = registry.by_id(profile_id) 3125 return ref.name if ref is not None else profile_id 3126 3127 def _target_is_active(profile_id: str) -> bool: 3128 registry = getattr(handle.ctx, "profile_registry", None) 3129 if registry is None: 3130 return False 3131 active = registry.active() 3132 return active is not None and active.id == profile_id 3133 3134 @app.post("/profiles/transfer/preview", response_class=HTMLResponse) 3135 async def profiles_transfer_preview( 3136 request: Request, 3137 from_id: str = Form(...), 3138 to_id: str = Form(...), 3139 types: list[str] = Form(default=[]), 3140 ) -> Any: 3141 try: 3142 data = await _dispatch(handle, type="beanTransfer", method="post", data={ 3143 "action": "preview", 3144 "from": from_id, "to": to_id, "types": types, 3145 }) 3146 except HTTPException as exc: 3147 return templates.TemplateResponse( 3148 request, "_bean_transfer_preview.html", 3149 {"error": exc.detail}, 3150 ) 3151 return templates.TemplateResponse( 3152 request, "_bean_transfer_preview.html", 3153 { 3154 "counts": data.get("counts", {}), 3155 "will_overwrite": data.get("willOverwrite", {}), 3156 "from_id": from_id, 3157 "to_id": to_id, 3158 "from_name": _resolve_profile_name(from_id), 3159 "to_name": _resolve_profile_name(to_id), 3160 "to_is_active": _target_is_active(to_id), 3161 "types": types, 3162 }, 3163 ) 3164 3165 @app.post("/profiles/transfer/apply", response_class=HTMLResponse) 3166 async def profiles_transfer_apply( 3167 request: Request, 3168 from_id: str = Form(...), 3169 to_id: str = Form(...), 3170 types: list[str] = Form(default=[]), 3171 ) -> Any: 3172 try: 3173 data = await _dispatch(handle, type="beanTransfer", method="post", data={ 3174 "action": "apply", 3175 "from": from_id, "to": to_id, "types": types, 3176 }) 3177 except HTTPException as exc: 3178 return templates.TemplateResponse( 3179 request, "_bean_transfer_applied.html", 3180 {"error": exc.detail}, 3181 ) 3182 return templates.TemplateResponse( 3183 request, "_bean_transfer_applied.html", 3184 { 3185 "copied": data.get("copied", {}), 3186 "target_is_active": data.get("targetIsActive", False), 3187 }, 3188 ) 3189 3190 # ----- signal systems ---------------------------------------------------- 3191 3192 @app.get("/signalsystems", response_class=HTMLResponse) 3193 async def signal_systems_home(request: Request) -> Any: 3194 systems = await _dispatch( 3195 handle, type="signalSystem", method="list", 3196 ) 3197 return templates.TemplateResponse( 3198 request, "signalsystems.html", 3199 _base_context(handle, "signalsystems") | {"systems": systems}, 3200 ) 3201 3202 # ----- logs widget ------------------------------------------------------- 3203 3204 _LOG_TAIL_MAX = 20 3205 3206 async def _log_tail_context(_handle: "AppHandle") -> dict[str, Any]: 3207 """Build the recent-log widget's template context. 3208 3209 Fetches via the ``logs`` service so the widget and any script 3210 or mobile client see byte-for-byte the same tail — no chance 3211 of divergent formatting between the widget and the JSON API. 3212 """ 3213 try: 3214 snapshot = await _dispatch( 3215 handle, type="logs", method="get", 3216 data={"tail": _LOG_TAIL_MAX}, 3217 ) 3218 except HTTPException: 3219 snapshot = {"path": None, "tail": _LOG_TAIL_MAX, "lines": []} 3220 return { 3221 "log_path": snapshot.get("path"), 3222 "log_lines": snapshot.get("lines") or [], 3223 "log_tail_max": snapshot.get("tail", _LOG_TAIL_MAX), 3224 } 3225 3226 @app.get("/logs/refresh", response_class=HTMLResponse) 3227 async def logs_refresh(request: Request) -> Any: 3228 ctx = await _log_tail_context(handle) 3229 return templates.TemplateResponse( 3230 request, "_log_tail.html", ctx, 3231 ) 3232 3233 # ----- logixng ----------------------------------------------------------- 3234 3235 async def _logixng_list_fragment(request: Request) -> Any: 3236 programs = await _dispatch(handle, type="logixng", method="list") 3237 return templates.TemplateResponse( 3238 request, "_logixng_list.html", {"programs": programs}, 3239 ) 3240 3241 @app.get("/logixng", response_class=HTMLResponse) 3242 async def logixng_home(request: Request) -> Any: 3243 programs = await _dispatch(handle, type="logixng", method="list") 3244 return templates.TemplateResponse( 3245 request, "logixng.html", 3246 _base_context(handle, "logixng") | {"programs": programs}, 3247 ) 3248 3249 @app.get("/logixng/refresh", response_class=HTMLResponse) 3250 async def logixng_refresh(request: Request) -> Any: 3251 return await _logixng_list_fragment(request) 3252 3253 @app.post("/logixng/create", response_class=HTMLResponse) 3254 async def logixng_create( 3255 request: Request, 3256 name: str = Form(...), 3257 userName: str = Form(""), 3258 ) -> Any: 3259 try: 3260 await _dispatch(handle, type="logixng", method="put", data={ 3261 "name": name, "userName": userName, 3262 }) 3263 except HTTPException as exc: 3264 programs = await _dispatch( 3265 handle, type="logixng", method="list", 3266 ) 3267 return templates.TemplateResponse( 3268 request, "_logixng_list.html", 3269 {"programs": programs, "create_error": exc.detail}, 3270 status_code=200, 3271 ) 3272 return await _logixng_list_fragment(request) 3273 3274 @app.post("/logixng/import-xml", response_class=HTMLResponse) 3275 async def logixng_import_xml( 3276 request: Request, file: UploadFile, 3277 ) -> Any: 3278 """Convert a JMRI Java LogixNG XML file into our tree 3279 format and load every program into the running manager. 3280 3281 Bypasses the JSON put path so we can (a) skip re-encoding 3282 each program to a dict and back, and (b) collect 3283 importer warnings for surfacing to the operator. Each 3284 program is persisted so a server restart doesn't lose the 3285 import.""" 3286 from jmri_logixng import import_jmri_logixng_xml # noqa: PLC0415 3287 from jmri_server.services.logixng import _require_manager # noqa: PLC0415 3288 3289 raw = await file.read() 3290 if not raw: 3291 return templates.TemplateResponse( 3292 request, "_logixng_import_result.html", 3293 {"error": "empty file"}, 3294 status_code=400, 3295 ) 3296 try: 3297 result = import_jmri_logixng_xml(raw) 3298 except Exception as e: # noqa: BLE001 3299 return templates.TemplateResponse( 3300 request, "_logixng_import_result.html", 3301 {"error": f"failed to parse XML: {e}"}, 3302 status_code=400, 3303 ) 3304 3305 # Install each imported program under its JMRI system name 3306 # (the ``node_id`` we recorded during walk_programs). Persist 3307 # to the profile's logixng dir so the import survives a 3308 # restart — same behaviour as the JSON put path. 3309 try: 3310 mgr = _require_manager(handle.ctx) 3311 except HTTPException as exc: 3312 return templates.TemplateResponse( 3313 request, "_logixng_import_result.html", 3314 {"error": f"manager unavailable: {exc.detail}"}, 3315 status_code=503, 3316 ) 3317 logixng_dir = getattr(handle.ctx, "_logixng_dir", None) 3318 imported_names: list[str] = [] 3319 for program in result.programs: 3320 # Node stores the constructor's ``node_id`` on ``.id`` 3321 # (with a uuid-suffixed fallback when the caller 3322 # didn't supply one). Prefer that, fall back to 3323 # user_name for the rare case the imported program 3324 # had neither. 3325 raw_name = program.id or program.user_name or "imported" 3326 # Colons are legal in JMRI Java system names (``IQ:AUTO:0001``) 3327 # but not in Windows filenames — replace with dots so the 3328 # persisted YAML file is portable. On Linux/Pi the raw 3329 # name works fine; keeping the sanitization uniform 3330 # simplifies cross-platform testing. 3331 name = raw_name.replace(":", ".") 3332 await mgr.put(name, program) 3333 imported_names.append(name) 3334 if logixng_dir is not None: 3335 mgr.save_one(logixng_dir, name) 3336 3337 # Load programs list for the out-of-band swap. 3338 programs = await _dispatch( 3339 handle, request, type="logixng", method="list", 3340 ) 3341 return templates.TemplateResponse( 3342 request, "_logixng_import_result.html", 3343 { 3344 "imported": imported_names, 3345 "warnings": result.warnings, 3346 "programs": programs, 3347 }, 3348 ) 3349 3350 @app.post("/logixng/{name}/delete", response_class=HTMLResponse) 3351 async def logixng_delete(request: Request, name: str) -> Any: 3352 try: 3353 await _dispatch( 3354 handle, type="logixng", method="delete", data={"name": name}, 3355 ) 3356 except HTTPException: 3357 pass 3358 return await _logixng_list_fragment(request) 3359 3360 @app.post("/logixng/{name}/enable", response_class=HTMLResponse) 3361 async def logixng_enable(request: Request, name: str) -> Any: 3362 await _dispatch(handle, type="logixng", method="post", data={ 3363 "name": name, "command": "enable", 3364 }) 3365 return await _logixng_list_fragment(request) 3366 3367 @app.post("/logixng/{name}/disable", response_class=HTMLResponse) 3368 async def logixng_disable(request: Request, name: str) -> Any: 3369 await _dispatch(handle, type="logixng", method="post", data={ 3370 "name": name, "command": "disable", 3371 }) 3372 return await _logixng_list_fragment(request) 3373 3374 @app.post("/logixng/{name}/run", response_class=HTMLResponse) 3375 async def logixng_run(request: Request, name: str) -> Any: 3376 try: 3377 await _dispatch(handle, type="logixng", method="post", data={ 3378 "name": name, "command": "run", 3379 }) 3380 except HTTPException as exc: 3381 programs = await _dispatch( 3382 handle, type="logixng", method="list", 3383 ) 3384 return templates.TemplateResponse( 3385 request, "_logixng_list.html", 3386 {"programs": programs, "run_error": exc.detail}, 3387 status_code=200, 3388 ) 3389 return await _logixng_list_fragment(request) 3390 3391 @app.post("/logixng/{name}/run-conditional", response_class=HTMLResponse) 3392 async def logixng_run_conditional( 3393 request: Request, name: str, 3394 conditional_id: str = Form(...), 3395 ) -> Any: 3396 try: 3397 await _dispatch(handle, type="logixng", method="post", data={ 3398 "name": name, "command": "run_conditional", 3399 "conditional_id": conditional_id, 3400 }) 3401 except HTTPException as exc: 3402 programs = await _dispatch( 3403 handle, type="logixng", method="list", 3404 ) 3405 return templates.TemplateResponse( 3406 request, "_logixng_list.html", 3407 {"programs": programs, "run_error": exc.detail}, 3408 status_code=200, 3409 ) 3410 return await _logixng_list_fragment(request) 3411 3412 # ----- consists ---------------------------------------------------------- 3413 3414 async def _consist_table_fragment(request: Request) -> Any: 3415 consists = await _dispatch(handle, type="consist", method="list") 3416 return templates.TemplateResponse( 3417 request, "_consists_table.html", {"consists": consists}, 3418 ) 3419 3420 @app.get("/consists", response_class=HTMLResponse) 3421 async def consists_home(request: Request) -> Any: 3422 consists = await _dispatch(handle, type="consist", method="list") 3423 return templates.TemplateResponse( 3424 request, "consists.html", 3425 _base_context(handle, "consists") | {"consists": consists}, 3426 ) 3427 3428 @app.get("/consists/refresh", response_class=HTMLResponse) 3429 async def consists_refresh(request: Request) -> Any: 3430 return await _consist_table_fragment(request) 3431 3432 @app.post("/consists/create", response_class=HTMLResponse) 3433 async def consists_create( 3434 request: Request, 3435 address: int = Form(...), 3436 kind: str = Form("command_station"), 3437 comment: str = Form(""), 3438 ) -> Any: 3439 # Consist addresses are short 1..127 by spec (CV19); the service 3440 # enforces range + roster-collision. Any error there comes back 3441 # as an HTTPException — render it into the notice so htmx sees a 3442 # 200 with the message rather than swapping in the FastAPI 3443 # default JSON error page. 3444 try: 3445 await _dispatch(handle, type="consist", method="put", data={ 3446 "address": address, "kind": kind, "comment": comment, 3447 }) 3448 except HTTPException as exc: 3449 return templates.TemplateResponse( 3450 request, "_consists_table.html", 3451 { 3452 "consists": await _dispatch( 3453 handle, type="consist", method="list" 3454 ), 3455 "create_error": exc.detail, 3456 }, 3457 status_code=200, 3458 ) 3459 return await _consist_table_fragment(request) 3460 3461 @app.post("/consists/{address}/delete", response_class=HTMLResponse) 3462 async def consists_delete(request: Request, address: int) -> Any: 3463 try: 3464 await _dispatch( 3465 handle, type="consist", method="delete", 3466 data={"address": address}, 3467 ) 3468 except HTTPException: 3469 pass # 404 — table refresh below will just show whatever's left. 3470 return await _consist_table_fragment(request) 3471 3472 @app.post("/consists/{address}/add-member", response_class=HTMLResponse) 3473 async def consists_add_member( 3474 request: Request, address: int, 3475 member_address: int = Form(...), 3476 long_address: str | None = Form(None), 3477 reversed: str | None = Form(None), 3478 ) -> Any: 3479 # Form has a "Reversed" checkbox — unchecked (the default) means 3480 # the loco is oriented forward in the consist. 3481 await _dispatch(handle, type="consist", method="post", data={ 3482 "address": address, "action": "add_member", 3483 "member": { 3484 "address": member_address, 3485 "long_address": bool(long_address), 3486 "direction_normal": not bool(reversed), 3487 }, 3488 }) 3489 return await _consist_table_fragment(request) 3490 3491 @app.post("/consists/{address}/remove-member", response_class=HTMLResponse) 3492 async def consists_remove_member( 3493 request: Request, address: int, 3494 member_address: int = Form(...), 3495 ) -> Any: 3496 try: 3497 await _dispatch(handle, type="consist", method="post", data={ 3498 "address": address, "action": "remove_member", 3499 "member_address": member_address, 3500 }) 3501 except HTTPException: 3502 pass 3503 return await _consist_table_fragment(request) 3504 3505 @app.post("/consists/{address}/set-speed", response_class=HTMLResponse) 3506 async def consists_set_speed( 3507 request: Request, address: int, speed: float = Form(...), 3508 ) -> Any: 3509 await _dispatch(handle, type="consist", method="post", data={ 3510 "address": address, "action": "set_speed", "speed": speed, 3511 }) 3512 return await _consist_table_fragment(request) 3513 3514 @app.post("/consists/{address}/direction", response_class=HTMLResponse) 3515 async def consists_set_direction( 3516 request: Request, address: int, direction: str = Form(...), 3517 ) -> Any: 3518 await _dispatch(handle, type="consist", method="post", data={ 3519 "address": address, "action": "set_direction", "direction": direction, 3520 }) 3521 return await _consist_table_fragment(request) 3522 3523 @app.post("/consists/{address}/estop", response_class=HTMLResponse) 3524 async def consists_estop(request: Request, address: int) -> Any: 3525 await _dispatch(handle, type="consist", method="post", data={ 3526 "address": address, "action": "emergency_stop", 3527 }) 3528 return await _consist_table_fragment(request) 3529 3530 @app.post("/panel/upload", response_class=HTMLResponse) 3531 async def panel_upload(request: Request, file: UploadFile) -> Any: 3532 from jmri_core.persistence import load_panel_bytes # noqa: PLC0415 3533 3534 data = await file.read() 3535 try: 3536 panel_file = load_panel_bytes(data) 3537 except Exception as e: # noqa: BLE001 3538 return templates.TemplateResponse( 3539 request, "_panel_summary.html", 3540 {"error": f"failed to parse: {e}"}, 3541 status_code=400, 3542 ) 3543 summary = { 3544 "filename": file.filename, 3545 "root_tag": panel_file.root_tag, 3546 "jmri_version": panel_file.jmri_version, 3547 "panels": [p.get("name") or "(unnamed)" for p in panel_file.panels()], 3548 "layout_editors": [ 3549 le.get("name") or "(unnamed)" for le in panel_file.layout_editors() 3550 ], 3551 "top_level_child_count": len(list(panel_file.root)), 3552 } 3553 return templates.TemplateResponse( 3554 request, "_panel_summary.html", {"summary": summary}, 3555 )