Coverage for nexusLIMS/tui/apps/config/screens.py: 100%
682 statements
« prev ^ index » next coverage.py v7.11.3, created at 2026-03-24 05:23 +0000
1"""
2Screens for the NexusLIMS configuration TUI.
4Provides :class:`ConfigScreen` (the main tabbed form) and
5:class:`FieldDetailScreen` (popup help modal for configuration fields).
6"""
8import contextlib
9import json
10import re
11from pathlib import Path
12from typing import ClassVar
14import pytz
15from dotenv import dotenv_values
16from pydantic_core import PydanticUndefined
17from textual import on
18from textual.app import ComposeResult
19from textual.containers import Horizontal, Vertical, VerticalScroll
20from textual.screen import ModalScreen, Screen
21from textual.widgets import (
22 Button,
23 Footer,
24 Header,
25 Input,
26 Label,
27 Select,
28 Static,
29 Switch,
30 TabbedContent,
31 TabPane,
32 Tabs,
33 TextArea,
34)
36from nexusLIMS.cli.config import (
37 _flatten_to_env,
38 _write_env_file,
39)
40from nexusLIMS.config import EmailConfig, NemoHarvesterConfig, Settings
41from nexusLIMS.tui.apps.config.validators import (
42 validate_float_nonneg,
43 validate_float_positive,
44 validate_nemo_address,
45 validate_optional_iana_timezone,
46 validate_optional_int,
47 validate_optional_url,
48 validate_smtp_port,
49)
50from nexusLIMS.tui.common.base_screens import ConfirmDialog
51from nexusLIMS.tui.common.validators import validate_required, validate_url
52from nexusLIMS.tui.common.widgets import AutocompleteInput, FormField
54# --------------------------------------------------------------------------- #
55# Constants #
56# --------------------------------------------------------------------------- #
58_DEFAULT_STRFTIME = "%Y-%m-%dT%H:%M:%S%z"
59_DEFAULT_STRPTIME = "%Y-%m-%dT%H:%M:%S%z"
62# --------------------------------------------------------------------------- #
63# Helpers: pull descriptions/defaults from the Settings model #
64# --------------------------------------------------------------------------- #
67def _fdesc(name: str) -> str:
68 """Return the field description from Settings for the given env var name."""
69 field = Settings.model_fields.get(name)
70 if field and field.description:
71 return field.description
72 return ""
75def _fdefault(name: str) -> str:
76 """Return the field default from Settings as a string, or empty string."""
77 field = Settings.model_fields.get(name)
78 if field is None:
79 return ""
80 default = field.default
81 if default is PydanticUndefined or default is None:
82 return ""
83 if isinstance(default, list):
84 return ", ".join(str(v) for v in default)
85 return str(default)
88def _edesc(name: str) -> str:
89 """Return the field description from EmailConfig for the given field name."""
90 field = EmailConfig.model_fields.get(name)
91 if field and field.description:
92 return field.description
93 return ""
96def _edefault(name: str) -> str:
97 """Return the field default from EmailConfig as a string, or empty string."""
98 field = EmailConfig.model_fields.get(name)
99 if field is None:
100 return ""
101 default = field.default
102 if default is PydanticUndefined or default is None:
103 return ""
104 return str(default)
107def _md_to_rich(text: str) -> str:
108 """Convert a small subset of Markdown to Rich markup for TUI display.
110 Handles:
111 - ``[label](url)`` → ``label (url)``
112 - bare URLs (http/https not inside a markdown link) → plain text URL
113 - `` `code` `` → ``[bold]code[/bold]``
115 Uses a single combined regex pass so that matched spans are never
116 processed a second time, avoiding broken nested markup.
117 """
118 # Combined pattern — alternatives are tried left-to-right, first match wins:
119 # 1. backtick code span
120 # 2. markdown link [label](url)
121 # 3. bare http/https URL
122 pattern = re.compile(
123 r"`([^`]+)`" # group 1: backtick code span
124 r"|"
125 r"\[([^\]]+)\]\((https?://[^\)]+)\)" # group 2+3: markdown link
126 r"|"
127 r"(https?://\S+)" # group 4: bare URL
128 )
130 def _replace(m: re.Match) -> str:
131 if m.group(1) is not None:
132 return f"[bold]{m.group(1)}[/bold]"
133 if m.group(2) is not None:
134 # Render as "label (url)" — avoids MarkupError with URLs in [link=...]
135 return f"{m.group(2)} ({m.group(3)})"
136 # Bare URL — render as plain text to avoid MarkupError with "://"
137 return m.group(4)
139 return pattern.sub(_replace, text)
142def _fdetail(name: str) -> str:
143 """Return extended detail text from Settings.json_schema_extra['detail']."""
144 field = Settings.model_fields.get(name)
145 if field is None:
146 return ""
147 jse = getattr(field, "json_schema_extra", None) or {}
148 if callable(jse):
149 return ""
150 return _md_to_rich(jse.get("detail", ""))
153def _edetail(name: str) -> str:
154 """Return extended detail text from EmailConfig.json_schema_extra['detail']."""
155 field = EmailConfig.model_fields.get(name)
156 if field is None:
157 return ""
158 jse = getattr(field, "json_schema_extra", None) or {}
159 if callable(jse):
160 return ""
161 return _md_to_rich(jse.get("detail", ""))
164def _ndetail(name: str) -> str:
165 """Return extended detail text from NemoHarvesterConfig for the given field."""
166 field = NemoHarvesterConfig.model_fields.get(name)
167 if field is None:
168 return ""
169 jse = getattr(field, "json_schema_extra", None) or {}
170 if callable(jse):
171 return ""
172 return _md_to_rich(jse.get("detail", ""))
175# Maps NEMO Input widget id prefixes → NemoHarvesterConfig field name.
176# Used for dynamic ids like "nemo-address-1", "nemo-token-2", etc.
177_NEMO_INPUT_PREFIX_TO_FIELD: dict[str, str] = {
178 "nemo-address-": "address",
179 "nemo-token-": "token",
180 "nemo-tz-": "tz",
181 "nemo-strftime-": "strftime_fmt",
182 "nemo-strptime-": "strptime_fmt",
183}
185# Maps Input widget ids → (model_class, field_name) for detail lookup.
186# Select widgets (nx-file-strategy, nx-export-strategy) and TextArea
187# (nx-cert-bundle) are handled inline in action_show_field_detail.
188# Switch widgets with detail text are handled inline in action_show_field_detail.
189_INPUT_ID_TO_FIELD: dict[str, tuple[str, str]] = {
190 "nx-instrument-data-path": ("settings", "NX_INSTRUMENT_DATA_PATH"),
191 "nx-data-path": ("settings", "NX_DATA_PATH"),
192 "nx-db-path": ("settings", "NX_DB_PATH"),
193 "nx-log-path": ("settings", "NX_LOG_PATH"),
194 "nx-records-path": ("settings", "NX_RECORDS_PATH"),
195 "nx-local-profiles-path": ("settings", "NX_LOCAL_PROFILES_PATH"),
196 "nx-cdcs-url": ("settings", "NX_CDCS_URL"),
197 "nx-cdcs-token": ("settings", "NX_CDCS_TOKEN"),
198 "nx-ignore-patterns": ("settings", "NX_IGNORE_PATTERNS"),
199 "nx-file-delay-days": ("settings", "NX_FILE_DELAY_DAYS"),
200 "nx-clustering-sensitivity": ("settings", "NX_CLUSTERING_SENSITIVITY"),
201 "nx-elabftw-url": ("settings", "NX_ELABFTW_URL"),
202 "nx-elabftw-api-key": ("settings", "NX_ELABFTW_API_KEY"),
203 "nx-elabftw-category": ("settings", "NX_ELABFTW_EXPERIMENT_CATEGORY"),
204 "nx-elabftw-status": ("settings", "NX_ELABFTW_EXPERIMENT_STATUS"),
205 "nx-labarchives-url": ("settings", "NX_LABARCHIVES_URL"),
206 "nx-labarchives-access-key-id": ("settings", "NX_LABARCHIVES_ACCESS_KEY_ID"),
207 "nx-labarchives-access-password": ("settings", "NX_LABARCHIVES_ACCESS_PASSWORD"),
208 "nx-labarchives-user-id": ("settings", "NX_LABARCHIVES_USER_ID"),
209 "nx-labarchives-notebook-id": ("settings", "NX_LABARCHIVES_NOTEBOOK_ID"),
210 "nx-email-smtp-host": ("email", "smtp_host"),
211 "nx-email-smtp-port": ("email", "smtp_port"),
212 "nx-email-smtp-username": ("email", "smtp_username"),
213 "nx-email-smtp-password": ("email", "smtp_password"),
214 "nx-email-sender": ("email", "sender"),
215 "nx-email-recipients": ("email", "recipients"),
216 "nx-cert-bundle-file": ("settings", "NX_CERT_BUNDLE_FILE"),
217}
220# --------------------------------------------------------------------------- #
221# FieldDetailScreen #
222# --------------------------------------------------------------------------- #
225class FieldDetailScreen(ModalScreen):
226 """
227 Modal popup displaying extended help text for a configuration field.
229 Invoked by pressing F1 while an Input or Select is focused in
230 ConfigScreen. Dismisses on Escape, F1, or the Close button.
232 Parameters
233 ----------
234 field_name : str
235 The environment variable / field name shown as the popup title.
236 detail_text : str
237 The extended description to display in the scrollable body.
238 """
240 CSS_PATH: ClassVar = [
241 Path(__file__).parent.parent.parent / "styles" / "config" / "screens.tcss"
242 ]
244 BINDINGS: ClassVar = [
245 ("escape", "dismiss_detail", "Close"),
246 ("f1", "dismiss_detail", "Close"),
247 ]
249 def __init__(self, field_name: str, detail_text: str, **kwargs) -> None:
250 """Initialize with the field name and detail text to display."""
251 super().__init__(**kwargs)
252 self._field_name = field_name
253 self._detail_text = detail_text
255 def compose(self) -> ComposeResult:
256 """Compose the modal layout."""
257 with Vertical(id="field-detail-dialog"):
258 yield Label(self._field_name, id="field-detail-title")
259 with VerticalScroll(id="field-detail-body"):
260 yield Static(self._detail_text, id="field-detail-text")
261 with Horizontal(id="field-detail-footer"):
262 yield Button(
263 "Close (Esc)", id="field-detail-close-btn", variant="default"
264 )
266 def action_dismiss_detail(self) -> None:
267 """Dismiss this modal."""
268 self.dismiss()
270 @on(Button.Pressed, "#field-detail-close-btn")
271 def _on_close_btn(self) -> None:
272 self.dismiss()
275# --------------------------------------------------------------------------- #
276# LabArchivesGetUidDialog #
277# --------------------------------------------------------------------------- #
280class LabArchivesGetUidDialog(ModalScreen[str | None]):
281 """Modal dialog to look up a LabArchives user ID.
283 Prompts for the user's email and LabArchives account password/app token,
284 calls the ``user_access_info`` API endpoint, and dismisses with the
285 returned UID string on success or ``None`` on cancel / error.
287 Parameters
288 ----------
289 base_url : str
290 LabArchives API base URL including the ``/api`` path
291 (e.g. ``"https://api.labarchives.com/api"``).
292 akid : str
293 Access Key ID (NX_LABARCHIVES_ACCESS_KEY_ID).
294 access_password : str
295 HMAC signing secret (NX_LABARCHIVES_ACCESS_PASSWORD).
296 """
298 CSS_PATH: ClassVar = [
299 Path(__file__).parent.parent.parent / "styles" / "config" / "screens.tcss"
300 ]
302 BINDINGS: ClassVar = [
303 ("escape", "cancel_dialog", "Cancel"),
304 ]
306 def __init__(
307 self, base_url: str, akid: str, access_password: str, **kwargs
308 ) -> None:
309 """Initialize with API credentials."""
310 super().__init__(**kwargs)
311 self._base_url = base_url
312 self._akid = akid
313 self._access_password = access_password
315 def compose(self) -> ComposeResult:
316 """Compose the dialog layout."""
317 with Vertical(id="field-detail-dialog"):
318 yield Label("Look Up LabArchives User ID", id="field-detail-title")
319 with VerticalScroll(id="field-detail-body"):
320 yield Static(
321 "Enter your LabArchives login credentials to retrieve your UID.\n\n"
322 "[bold]Email[/bold]: your LabArchives account email address.\n"
323 "[bold]LA Password[/bold]: your LabArchives external app password "
324 "(click your username in the top-right of the LabArchives web UI → "
325 "[italic]External App authentication[/italic]). "
326 "This is NOT the NX_LABARCHIVES_ACCESS_PASSWORD signing secret.",
327 id="field-detail-text",
328 )
329 yield Label("Email", classes="form-label")
330 yield Input(
331 placeholder="you@example.com",
332 id="la-uid-email",
333 )
334 yield Label("LabArchives Password / App Token", classes="form-label")
335 yield Input(
336 placeholder="your LabArchives password or app token",
337 password=True,
338 id="la-uid-password",
339 )
340 yield Static("", id="la-uid-error", classes="form-error")
341 with Horizontal(id="field-detail-footer"):
342 yield Button(
343 "Look Up",
344 id="la-uid-lookup-btn",
345 variant="primary",
346 disabled=True,
347 )
348 yield Button(
349 "Cancel (Esc)", id="field-detail-close-btn", variant="default"
350 )
352 @on(Input.Changed, "#la-uid-email")
353 @on(Input.Changed, "#la-uid-password")
354 def _update_lookup_btn(self, _event: Input.Changed) -> None:
355 email = self.query_one("#la-uid-email", Input).value.strip()
356 pw = self.query_one("#la-uid-password", Input).value.strip()
357 self.query_one("#la-uid-lookup-btn", Button).disabled = not (email and pw)
359 @on(Button.Pressed, "#la-uid-lookup-btn")
360 def _on_lookup(self) -> None:
361 from nexusLIMS.utils.labarchives import ( # noqa: PLC0415
362 LabArchivesClient,
363 LabArchivesError,
364 )
366 email = self.query_one("#la-uid-email", Input).value.strip()
367 password = self.query_one("#la-uid-password", Input).value.strip()
368 error_widget = self.query_one("#la-uid-error", Static)
370 client = LabArchivesClient(
371 base_url=self._base_url,
372 akid=self._akid,
373 password=self._access_password,
374 uid="",
375 )
376 try:
377 info = client.get_user_info(email, password)
378 except LabArchivesError as exc:
379 error_widget.update(f"API error: {exc}")
380 error_widget.add_class("visible")
381 return
383 uid = info.get("uid")
384 if not uid:
385 error_widget.update("No UID returned. Check your email and password.")
386 error_widget.add_class("visible")
387 return
389 self.dismiss(uid)
391 @on(Button.Pressed, "#field-detail-close-btn")
392 def _on_cancel_btn(self) -> None:
393 self.dismiss(None)
395 def action_cancel_dialog(self) -> None:
396 """Dismiss without a result."""
397 self.dismiss(None)
400# --------------------------------------------------------------------------- #
401# ConfigScreen #
402# --------------------------------------------------------------------------- #
405class ConfigScreen(Screen):
406 """
407 Main configuration screen with 7 tabbed sections.
409 Reads an existing ``.env`` file (if present), pre-populates all fields,
410 and writes a new ``.env`` when the user saves.
412 Parameters
413 ----------
414 env_path : pathlib.Path
415 Path to the ``.env`` file to read/write.
416 """
418 CSS_PATH: ClassVar = [
419 Path(__file__).parent.parent.parent / "styles" / "config" / "screens.tcss"
420 ]
422 BINDINGS: ClassVar = [
423 ("ctrl+s", "save", "Save"),
424 ("escape", "cancel", "Cancel"),
425 ("f1", "show_field_detail", "Field Help"),
426 ("<", "previous_tab", "Previous tab"),
427 (">", "next_tab", "Next tab"),
428 ("?", "app.help", "Help"),
429 ]
431 def __init__(self, env_path: Path, **kwargs):
432 """Initialize the config screen from an optional existing .env file."""
433 super().__init__(**kwargs)
434 self._env_path = env_path
435 self._existing: dict[str, str] = (
436 dotenv_values(env_path) if env_path.exists() else {}
437 )
438 self._nemo_harvesters: dict[int, dict] = {}
439 self._parse_nemo_harvesters()
440 # Snapshot of the env as loaded — used for unsaved-changes detection.
441 self._initial_env: dict[str, str] = dict(self._existing)
443 # ---------------------------------------------------------------------- #
444 # Helpers for reading existing env #
445 # ---------------------------------------------------------------------- #
447 def _get(self, key: str, default: str = "") -> str:
448 val = self._existing.get(key, default)
449 return val if val is not None else default
451 def _get_bool(self, key: str, *, default: bool = False) -> bool:
452 val = self._existing.get(key, "").lower()
453 if val in ("true", "1", "yes"):
454 return True
455 if val in ("false", "0", "no"):
456 return False
457 return default
459 def _parse_nemo_harvesters(self) -> None:
460 """Populate ``self._nemo_harvesters`` from the existing env vars."""
461 n = 1
462 while f"NX_NEMO_ADDRESS_{n}" in self._existing:
463 self._nemo_harvesters[n] = {
464 "address": self._existing.get(f"NX_NEMO_ADDRESS_{n}", ""),
465 "token": self._existing.get(f"NX_NEMO_TOKEN_{n}", ""),
466 "tz": self._existing.get(f"NX_NEMO_TZ_{n}"),
467 "strftime_fmt": self._existing.get(
468 f"NX_NEMO_STRFTIME_FMT_{n}", _DEFAULT_STRFTIME
469 ),
470 "strptime_fmt": self._existing.get(
471 f"NX_NEMO_STRPTIME_FMT_{n}", _DEFAULT_STRPTIME
472 ),
473 }
474 n += 1
476 def _has_elabftw(self) -> bool:
477 return bool(
478 self._existing.get("NX_ELABFTW_URL")
479 or self._existing.get("NX_ELABFTW_API_KEY")
480 )
482 def _has_email(self) -> bool:
483 return bool(
484 self._existing.get("NX_EMAIL_SMTP_HOST")
485 or self._existing.get("NX_EMAIL_SENDER")
486 )
488 def _has_labarchives(self) -> bool:
489 return bool(
490 self._existing.get("NX_LABARCHIVES_URL")
491 or self._existing.get("NX_LABARCHIVES_ACCESS_KEY_ID")
492 )
494 # ---------------------------------------------------------------------- #
495 # Compose #
496 # ---------------------------------------------------------------------- #
498 def compose(self) -> ComposeResult:
499 """Compose the tabbed config form layout."""
500 yield Header()
502 with TabbedContent():
503 with TabPane("Core Paths", id="tab-core-paths"):
504 yield from self._compose_core_paths()
505 with TabPane("CDCS", id="tab-cdcs"):
506 yield from self._compose_cdcs()
507 with TabPane("File Processing", id="tab-file-processing"):
508 yield from self._compose_file_processing()
509 with TabPane("NEMO Harvesters", id="tab-nemo"):
510 yield from self._compose_nemo()
511 with TabPane("eLabFTW", id="tab-elabftw"):
512 yield from self._compose_elabftw()
513 with TabPane("LabArchives", id="tab-labarchives"):
514 yield from self._compose_labarchives()
515 with TabPane("Email", id="tab-email"):
516 yield from self._compose_email()
517 with TabPane("SSL / Certs", id="tab-ssl"):
518 yield from self._compose_ssl()
520 with Horizontal(id="config-footer-buttons"):
521 yield Button("Save (Ctrl+S)", id="config-save-btn", variant="primary")
522 yield Button("Cancel (Esc)", id="config-cancel-btn", variant="default")
524 yield Footer()
526 # ---------------------------------------------------------------------- #
527 # Tab content composers #
528 # ---------------------------------------------------------------------- #
530 def _compose_core_paths(self) -> ComposeResult:
531 with VerticalScroll():
532 yield Label(
533 "Core file paths for NexusLIMS operation",
534 classes="tab-description",
535 )
536 with Horizontal(classes="form-columns"):
537 with Vertical(classes="form-column"):
538 yield FormField(
539 "NX_INSTRUMENT_DATA_PATH",
540 Input(
541 value=self._get("NX_INSTRUMENT_DATA_PATH"),
542 placeholder="/mnt/instrument_data",
543 id="nx-instrument-data-path",
544 ),
545 required=True,
546 help_text=_fdesc("NX_INSTRUMENT_DATA_PATH"),
547 )
548 yield FormField(
549 "NX_DATA_PATH",
550 Input(
551 value=self._get("NX_DATA_PATH"),
552 placeholder="/mnt/nexuslims_data",
553 id="nx-data-path",
554 ),
555 required=True,
556 help_text=_fdesc("NX_DATA_PATH"),
557 )
558 yield FormField(
559 "NX_DB_PATH",
560 Input(
561 value=self._get("NX_DB_PATH"),
562 placeholder="/mnt/nexuslims_data/nexuslims.db",
563 id="nx-db-path",
564 ),
565 required=True,
566 help_text=_fdesc("NX_DB_PATH"),
567 )
568 with Vertical(classes="form-column"):
569 yield FormField(
570 "NX_LOG_PATH (optional)",
571 Input(
572 value=self._get("NX_LOG_PATH"),
573 placeholder="(defaults to NX_DATA_PATH/logs/)",
574 id="nx-log-path",
575 ),
576 help_text=_fdesc("NX_LOG_PATH"),
577 )
578 yield FormField(
579 "NX_RECORDS_PATH (optional)",
580 Input(
581 value=self._get("NX_RECORDS_PATH"),
582 placeholder="(defaults to NX_DATA_PATH/records/)",
583 id="nx-records-path",
584 ),
585 help_text=_fdesc("NX_RECORDS_PATH"),
586 )
587 yield FormField(
588 "NX_LOCAL_PROFILES_PATH (optional)",
589 Input(
590 value=self._get("NX_LOCAL_PROFILES_PATH"),
591 placeholder="(leave blank if unused)",
592 id="nx-local-profiles-path",
593 ),
594 help_text=_fdesc("NX_LOCAL_PROFILES_PATH"),
595 )
597 def _compose_cdcs(self) -> ComposeResult:
598 with VerticalScroll():
599 yield Label("CDCS front-end connection settings", classes="tab-description")
600 with Horizontal(classes="form-columns"):
601 with Vertical(classes="form-column"):
602 yield FormField(
603 "NX_CDCS_URL",
604 Input(
605 value=self._get("NX_CDCS_URL"),
606 placeholder="https://cdcs.example.com",
607 id="nx-cdcs-url",
608 ),
609 required=True,
610 help_text=_fdesc("NX_CDCS_URL"),
611 )
612 with Vertical(classes="form-column"):
613 yield FormField(
614 "NX_CDCS_TOKEN",
615 Input(
616 value=self._get("NX_CDCS_TOKEN"),
617 placeholder="your-cdcs-api-token",
618 password=True,
619 id="nx-cdcs-token",
620 ),
621 required=True,
622 help_text=_fdesc("NX_CDCS_TOKEN"),
623 )
625 def _compose_file_processing(self) -> ComposeResult:
626 with VerticalScroll():
627 yield Label(
628 "Controls file discovery and record building",
629 classes="tab-description",
630 )
632 raw_patterns = self._get("NX_IGNORE_PATTERNS")
633 if raw_patterns:
634 try:
635 patterns_list = json.loads(raw_patterns)
636 patterns_display = ", ".join(patterns_list)
637 except (json.JSONDecodeError, TypeError):
638 patterns_display = raw_patterns
639 else:
640 patterns_display = "*.mib, *.db, *.emi, *.hdr"
642 with Horizontal(classes="form-columns"):
643 with Vertical(classes="form-column"):
644 strategy_opts = [
645 (
646 "exclusive \u2014 only files with known extractors",
647 "exclusive",
648 ),
649 (
650 "inclusive \u2014 all files (basic metadata for unknowns)",
651 "inclusive",
652 ),
653 ]
654 current_strategy = self._get(
655 "NX_FILE_STRATEGY", _fdefault("NX_FILE_STRATEGY")
656 )
657 yield FormField(
658 "NX_FILE_STRATEGY",
659 Select(
660 options=strategy_opts,
661 value=current_strategy,
662 id="nx-file-strategy",
663 ),
664 help_text=_fdesc("NX_FILE_STRATEGY"),
665 )
667 export_opts = [
668 (
669 "all \u2014 all destinations must succeed (recommended)",
670 "all",
671 ),
672 (
673 "first_success \u2014 stop after first success",
674 "first_success",
675 ),
676 (
677 "best_effort \u2014 try all, succeed if any succeed",
678 "best_effort",
679 ),
680 ]
681 current_export = self._get(
682 "NX_EXPORT_STRATEGY", _fdefault("NX_EXPORT_STRATEGY")
683 )
684 yield FormField(
685 "NX_EXPORT_STRATEGY",
686 Select(
687 options=export_opts,
688 value=current_export,
689 id="nx-export-strategy",
690 ),
691 help_text=_fdesc("NX_EXPORT_STRATEGY"),
692 )
694 yield FormField(
695 "NX_IGNORE_PATTERNS",
696 Input(
697 value=patterns_display,
698 placeholder=_fdefault("NX_IGNORE_PATTERNS"),
699 id="nx-ignore-patterns",
700 ),
701 help_text=_fdesc("NX_IGNORE_PATTERNS"),
702 )
704 with Vertical(classes="form-column"):
705 yield FormField(
706 "NX_FILE_DELAY_DAYS",
707 Input(
708 value=self._get(
709 "NX_FILE_DELAY_DAYS", _fdefault("NX_FILE_DELAY_DAYS")
710 ),
711 placeholder=_fdefault("NX_FILE_DELAY_DAYS"),
712 id="nx-file-delay-days",
713 ),
714 help_text=_fdesc("NX_FILE_DELAY_DAYS"),
715 )
717 yield FormField(
718 "NX_CLUSTERING_SENSITIVITY",
719 Input(
720 value=self._get(
721 "NX_CLUSTERING_SENSITIVITY",
722 _fdefault("NX_CLUSTERING_SENSITIVITY"),
723 ),
724 placeholder=_fdefault("NX_CLUSTERING_SENSITIVITY"),
725 id="nx-clustering-sensitivity",
726 ),
727 help_text=_fdesc("NX_CLUSTERING_SENSITIVITY"),
728 )
730 def _compose_nemo(self) -> ComposeResult:
731 with VerticalScroll():
732 yield Label(
733 "NEMO harvester instances — one group per NEMO server",
734 classes="tab-description",
735 )
736 yield Vertical(id="nemo-groups-container")
737 with Horizontal(classes="nemo-action-bar"):
738 yield Button(
739 "+ Add NEMO Harvester", id="nemo-add-btn", variant="primary"
740 )
742 # ---------------------------------------------------------------------- #
743 # NEMO group helpers #
744 # ---------------------------------------------------------------------- #
746 def _nemo_group_widget(self, n: int, data: dict) -> Vertical:
747 """Build and return a single NEMO harvester group widget."""
748 left_col = Vertical(
749 FormField(
750 "API Address",
751 Input(
752 value=data.get("address", ""),
753 placeholder="https://nemo.example.com/api/",
754 id=f"nemo-address-{n}",
755 ),
756 required=True,
757 help_text="Full URL to the NEMO API root (must end with '/')",
758 ),
759 FormField(
760 "API Token",
761 Input(
762 value=data.get("token", ""),
763 placeholder="your-api-token-here",
764 password=True,
765 id=f"nemo-token-{n}",
766 ),
767 required=True,
768 help_text=("Authentication token from the NEMO administration page"),
769 ),
770 FormField(
771 "Timezone (optional)",
772 AutocompleteInput(
773 suggestions=pytz.common_timezones,
774 value=data.get("tz") or "",
775 placeholder=("America/New_York (leave blank to use NEMO default)"),
776 id=f"nemo-tz-{n}",
777 ),
778 help_text="IANA timezone for coercing NEMO datetime strings",
779 ),
780 classes="form-column",
781 )
782 right_col = Vertical(
783 FormField(
784 "strftime format (optional)",
785 Input(
786 value=data.get("strftime_fmt", _DEFAULT_STRFTIME),
787 placeholder=_DEFAULT_STRFTIME,
788 id=f"nemo-strftime-{n}",
789 ),
790 help_text="Python strftime format sent to the NEMO API",
791 ),
792 FormField(
793 "strptime format (optional)",
794 Input(
795 value=data.get("strptime_fmt", _DEFAULT_STRPTIME),
796 placeholder=_DEFAULT_STRPTIME,
797 id=f"nemo-strptime-{n}",
798 ),
799 help_text=("Python strptime format for parsing NEMO API responses"),
800 ),
801 classes="form-column",
802 )
803 return Vertical(
804 Horizontal(
805 Label(f"NEMO Harvester #{n}", classes="nemo-group-title"),
806 Button(
807 "Delete",
808 id=f"nemo-delete-{n}",
809 classes="nemo-delete-btn",
810 ),
811 classes="nemo-group-header",
812 ),
813 Horizontal(left_col, right_col, classes="form-columns"),
814 id=f"nemo-group-{n}",
815 classes="nemo-group",
816 )
818 def _compose_elabftw(self) -> ComposeResult:
819 with VerticalScroll():
820 yield Label(
821 "Export experiment records to an eLabFTW instance",
822 classes="tab-description",
823 )
824 with Horizontal(classes="section-toggle-row", id="elabftw-toggle-row"):
825 yield Label(
826 "Enable eLabFTW integration",
827 classes="section-toggle-label",
828 )
829 yield Switch(
830 value=self._has_elabftw(),
831 id="elabftw-enabled",
832 )
834 enabled = self._has_elabftw()
835 with Horizontal(classes="form-columns"):
836 with Vertical(classes="form-column"):
837 yield FormField(
838 "NX_ELABFTW_URL",
839 Input(
840 value=self._get("NX_ELABFTW_URL"),
841 placeholder="https://elabftw.example.com",
842 id="nx-elabftw-url",
843 disabled=not enabled,
844 ),
845 help_text=_fdesc("NX_ELABFTW_URL"),
846 )
847 yield FormField(
848 "NX_ELABFTW_API_KEY",
849 Input(
850 value=self._get("NX_ELABFTW_API_KEY"),
851 placeholder="your-elabftw-api-key",
852 password=True,
853 id="nx-elabftw-api-key",
854 disabled=not enabled,
855 ),
856 help_text=_fdesc("NX_ELABFTW_API_KEY"),
857 )
858 with Vertical(classes="form-column"):
859 yield FormField(
860 "NX_ELABFTW_EXPERIMENT_CATEGORY (optional)",
861 Input(
862 value=self._get("NX_ELABFTW_EXPERIMENT_CATEGORY"),
863 placeholder="(integer category ID)",
864 id="nx-elabftw-category",
865 disabled=not enabled,
866 ),
867 help_text=_fdesc("NX_ELABFTW_EXPERIMENT_CATEGORY"),
868 )
869 yield FormField(
870 "NX_ELABFTW_EXPERIMENT_STATUS (optional)",
871 Input(
872 value=self._get("NX_ELABFTW_EXPERIMENT_STATUS"),
873 placeholder="(integer status ID)",
874 id="nx-elabftw-status",
875 disabled=not enabled,
876 ),
877 help_text=_fdesc("NX_ELABFTW_EXPERIMENT_STATUS"),
878 )
880 def _compose_labarchives(self) -> ComposeResult:
881 with VerticalScroll():
882 yield Label(
883 "Export experiment records to a LabArchives notebook",
884 classes="tab-description",
885 )
886 with Horizontal(classes="section-toggle-row", id="labarchives-toggle-row"):
887 yield Label(
888 "Enable LabArchives integration",
889 classes="section-toggle-label",
890 )
891 yield Switch(
892 value=self._has_labarchives(),
893 id="labarchives-enabled",
894 )
896 enabled = self._has_labarchives()
897 _la_creds_ready = bool(
898 self._get("NX_LABARCHIVES_URL")
899 and self._get("NX_LABARCHIVES_ACCESS_KEY_ID")
900 and self._get("NX_LABARCHIVES_ACCESS_PASSWORD")
901 )
902 with Horizontal(classes="form-columns"):
903 with Vertical(classes="form-column"):
904 yield FormField(
905 "NX_LABARCHIVES_URL",
906 Input(
907 value=self._get(
908 "NX_LABARCHIVES_URL",
909 _fdefault("NX_LABARCHIVES_URL"),
910 ),
911 placeholder="https://api.labarchives.com/api",
912 id="nx-labarchives-url",
913 disabled=not enabled,
914 ),
915 help_text=_fdesc("NX_LABARCHIVES_URL"),
916 )
917 yield FormField(
918 "NX_LABARCHIVES_ACCESS_KEY_ID",
919 Input(
920 value=self._get("NX_LABARCHIVES_ACCESS_KEY_ID"),
921 placeholder="your-access-key-id",
922 id="nx-labarchives-access-key-id",
923 disabled=not enabled,
924 ),
925 help_text=_fdesc("NX_LABARCHIVES_ACCESS_KEY_ID"),
926 )
927 with Vertical(classes="form-column"):
928 yield FormField(
929 "NX_LABARCHIVES_ACCESS_PASSWORD",
930 Input(
931 value=self._get("NX_LABARCHIVES_ACCESS_PASSWORD"),
932 placeholder="your-access-password",
933 password=True,
934 id="nx-labarchives-access-password",
935 disabled=not enabled,
936 ),
937 help_text=_fdesc("NX_LABARCHIVES_ACCESS_PASSWORD"),
938 )
939 with Vertical(classes="form-field"):
940 yield Label("NX_LABARCHIVES_USER_ID", classes="field-label")
941 yield Static(
942 _fdesc("NX_LABARCHIVES_USER_ID"), classes="field-help"
943 )
944 with Horizontal(classes="field-with-button"):
945 yield Input(
946 value=self._get("NX_LABARCHIVES_USER_ID"),
947 placeholder="your-uid",
948 id="nx-labarchives-user-id",
949 disabled=not enabled,
950 )
951 yield Button(
952 "Get My UID",
953 id="labarchives-get-uid-btn",
954 variant="primary",
955 disabled=not (enabled and _la_creds_ready),
956 )
957 yield Static(
958 "",
959 classes="field-error",
960 id="nx-labarchives-user-id-error",
961 )
962 yield FormField(
963 "NX_LABARCHIVES_NOTEBOOK_ID (optional)",
964 Input(
965 value=self._get("NX_LABARCHIVES_NOTEBOOK_ID"),
966 placeholder="(leave blank to use Inbox)",
967 id="nx-labarchives-notebook-id",
968 disabled=not enabled,
969 ),
970 help_text=_fdesc("NX_LABARCHIVES_NOTEBOOK_ID"),
971 )
973 def _compose_email(self) -> ComposeResult:
974 with VerticalScroll():
975 yield Label(
976 "Send notifications on record builder errors",
977 classes="tab-description",
978 )
979 with Horizontal(classes="section-toggle-row", id="email-toggle-row"):
980 yield Label(
981 "Enable email notifications",
982 classes="section-toggle-label",
983 )
984 yield Switch(
985 value=self._has_email(),
986 id="email-enabled",
987 )
989 enabled = self._has_email()
990 with Horizontal(classes="form-columns"):
991 with Vertical(classes="form-column"):
992 yield FormField(
993 "SMTP Host",
994 Input(
995 value=self._get("NX_EMAIL_SMTP_HOST"),
996 placeholder="smtp.example.com",
997 id="nx-email-smtp-host",
998 disabled=not enabled,
999 ),
1000 help_text=_edesc("smtp_host"),
1001 )
1002 yield FormField(
1003 "SMTP Port",
1004 Input(
1005 value=self._get(
1006 "NX_EMAIL_SMTP_PORT", _edefault("smtp_port")
1007 ),
1008 placeholder=_edefault("smtp_port"),
1009 id="nx-email-smtp-port",
1010 disabled=not enabled,
1011 ),
1012 help_text=_edesc("smtp_port"),
1013 )
1014 yield FormField(
1015 "SMTP Username (optional)",
1016 Input(
1017 value=self._get("NX_EMAIL_SMTP_USERNAME"),
1018 placeholder="(leave blank if not required)",
1019 id="nx-email-smtp-username",
1020 disabled=not enabled,
1021 ),
1022 help_text=_edesc("smtp_username"),
1023 )
1024 yield FormField(
1025 "SMTP Password (optional)",
1026 Input(
1027 value=self._get("NX_EMAIL_SMTP_PASSWORD"),
1028 placeholder="(leave blank if not required)",
1029 password=True,
1030 id="nx-email-smtp-password",
1031 disabled=not enabled,
1032 ),
1033 help_text=_edesc("smtp_password"),
1034 )
1035 with Vertical(classes="form-column"):
1036 with Horizontal(classes="section-toggle-row"):
1037 yield Label("Use TLS", classes="section-toggle-label")
1038 yield Switch(
1039 value=self._get_bool("NX_EMAIL_USE_TLS", default=True),
1040 id="nx-email-use-tls",
1041 disabled=not enabled,
1042 )
1043 yield FormField(
1044 "Sender Address",
1045 Input(
1046 value=self._get("NX_EMAIL_SENDER"),
1047 placeholder="nexuslims@example.com",
1048 id="nx-email-sender",
1049 disabled=not enabled,
1050 ),
1051 help_text=_edesc("sender"),
1052 )
1053 yield FormField(
1054 "Recipients",
1055 Input(
1056 value=self._get("NX_EMAIL_RECIPIENTS"),
1057 placeholder="admin@example.com, user2@example.com",
1058 id="nx-email-recipients",
1059 disabled=not enabled,
1060 ),
1061 help_text=_edesc("recipients"),
1062 )
1064 def _compose_ssl(self) -> ComposeResult:
1065 with VerticalScroll():
1066 yield Label("SSL / Certificate configuration", classes="tab-description")
1067 yield FormField(
1068 "NX_CERT_BUNDLE_FILE (optional)",
1069 Input(
1070 value=self._get("NX_CERT_BUNDLE_FILE"),
1071 placeholder="/path/to/ca-bundle.crt",
1072 id="nx-cert-bundle-file",
1073 ),
1074 help_text=_fdesc("NX_CERT_BUNDLE_FILE"),
1075 )
1076 yield Label("NX_CERT_BUNDLE (optional)", classes="field-label")
1077 yield Static(
1078 _fdesc("NX_CERT_BUNDLE"),
1079 classes="field-help",
1080 )
1081 yield TextArea(
1082 text=self._get("NX_CERT_BUNDLE"),
1083 id="nx-cert-bundle",
1084 )
1086 disable_ssl = self._get_bool("NX_DISABLE_SSL_VERIFY", default=False)
1087 with Horizontal(classes="section-toggle-row ssl-verify-row"):
1088 yield Label(
1089 "NX_DISABLE_SSL_VERIFY",
1090 classes="section-toggle-label",
1091 )
1092 yield Switch(
1093 value=disable_ssl,
1094 id="nx-disable-ssl-verify",
1095 )
1096 yield Static(
1097 "WARNING: Disabling SSL verification is insecure. "
1098 "Only use for local development with self-signed certificates.",
1099 id="ssl-verify-warning",
1100 classes="ssl-warning" + (" visible" if disable_ssl else ""),
1101 )
1103 # ---------------------------------------------------------------------- #
1104 # Lifecycle hooks #
1105 # ---------------------------------------------------------------------- #
1107 def on_mount(self) -> None:
1108 """Populate NEMO harvester groups and configure toggle rows after mount."""
1109 container = self.query_one("#nemo-groups-container", Vertical)
1110 for n, data in sorted(self._nemo_harvesters.items()):
1111 container.mount(self._nemo_group_widget(n, data))
1112 self.query_one("#elabftw-toggle-row").set_class(self._has_elabftw(), "-on")
1113 self.query_one("#labarchives-toggle-row").set_class(
1114 self._has_labarchives(), "-on"
1115 )
1116 self.query_one("#email-toggle-row").set_class(self._has_email(), "-on")
1118 def _next_nemo_index(self) -> int:
1119 """Return the next available NEMO harvester index."""
1120 existing = [
1121 int(w.id.split("-")[-1])
1122 for w in self.query(".nemo-group")
1123 if w.id and w.id.startswith("nemo-group-")
1124 ]
1125 return max(existing, default=0) + 1
1127 # ---------------------------------------------------------------------- #
1128 # Event handlers #
1129 # ---------------------------------------------------------------------- #
1131 @on(Button.Pressed, "#config-save-btn")
1132 def _on_save_btn(self) -> None:
1133 self.action_save()
1135 @on(Button.Pressed, "#config-cancel-btn")
1136 def _on_cancel_btn(self) -> None:
1137 self.action_cancel()
1139 @on(Button.Pressed, "#nemo-add-btn")
1140 def _on_nemo_add(self) -> None:
1141 n = self._next_nemo_index()
1142 container = self.query_one("#nemo-groups-container", Vertical)
1143 container.mount(self._nemo_group_widget(n, {}))
1144 self.app.notify(f"Added NEMO Harvester #{n}", timeout=2)
1146 def on_button_pressed(self, event: Button.Pressed) -> None:
1147 """Handle delete buttons on individual NEMO harvester groups."""
1148 if event.button.has_class("nemo-delete-btn"):
1149 group = event.button.parent.parent # Button → header → group
1150 if group is not None:
1151 group.remove()
1152 event.stop()
1154 @on(Switch.Changed, "#elabftw-enabled")
1155 def _on_elabftw_toggle(self, event: Switch.Changed) -> None:
1156 enabled = event.value
1157 self.query_one("#elabftw-toggle-row").set_class(enabled, "-on")
1158 for field_id in (
1159 "nx-elabftw-url",
1160 "nx-elabftw-api-key",
1161 "nx-elabftw-category",
1162 "nx-elabftw-status",
1163 ):
1164 with contextlib.suppress(Exception):
1165 self.query_one(f"#{field_id}", Input).disabled = not enabled
1167 @on(Switch.Changed, "#labarchives-enabled")
1168 def _on_labarchives_toggle(self, event: Switch.Changed) -> None:
1169 enabled = event.value
1170 self.query_one("#labarchives-toggle-row").set_class(enabled, "-on")
1171 for field_id in (
1172 "nx-labarchives-url",
1173 "nx-labarchives-access-key-id",
1174 "nx-labarchives-access-password",
1175 "nx-labarchives-user-id",
1176 "nx-labarchives-notebook-id",
1177 ):
1178 with contextlib.suppress(Exception):
1179 self.query_one(f"#{field_id}", Input).disabled = not enabled
1180 self._update_la_get_uid_btn()
1182 @on(Input.Changed, "#nx-labarchives-url")
1183 @on(Input.Changed, "#nx-labarchives-access-key-id")
1184 @on(Input.Changed, "#nx-labarchives-access-password")
1185 def _on_labarchives_cred_changed(self, _event: Input.Changed) -> None:
1186 self._update_la_get_uid_btn()
1188 def _update_la_get_uid_btn(self) -> None:
1189 """Enable the 'Get My UID' button only when all three credentials are set."""
1190 with contextlib.suppress(Exception):
1191 enabled = self.query_one("#labarchives-enabled", Switch).value
1192 url = self.query_one("#nx-labarchives-url", Input).value.strip()
1193 akid = self.query_one("#nx-labarchives-access-key-id", Input).value.strip()
1194 pw = self.query_one("#nx-labarchives-access-password", Input).value.strip()
1195 ready = enabled and bool(url and akid and pw)
1196 self.query_one("#labarchives-get-uid-btn", Button).disabled = not ready
1198 @on(Button.Pressed, "#labarchives-get-uid-btn")
1199 def _on_labarchives_get_uid(self) -> None:
1200 """Open the UID lookup dialog and populate the User ID field on success."""
1201 with contextlib.suppress(Exception):
1202 url = self.query_one("#nx-labarchives-url", Input).value.strip()
1203 akid = self.query_one("#nx-labarchives-access-key-id", Input).value.strip()
1204 pw = self.query_one("#nx-labarchives-access-password", Input).value.strip()
1205 self.app.push_screen(
1206 LabArchivesGetUidDialog(base_url=url, akid=akid, access_password=pw),
1207 self._on_uid_lookup_result,
1208 )
1210 def _on_uid_lookup_result(self, uid: str | None) -> None:
1211 """Populate the User ID input with the returned UID."""
1212 if uid:
1213 with contextlib.suppress(Exception):
1214 self.query_one("#nx-labarchives-user-id", Input).value = uid
1215 self.app.notify(
1216 f"LabArchives UID retrieved: {uid}", severity="information", timeout=5
1217 )
1219 @on(Switch.Changed, "#email-enabled")
1220 def _on_email_toggle(self, event: Switch.Changed) -> None:
1221 enabled = event.value
1222 self.query_one("#email-toggle-row").set_class(enabled, "-on")
1223 for field_id in (
1224 "nx-email-smtp-host",
1225 "nx-email-smtp-port",
1226 "nx-email-smtp-username",
1227 "nx-email-smtp-password",
1228 "nx-email-use-tls",
1229 "nx-email-sender",
1230 "nx-email-recipients",
1231 ):
1232 with contextlib.suppress(Exception):
1233 self.query_one(f"#{field_id}").disabled = not enabled
1235 @on(Switch.Changed, "#nx-disable-ssl-verify")
1236 def _on_ssl_verify_toggle(self, event: Switch.Changed) -> None:
1237 warning = self.query_one("#ssl-verify-warning", Static)
1238 if event.value:
1239 warning.add_class("visible")
1240 else:
1241 warning.remove_class("visible")
1243 # ---------------------------------------------------------------------- #
1244 # Actions #
1245 # ---------------------------------------------------------------------- #
1247 def action_save(self) -> None:
1248 """Validate all fields and write the .env file."""
1249 errors = self._validate_all()
1250 if errors:
1251 msg = f"Cannot save: {len(errors)} error(s). " + "; ".join(errors[:2])
1252 if len(errors) > 2:
1253 msg += f" (and {len(errors) - 2} more)"
1254 self.app.notify(msg, severity="error", timeout=6)
1255 return
1257 try:
1258 config_dict = self._build_config_dict()
1259 env_vars = _flatten_to_env(config_dict)
1260 _write_env_file(env_vars, self._env_path)
1261 self.app.notify(
1262 f"Configuration saved to {self._env_path}",
1263 severity="information",
1264 timeout=4,
1265 )
1266 self.app.exit()
1267 except Exception as exc:
1268 self.app.notify(f"Failed to save: {exc}", severity="error", timeout=6)
1270 def _has_changes(self) -> bool:
1271 """Return True if the current form state differs from the loaded env."""
1272 try:
1273 current_env = _flatten_to_env(self._build_config_dict())
1274 except Exception:
1275 return True
1276 return current_env != self._initial_env
1278 def action_cancel(self) -> None:
1279 """Exit without saving, prompting if there are unsaved changes."""
1280 if not self._has_changes():
1281 self.app.exit()
1282 return
1283 self.app.push_screen(
1284 ConfirmDialog(
1285 "You have unsaved changes. Exit without saving?",
1286 title="Unsaved Changes",
1287 ),
1288 self._on_cancel_confirmed,
1289 )
1291 def _on_cancel_confirmed(self, confirmed: bool) -> None:
1292 if confirmed:
1293 self.app.exit()
1295 def action_next_tab(self) -> None:
1296 """Activate the next tab."""
1297 self.query_one(TabbedContent).query_one(Tabs).action_next_tab()
1299 def action_previous_tab(self) -> None:
1300 """Activate the previous tab."""
1301 self.query_one(TabbedContent).query_one(Tabs).action_previous_tab()
1303 def _resolve_focused_field_detail(self, focused) -> tuple[str | None, str]:
1304 """Return ``(field_name, detail)`` for the currently focused widget."""
1305 if isinstance(focused, Input):
1306 return self._resolve_input_field_detail(focused)
1307 if isinstance(focused, Switch):
1308 if focused.id == "nx-disable-ssl-verify":
1309 name = "NX_DISABLE_SSL_VERIFY"
1310 return name, _fdetail(name)
1311 elif isinstance(focused, TextArea):
1312 if focused.id == "nx-cert-bundle":
1313 name = "NX_CERT_BUNDLE"
1314 return name, _fdetail(name)
1315 elif isinstance(focused, Select):
1316 return self._resolve_select_field_detail(focused)
1317 return None, ""
1319 def _resolve_input_field_detail(self, focused: Input) -> tuple[str | None, str]:
1320 """Return ``(field_name, detail)`` for a focused Input widget."""
1321 input_id = focused.id or ""
1322 mapping = _INPUT_ID_TO_FIELD.get(input_id)
1323 if mapping:
1324 model_class, field_name = mapping
1325 detail = (
1326 _fdetail(field_name)
1327 if model_class == "settings"
1328 else _edetail(field_name)
1329 )
1330 return field_name, detail
1331 for prefix, nemo_field in _NEMO_INPUT_PREFIX_TO_FIELD.items():
1332 if input_id.startswith(prefix):
1333 field_name = f"NX_NEMO_{nemo_field.upper()}_N"
1334 return field_name, _ndetail(nemo_field)
1335 return None, ""
1337 def _resolve_select_field_detail(self, focused: Select) -> tuple[str | None, str]:
1338 """Return ``(field_name, detail)`` for a focused Select widget."""
1339 select_id_map = {
1340 "nx-file-strategy": "NX_FILE_STRATEGY",
1341 "nx-export-strategy": "NX_EXPORT_STRATEGY",
1342 }
1343 name = select_id_map.get(focused.id or "")
1344 if name:
1345 return name, _fdetail(name)
1346 return None, ""
1348 def action_show_field_detail(self) -> None:
1349 """Show extended help popup for the currently focused input or select."""
1350 field_name, detail = self._resolve_focused_field_detail(self.screen.focused)
1352 if not field_name or not detail:
1353 if field_name:
1354 self.app.notify(
1355 f"No extended help available for {field_name}.",
1356 severity="information",
1357 timeout=2,
1358 )
1359 return
1361 self.app.push_screen(FieldDetailScreen(field_name, detail))
1363 # ---------------------------------------------------------------------- #
1364 # Validation helpers #
1365 # ---------------------------------------------------------------------- #
1367 def _validate_core_paths(self) -> list[str]:
1368 errors: list[str] = []
1369 for field_id, label in [
1370 ("nx-instrument-data-path", "NX_INSTRUMENT_DATA_PATH"),
1371 ("nx-data-path", "NX_DATA_PATH"),
1372 ("nx-db-path", "NX_DB_PATH"),
1373 ]:
1374 val = self.query_one(f"#{field_id}", Input).value.strip()
1375 ok, msg = validate_required(val, label)
1376 if not ok:
1377 errors.append(msg)
1378 return errors
1380 def _validate_cdcs(self) -> list[str]:
1381 errors: list[str] = []
1382 cdcs_url = self.query_one("#nx-cdcs-url", Input).value.strip()
1383 ok, msg = validate_url(cdcs_url, "NX_CDCS_URL")
1384 if not ok:
1385 errors.append(msg)
1386 cdcs_token = self.query_one("#nx-cdcs-token", Input).value.strip()
1387 ok, msg = validate_required(cdcs_token, "NX_CDCS_TOKEN")
1388 if not ok:
1389 errors.append(msg)
1390 return errors
1392 def _validate_file_processing(self) -> list[str]:
1393 errors: list[str] = []
1394 ok, msg = validate_float_positive(
1395 self.query_one("#nx-file-delay-days", Input).value.strip(),
1396 "NX_FILE_DELAY_DAYS",
1397 )
1398 if not ok:
1399 errors.append(msg)
1400 ok, msg = validate_float_nonneg(
1401 self.query_one("#nx-clustering-sensitivity", Input).value.strip(),
1402 "NX_CLUSTERING_SENSITIVITY",
1403 )
1404 if not ok:
1405 errors.append(msg)
1406 return errors
1408 def _validate_elabftw(self) -> list[str]:
1409 if not self.query_one("#elabftw-enabled", Switch).value:
1410 return []
1411 errors: list[str] = []
1412 url = self.query_one("#nx-elabftw-url", Input).value.strip()
1413 ok, msg = validate_optional_url(url, "NX_ELABFTW_URL")
1414 if not ok:
1415 errors.append(msg)
1416 ok, msg = validate_required(
1417 self.query_one("#nx-elabftw-api-key", Input).value.strip(),
1418 "NX_ELABFTW_API_KEY",
1419 )
1420 if not ok:
1421 errors.append(msg)
1422 for field_id, label in [
1423 ("nx-elabftw-category", "NX_ELABFTW_EXPERIMENT_CATEGORY"),
1424 ("nx-elabftw-status", "NX_ELABFTW_EXPERIMENT_STATUS"),
1425 ]:
1426 ok, msg = validate_optional_int(
1427 self.query_one(f"#{field_id}", Input).value.strip(), label
1428 )
1429 if not ok:
1430 errors.append(msg)
1431 return errors
1433 def _validate_labarchives(self) -> list[str]:
1434 if not self.query_one("#labarchives-enabled", Switch).value:
1435 return []
1436 errors: list[str] = []
1437 url = self.query_one("#nx-labarchives-url", Input).value.strip()
1438 ok, msg = validate_optional_url(url, "NX_LABARCHIVES_URL")
1439 if not ok:
1440 errors.append(msg)
1441 for field_id, label in [
1442 ("nx-labarchives-access-key-id", "NX_LABARCHIVES_ACCESS_KEY_ID"),
1443 ("nx-labarchives-access-password", "NX_LABARCHIVES_ACCESS_PASSWORD"),
1444 ("nx-labarchives-user-id", "NX_LABARCHIVES_USER_ID"),
1445 ]:
1446 ok, msg = validate_required(
1447 self.query_one(f"#{field_id}", Input).value.strip(), label
1448 )
1449 if not ok:
1450 errors.append(msg)
1451 return errors
1453 def _validate_email(self) -> list[str]:
1454 if not self.query_one("#email-enabled", Switch).value:
1455 return []
1456 errors: list[str] = []
1457 for field_id, label in [
1458 ("nx-email-smtp-host", "SMTP Host"),
1459 ("nx-email-sender", "Email Sender"),
1460 ("nx-email-recipients", "Email Recipients"),
1461 ]:
1462 ok, msg = validate_required(
1463 self.query_one(f"#{field_id}", Input).value.strip(), label
1464 )
1465 if not ok:
1466 errors.append(msg)
1467 ok, msg = validate_smtp_port(
1468 self.query_one("#nx-email-smtp-port", Input).value.strip()
1469 )
1470 if not ok:
1471 errors.append(msg)
1472 return errors
1474 def _validate_nemo(self) -> list[str]:
1475 errors: list[str] = []
1476 for group in self.query(".nemo-group"):
1477 if group.id is None or not group.id.startswith("nemo-group-"):
1478 continue
1479 n = group.id.split("-")[-1]
1480 address = self.query_one(f"#nemo-address-{n}", Input).value.strip()
1481 ok, msg = validate_nemo_address(address)
1482 if not ok:
1483 errors.append(f"Harvester #{n} API Address: {msg}")
1484 token = self.query_one(f"#nemo-token-{n}", Input).value.strip()
1485 ok, msg = validate_required(token, f"Harvester #{n} API Token")
1486 if not ok:
1487 errors.append(msg)
1488 tz_raw = self.query_one(f"#nemo-tz-{n}", Input).value.strip()
1489 if tz_raw:
1490 ok, msg = validate_optional_iana_timezone(tz_raw)
1491 if not ok:
1492 errors.append(f"Harvester #{n} Timezone: {msg}")
1493 return errors
1495 def _validate_all(self) -> list[str]:
1496 return (
1497 self._validate_core_paths()
1498 + self._validate_cdcs()
1499 + self._validate_file_processing()
1500 + self._validate_nemo()
1501 + self._validate_elabftw()
1502 + self._validate_labarchives()
1503 + self._validate_email()
1504 )
1506 # ---------------------------------------------------------------------- #
1507 # Config dict builder helpers #
1508 # ---------------------------------------------------------------------- #
1510 def _build_paths_config(self) -> dict:
1511 config: dict = {}
1512 for field_id, key in [
1513 ("nx-instrument-data-path", "NX_INSTRUMENT_DATA_PATH"),
1514 ("nx-data-path", "NX_DATA_PATH"),
1515 ("nx-db-path", "NX_DB_PATH"),
1516 ("nx-log-path", "NX_LOG_PATH"),
1517 ("nx-records-path", "NX_RECORDS_PATH"),
1518 ("nx-local-profiles-path", "NX_LOCAL_PROFILES_PATH"),
1519 ]:
1520 val = self.query_one(f"#{field_id}", Input).value.strip()
1521 if val:
1522 config[key] = val
1523 return config
1525 def _build_cdcs_config(self) -> dict:
1526 config: dict = {}
1527 for field_id, key in [
1528 ("nx-cdcs-url", "NX_CDCS_URL"),
1529 ("nx-cdcs-token", "NX_CDCS_TOKEN"),
1530 ]:
1531 val = self.query_one(f"#{field_id}", Input).value.strip()
1532 if val:
1533 config[key] = val
1534 return config
1536 def _build_file_config(self) -> dict:
1537 config: dict = {}
1538 strategy_val = self.query_one("#nx-file-strategy", Select).value
1539 if strategy_val and strategy_val is not Select.BLANK:
1540 config["NX_FILE_STRATEGY"] = strategy_val
1541 export_val = self.query_one("#nx-export-strategy", Select).value
1542 if export_val and export_val is not Select.BLANK:
1543 config["NX_EXPORT_STRATEGY"] = export_val
1544 delay = self.query_one("#nx-file-delay-days", Input).value.strip()
1545 if delay:
1546 config["NX_FILE_DELAY_DAYS"] = float(delay)
1547 sensitivity = self.query_one("#nx-clustering-sensitivity", Input).value.strip()
1548 if sensitivity:
1549 config["NX_CLUSTERING_SENSITIVITY"] = float(sensitivity)
1550 patterns_raw = self.query_one("#nx-ignore-patterns", Input).value.strip()
1551 if patterns_raw:
1552 patterns_list = [p.strip() for p in patterns_raw.split(",") if p.strip()]
1553 config["NX_IGNORE_PATTERNS"] = patterns_list
1554 return config
1556 def _build_elabftw_config(self) -> dict:
1557 if not self.query_one("#elabftw-enabled", Switch).value:
1558 return {}
1559 config: dict = {}
1560 for field_id, key in [
1561 ("nx-elabftw-url", "NX_ELABFTW_URL"),
1562 ("nx-elabftw-api-key", "NX_ELABFTW_API_KEY"),
1563 ]:
1564 val = self.query_one(f"#{field_id}", Input).value.strip()
1565 if val:
1566 config[key] = val
1567 for field_id, key in [
1568 ("nx-elabftw-category", "NX_ELABFTW_EXPERIMENT_CATEGORY"),
1569 ("nx-elabftw-status", "NX_ELABFTW_EXPERIMENT_STATUS"),
1570 ]:
1571 val = self.query_one(f"#{field_id}", Input).value.strip()
1572 if val:
1573 config[key] = int(val)
1574 return config
1576 def _build_labarchives_config(self) -> dict:
1577 if not self.query_one("#labarchives-enabled", Switch).value:
1578 return {}
1579 config: dict = {}
1580 for field_id, key in [
1581 ("nx-labarchives-url", "NX_LABARCHIVES_URL"),
1582 ("nx-labarchives-access-key-id", "NX_LABARCHIVES_ACCESS_KEY_ID"),
1583 ("nx-labarchives-access-password", "NX_LABARCHIVES_ACCESS_PASSWORD"),
1584 ("nx-labarchives-user-id", "NX_LABARCHIVES_USER_ID"),
1585 ("nx-labarchives-notebook-id", "NX_LABARCHIVES_NOTEBOOK_ID"),
1586 ]:
1587 val = self.query_one(f"#{field_id}", Input).value.strip()
1588 if val:
1589 config[key] = val
1590 return config
1592 def _build_email_config(self) -> dict:
1593 if not self.query_one("#email-enabled", Switch).value:
1594 return {}
1595 smtp_host = self.query_one("#nx-email-smtp-host", Input).value.strip()
1596 smtp_port_str = self.query_one("#nx-email-smtp-port", Input).value.strip()
1597 smtp_user = self.query_one("#nx-email-smtp-username", Input).value.strip()
1598 smtp_pass = self.query_one("#nx-email-smtp-password", Input).value.strip()
1599 use_tls = self.query_one("#nx-email-use-tls", Switch).value
1600 sender = self.query_one("#nx-email-sender", Input).value.strip()
1601 recipients_raw = self.query_one("#nx-email-recipients", Input).value.strip()
1602 recipients = [r.strip() for r in recipients_raw.split(",") if r.strip()]
1604 email_inner: dict = {
1605 "smtp_host": smtp_host,
1606 "smtp_port": int(smtp_port_str) if smtp_port_str else 587,
1607 "use_tls": use_tls,
1608 "sender": sender,
1609 "recipients": recipients,
1610 }
1611 if smtp_user:
1612 email_inner["smtp_username"] = smtp_user
1613 if smtp_pass:
1614 email_inner["smtp_password"] = smtp_pass
1615 return {"email_config": email_inner}
1617 def _build_ssl_config(self) -> dict:
1618 config: dict = {}
1619 cert_file = self.query_one("#nx-cert-bundle-file", Input).value.strip()
1620 if cert_file:
1621 config["NX_CERT_BUNDLE_FILE"] = cert_file
1622 cert_bundle = self.query_one("#nx-cert-bundle", TextArea).text.strip()
1623 if cert_bundle:
1624 config["NX_CERT_BUNDLE"] = cert_bundle
1625 config["NX_DISABLE_SSL_VERIFY"] = self.query_one(
1626 "#nx-disable-ssl-verify", Switch
1627 ).value
1628 return config
1630 def _build_nemo_config(self) -> dict:
1631 """Build NEMO harvesters config from inline form groups."""
1632 harvesters: dict = {}
1633 for i, group in enumerate(self.query(".nemo-group"), start=1):
1634 if group.id is None or not group.id.startswith("nemo-group-"):
1635 continue
1636 n = group.id.split("-")[-1]
1637 address = self.query_one(f"#nemo-address-{n}", Input).value.strip()
1638 token = self.query_one(f"#nemo-token-{n}", Input).value.strip()
1639 tz_raw = self.query_one(f"#nemo-tz-{n}", Input).value.strip()
1640 strftime = self.query_one(f"#nemo-strftime-{n}", Input).value.strip()
1641 strptime = self.query_one(f"#nemo-strptime-{n}", Input).value.strip()
1642 hvst: dict = {
1643 "address": address,
1644 "token": token,
1645 "strftime_fmt": strftime or _DEFAULT_STRFTIME,
1646 "strptime_fmt": strptime or _DEFAULT_STRPTIME,
1647 }
1648 if tz_raw:
1649 hvst["tz"] = tz_raw
1650 harvesters[str(i)] = hvst
1651 return {"nemo_harvesters": harvesters} if harvesters else {}
1653 def _build_config_dict(self) -> dict:
1654 """Build the nested config dict consumed by ``_flatten_to_env``."""
1655 config: dict = {}
1656 config.update(self._build_paths_config())
1657 config.update(self._build_cdcs_config())
1658 config.update(self._build_file_config())
1659 config.update(self._build_nemo_config())
1660 config.update(self._build_elabftw_config())
1661 config.update(self._build_labarchives_config())
1662 config.update(self._build_email_config())
1663 config.update(self._build_ssl_config())
1664 return config