Coverage for nexusLIMS/cli/manage_instruments.py: 100%
101 statements
« prev ^ index » next coverage.py v7.11.3, created at 2026-03-24 05:23 +0000
1"""
2CLI entry point for the NexusLIMS instrument management TUI.
4Provides the ``nexuslims instruments manage`` and ``nexuslims instruments list``
5commands for interactive and scriptable database management.
7Usage
8-----
10.. code-block:: bash
12 # Launch the instrument management TUI
13 nexuslims instruments manage
15 # List instruments in a table
16 nexuslims instruments list
18 # List instruments as JSON
19 nexuslims instruments list --format json
21 # Show version information
22 nexuslims instruments manage --version
24Features
25--------
26- List all instruments in a searchable table
27- Add new instruments with form validation
28- Edit existing instruments
29- Delete instruments with confirmation prompts
30- Theme switching (dark/light mode)
31- Built-in help screen
32- Automatic database initialization (if NX_DB_PATH doesn't exist yet)
33"""
35import json
36import logging
37import os
38from pathlib import Path
40import click
42# Heavy imports are lazy-loaded inside the main function to keep
43# --help / --version fast (same pattern as other CLI tools).
45logger = logging.getLogger(__name__)
48def _ensure_database_initialized() -> None:
49 """
50 Ensure the NexusLIMS database exists and is initialized.
52 If NX_DB_PATH is not set or the database file doesn't exist,
53 automatically initializes it using the migration system.
54 """
55 # Load .env file if it exists (before checking environment variables)
56 from dotenv import find_dotenv, load_dotenv # noqa: PLC0415
58 load_dotenv(find_dotenv(usecwd=True))
60 # Get DB path from environment variable
61 db_path_str = os.getenv("NX_DB_PATH")
62 if not db_path_str:
63 click.secho(
64 "Error: NX_DB_PATH environment variable is not set.",
65 fg="red",
66 err=True,
67 )
68 click.echo("\nSet it via the interactive configurator:\n", err=True)
69 click.echo(" nexuslims config edit\n", err=True)
70 click.echo("Or set it directly, e.g.:\n", err=True)
71 click.echo(" export NX_DB_PATH=/path/to/database.db", err=True)
72 raise click.Abort
74 db_path = Path(db_path_str)
76 # If database doesn't exist, initialize it
77 if not db_path.exists():
78 click.echo(f"Database not found at {db_path}")
79 click.echo("Initializing new database...")
81 # Create parent directory if it doesn't exist
82 db_path.parent.mkdir(parents=True, exist_ok=True)
84 # Create empty database file
85 db_path.touch()
87 # Import migration utilities
88 from nexusLIMS.cli.migrate import ( # noqa: PLC0415
89 _get_current_revision,
90 _run_alembic_command,
91 )
93 # Run all migrations to create the schema
94 try:
95 _run_alembic_command("upgrade", "head")
96 click.secho("✓ Database initialized successfully", fg="green")
97 click.echo(f" Current version: {_get_current_revision()}\n")
98 except Exception as e:
99 click.secho(f"Error initializing database: {e}", fg="red", err=True)
100 # Clean up the empty database file on failure
101 if db_path.exists():
102 db_path.unlink()
103 raise click.Abort from e
106def _truncate_url_middle(url: str, max_width: int = 22) -> str:
107 """Truncate a URL in the middle to fit within max_width characters.
109 Preserves the beginning (scheme + host) and the end (query string / path
110 tail) so that both the server and the tool ID remain readable.
112 Parameters
113 ----------
114 url
115 The URL string to truncate.
116 max_width
117 Maximum character width of the result.
119 Returns
120 -------
121 str
122 The original URL if it fits, otherwise a middle-truncated version
123 with ``…`` in the middle.
124 """
125 if len(url) <= max_width:
126 return url
127 # Reserve 1 char for the ellipsis
128 half = (max_width - 1) // 2
129 return url[:half] + "…" + url[-(max_width - half - 1) :]
132def _list_instruments(output_format: str = "table") -> None:
133 """Print a summary of all instruments in the database.
135 Parameters
136 ----------
137 output_format
138 Either ``"table"`` (default, Rich formatted) or ``"json"``.
139 """
140 from sqlalchemy import func # noqa: PLC0415
141 from sqlmodel import Session as DBSession # noqa: PLC0415
142 from sqlmodel import select # noqa: PLC0415
144 from nexusLIMS.db.engine import get_engine # noqa: PLC0415
145 from nexusLIMS.db.enums import EventType # noqa: PLC0415
146 from nexusLIMS.db.models import Instrument, SessionLog # noqa: PLC0415
148 with DBSession(get_engine()) as session:
149 instruments = session.exec(select(Instrument)).all()
151 if not instruments:
152 click.echo(
153 "No instruments found. Use 'nexuslims instruments manage' to add "
154 "instruments."
155 )
156 return
158 # Build per-instrument stats: (total_sessions, last_session_dt)
159 stats: dict[str, tuple[int, object]] = {}
160 for inst in instruments:
161 row = session.exec(
162 select(
163 func.count(func.distinct(SessionLog.session_identifier)),
164 func.max(SessionLog.timestamp),
165 ).where(
166 SessionLog.instrument == inst.instrument_pid,
167 SessionLog.event_type == EventType.END,
168 )
169 ).one()
170 total_sessions, last_ts = row
171 stats[inst.instrument_pid] = (total_sessions or 0, last_ts)
173 if output_format == "json":
174 records = []
175 for inst in instruments:
176 total_sessions, last_ts = stats[inst.instrument_pid]
177 last_session_str: str | None = None
178 if last_ts is not None:
179 localized = inst.localize_datetime(last_ts)
180 last_session_str = localized.isoformat()
181 records.append(
182 {
183 "instrument_pid": inst.instrument_pid,
184 "display_name": inst.display_name,
185 "location": inst.location,
186 "api_url": inst.api_url,
187 "harvester": inst.harvester,
188 "sessions_total": total_sessions,
189 "last_session": last_session_str,
190 }
191 )
192 click.echo(json.dumps(records, indent=2))
193 return
195 # Rich table output
196 from rich.console import Console # noqa: PLC0415
197 from rich.table import Table # noqa: PLC0415
199 n = len(instruments)
200 table = Table(
201 title=f"NexusLIMS Instruments ({n} total)",
202 show_header=True,
203 header_style="bold",
204 )
205 table.add_column("ID", no_wrap=True)
206 table.add_column("Display Name")
207 table.add_column("Location")
208 table.add_column("API URL", no_wrap=True)
209 table.add_column("Sessions", justify="right")
210 table.add_column("Last Session", no_wrap=True)
212 for inst in instruments:
213 total_sessions, last_ts = stats[inst.instrument_pid]
214 if last_ts is not None:
215 localized = inst.localize_datetime(last_ts)
216 last_session_str = localized.strftime("%Y-%m-%d %H:%M %Z")
217 else:
218 last_session_str = "—"
219 table.add_row(
220 inst.instrument_pid,
221 inst.display_name,
222 inst.location,
223 _truncate_url_middle(inst.api_url),
224 str(total_sessions),
225 last_session_str,
226 )
228 console = Console()
229 console.print(table)
232def _run_instrument_manager() -> None:
233 """Launch the instrument management TUI.
235 This is separated from the Click command so it can be called from
236 both the standalone ``main()`` command and the unified CLI's
237 ``nexuslims instruments manage`` subcommand.
238 """
239 from nexusLIMS.tui.apps.instruments import InstrumentManagerApp # noqa: PLC0415
241 # Configure logging (quiet for TUI mode)
242 logging.basicConfig(
243 level=logging.WARNING,
244 format="%(levelname)s: %(message)s",
245 )
247 # Pass db_path explicitly so the TUI app doesn't need to access
248 # config.settings (which would require all settings to be valid)
249 db_path = Path(os.getenv("NX_DB_PATH"))
251 # Launch the TUI app
252 try:
253 app = InstrumentManagerApp(db_path=db_path)
254 app.run()
255 except KeyboardInterrupt:
256 # Clean exit on Ctrl+C
257 click.echo("\nExiting...", err=True)
258 except Exception as e:
259 # Show error and exit
260 click.echo(f"Error: {e}", err=True)
261 logger.exception("Failed to run instrument manager")
262 raise click.Abort from e
265@click.command()
266@click.version_option(
267 version=None,
268 message="This standalone command is deprecated. Use: nexuslims instruments manage",
269)
270def main():
271 """
272 Manage NexusLIMS instruments database.
274 Launch an interactive terminal UI for adding, editing, and deleting
275 instruments in the NexusLIMS database. Provides form validation,
276 uniqueness checks, and confirmation prompts for destructive actions.
278 \b
279 Keybindings:
280 ------------
281 - a Add new instrument
282 - e Edit selected instrument
283 - d Delete selected instrument
284 - r Refresh list
285 - Ctrl+T Toggle theme (dark/light)
286 - ? Show help
287 - Ctrl+Q Quit
289 \b
290 Example:
291 --------
292 $ nexuslims instruments manage
294 The TUI will display a table of all instruments. Use arrow keys to
295 navigate and press Enter or 'e' to edit the selected instrument.
296 """ # noqa: D301
297 _ensure_database_initialized()
298 _run_instrument_manager()
301if __name__ == "__main__": # pragma: no cover
302 main()