Coverage for nexusLIMS/tui/apps/instruments/screens.py: 100%
189 statements
« prev ^ index » next coverage.py v7.11.3, created at 2026-03-24 05:23 +0000
1"""
2Screens for the instrument management TUI application.
4Provides List, Add, Edit, and Delete screens for instrument CRUD operations.
5"""
7from pathlib import Path
8from typing import ClassVar
10import pytz
11from sqlmodel import select
12from textual.app import ComposeResult
13from textual.containers import Vertical
14from textual.screen import ModalScreen
15from textual.widgets import Button, Input, Label, Select, Static
17from nexusLIMS.db.models import Instrument
18from nexusLIMS.tui.apps.instruments.validators import (
19 get_example_values,
20 validate_api_url_unique,
21 validate_instrument_pid,
22)
23from nexusLIMS.tui.common.base_screens import (
24 BaseFormScreen,
25 BaseListScreen,
26 ConfirmDialog,
27)
28from nexusLIMS.tui.common.db_utils import get_session_log_count
29from nexusLIMS.tui.common.validators import (
30 validate_timezone,
31)
32from nexusLIMS.tui.common.widgets import AutocompleteInput, FormField, NumpadInput
35class WelcomeDialog(ModalScreen):
36 """Welcome dialog shown when instruments table is empty."""
38 CSS_PATH: ClassVar = [
39 Path(__file__).parent.parent.parent
40 / "styles"
41 / "instruments"
42 / "welcome_dialog.tcss"
43 ]
45 def compose(self) -> ComposeResult:
46 """Compose the welcome dialog."""
47 with Vertical(id="welcome-dialog"):
48 yield Label(
49 "Welcome to the NexusLIMS Instrument Manager!", id="welcome-title"
50 )
51 yield Static(
52 "No instruments were found in the database! If this is your first time "
53 "running NexusLIMS, welcome!\n\n"
54 "To get started, add your first instrument by pressing 'a' or clicking "
55 "the button below.\n\n"
56 "You can also:\n"
57 "• Press '?' for help and keybindings\n"
58 "• Press Ctrl+T to toggle dark/light theme\n"
59 "• Press Ctrl+Q or 'q' to quit",
60 id="welcome-message",
61 )
62 with Vertical(id="welcome-buttons"):
63 yield Button(
64 "Add First Instrument (a)", id="add-btn", variant="primary"
65 )
66 yield Button("Close", id="close-btn", variant="default")
68 def on_button_pressed(self, event: Button.Pressed) -> None:
69 """Handle button presses."""
70 if event.button.id == "add-btn":
71 self.dismiss(True) # Signal to open add screen
72 else:
73 self.dismiss(False) # Just close
76class InstrumentListScreen(BaseListScreen):
77 """Screen displaying all instruments in a table."""
79 def on_mount(self) -> None:
80 """Set up the screen and check if welcome dialog should be shown."""
81 super().on_mount()
83 # Check if instruments table is empty
84 instruments = self.app.db_session.exec(select(Instrument)).all()
86 if not instruments:
87 # Show welcome dialog
88 self.app.push_screen(WelcomeDialog(), self.on_welcome_complete)
90 def on_welcome_complete(self, should_add: bool) -> None:
91 """Handle welcome dialog completion."""
92 if should_add:
93 # User wants to add first instrument
94 self.action_add()
96 def get_columns(self) -> list[str]:
97 """Get column headers."""
98 return [
99 "Display Name",
100 "PID",
101 "API URL",
102 "Filestore Path",
103 "Location",
104 "Property Tag",
105 "Timezone",
106 ]
108 def get_data(self) -> list[dict]:
109 """Get instrument data from database."""
110 instruments = self.app.db_session.exec(select(Instrument)).all()
112 data = []
113 for instr in instruments:
114 data.append(
115 {
116 "Display Name": instr.display_name,
117 "PID": instr.instrument_pid,
118 "API URL": instr.api_url,
119 "Filestore Path": instr.filestore_path,
120 "Location": instr.location,
121 "Property Tag": instr.property_tag,
122 "Timezone": instr.timezone_str,
123 }
124 )
126 return data
128 def on_row_selected(self, _row_key, row_data: dict) -> None:
129 """Handle row selection - show edit screen."""
130 # Load full instrument from database
131 instrument_pid = row_data["PID"]
132 instrument = self.app.db_session.get(Instrument, instrument_pid)
134 if instrument:
135 # Push edit screen
136 edit_screen = InstrumentEditScreen(instrument)
137 self.app.push_screen(edit_screen, self.on_edit_complete)
139 def on_edit_complete(self, result) -> None:
140 """Handle edit screen completion."""
141 if result:
142 self.refresh_data()
144 def action_add(self) -> None:
145 """Show add instrument screen."""
146 add_screen = InstrumentAddScreen()
147 self.app.push_screen(add_screen, self.on_add_complete)
149 def on_add_complete(self, result) -> None:
150 """Handle add screen completion."""
151 if result:
152 self.refresh_data()
154 def action_delete(self) -> None:
155 """Delete selected instrument."""
156 table = self.query_one("DataTable")
157 if table.cursor_row is not None and table.row_count > 0:
158 # Get instrument PID from current row
159 columns = self.get_columns()
160 cursor_row = table.cursor_row
161 row_data = {}
163 for i, column in enumerate(columns):
164 row_data[column] = table.get_row_at(cursor_row)[i]
166 instrument_pid = row_data["PID"]
168 # Load instrument
169 instrument = self.app.db_session.get(Instrument, instrument_pid)
170 if instrument:
171 # Check for session logs
172 session_count = get_session_log_count(
173 self.app.db_session, instrument_pid
174 )
176 # Build confirmation message
177 message = f"Delete instrument '{instrument_pid}'?"
178 if session_count > 0:
179 message += (
180 f"\n\nWarning: This instrument has {session_count} "
181 "session log entries.\nThese entries will NOT be deleted "
182 "but will reference a non-existent instrument."
183 )
185 # Show confirmation dialog
186 def on_confirm(confirmed: bool):
187 if confirmed:
188 self.delete_instrument(instrument_pid)
190 self.app.push_screen(
191 ConfirmDialog(message, title="Confirm Delete"),
192 on_confirm,
193 )
195 def delete_instrument(self, instrument_pid: str) -> None:
196 """Delete an instrument from the database."""
197 try:
198 instrument = self.app.db_session.get(Instrument, instrument_pid)
199 if instrument:
200 self.app.db_session.delete(instrument)
201 self.app.db_session.commit()
202 self.app.show_success(f"Deleted instrument: {instrument_pid}")
203 self.refresh_data()
204 except Exception as e:
205 self.app.db_session.rollback()
206 self.app.show_error(f"Failed to delete instrument: {e}")
209class InstrumentAddScreen(BaseFormScreen):
210 """Screen for adding a new instrument."""
212 # Disable auto-focus to prevent scrolling to first input
213 AUTO_FOCUS = ""
215 CSS_PATH: ClassVar = [
216 Path(__file__).parent.parent.parent / "styles" / "instruments" / "screens.tcss"
217 ]
219 def __init__(self, **kwargs):
220 """Initialize add screen."""
221 super().__init__(title="Add New Instrument", **kwargs)
222 self.examples = get_example_values()
224 def on_mount(self) -> None:
225 """Focus first input without scrolling."""
226 # Focus the first input without scrolling the viewport
227 first_input = self.query_one("#instrument_pid", Input)
228 first_input.focus(scroll_visible=False)
230 def get_form_fields(self) -> ComposeResult:
231 """Generate form fields for instrument creation in two columns."""
232 with Vertical(classes="form-column"):
233 yield FormField(
234 "Instrument PID",
235 NumpadInput(
236 placeholder=self.examples["instrument_pid"],
237 id="instrument_pid",
238 ),
239 required=True,
240 help_text=(
241 f"Unique identifier (e.g., {self.examples['instrument_pid']})"
242 ),
243 )
244 yield FormField(
245 "API URL",
246 Input(
247 placeholder=self.examples["api_url"],
248 id="api_url",
249 ),
250 required=True,
251 help_text=(
252 f"Calendar API endpoint URL (e.g., {self.examples['api_url']})"
253 ),
254 )
255 yield FormField(
256 "Calendar URL",
257 Input(
258 placeholder=self.examples["calendar_url"],
259 id="calendar_url",
260 ),
261 required=True,
262 help_text=(
263 "Web-accessible calendar URL"
264 f" (e.g., {self.examples['calendar_url']})"
265 ),
266 )
267 yield FormField(
268 "Location",
269 Input(
270 placeholder=self.examples["location"],
271 id="location",
272 ),
273 required=True,
274 help_text=f"Physical location (e.g., {self.examples['location']})",
275 )
276 yield FormField(
277 "Display Name",
278 Input(
279 placeholder=self.examples["display_name"],
280 id="display_name",
281 ),
282 required=True,
283 help_text=(
284 f"Human-readable instrument name for NexusLIMS records "
285 f"(e.g., {self.examples['display_name']})"
286 ),
287 )
289 with Vertical(classes="form-column"):
290 yield FormField(
291 "Property Tag",
292 Input(
293 placeholder=self.examples["property_tag"],
294 id="property_tag",
295 ),
296 required=True,
297 help_text=(
298 f"Unique numeric identifier (e.g., {self.examples['property_tag']})"
299 ),
300 )
301 yield FormField(
302 "Filestore Path",
303 Input(
304 placeholder=self.examples["filestore_path"],
305 id="filestore_path",
306 ),
307 required=True,
308 help_text=(
309 f"Relative path under NX_INSTRUMENT_DATA_PATH "
310 f"(e.g., {self.examples['filestore_path']})"
311 ),
312 )
313 yield FormField(
314 "Harvester",
315 Select(
316 [("nemo", "nemo")],
317 value="nemo",
318 id="harvester",
319 ),
320 required=True,
321 help_text='Harvester module ("nemo" is the only option, currently)',
322 )
323 yield FormField(
324 "Timezone",
325 AutocompleteInput(
326 suggestions=pytz.common_timezones,
327 placeholder=self.examples["timezone_str"],
328 value="America/New_York",
329 id="timezone_str",
330 ),
331 required=True,
332 help_text="IANA timezone (e.g., America/New_York)",
333 )
335 def collect_form_data(self) -> dict:
336 """Collect data from form fields."""
337 # Get form fields
338 return {
339 "instrument_pid": self.query_one("#instrument_pid", Input).value,
340 "api_url": self.query_one("#api_url", Input).value,
341 "calendar_url": self.query_one("#calendar_url", Input).value,
342 "location": self.query_one("#location", Input).value,
343 "display_name": self.query_one("#display_name", Input).value,
344 "property_tag": self.query_one("#property_tag", Input).value,
345 "filestore_path": self.query_one("#filestore_path", Input).value,
346 "harvester": self.query_one("#harvester", Select).value,
347 "timezone_str": self.query_one("#timezone_str", Input).value,
348 }
350 def validate_form(self) -> dict[str, str]:
351 """Validate form data."""
352 errors = {}
353 data = self.collect_form_data()
355 # Validate instrument_pid
356 is_valid, error = validate_instrument_pid(data["instrument_pid"])
357 if not is_valid:
358 errors["instrument_pid"] = error
360 # Validate api_url (with uniqueness check)
361 is_valid, error = validate_api_url_unique(self.app.db_session, data["api_url"])
362 if not is_valid:
363 errors["api_url"] = error
365 # Validate timezone
366 is_valid, error = validate_timezone(data["timezone_str"])
367 if not is_valid:
368 errors["timezone_str"] = error
370 # Show a notification if there are validation errors (without logging as error)
371 if errors:
372 error_count = len(errors)
373 field_names = ", ".join(errors.keys())
374 msg = (
375 f"Validation failed: {error_count} error(s) in {field_names}. "
376 "See details at bottom of form."
377 )
378 self.app.notify(msg, severity="warning", timeout=5)
380 return errors
382 def on_save(self, data: dict) -> None:
383 """Save new instrument to database."""
384 try:
385 # Create instrument
386 instrument = Instrument(**data)
388 # Add to database
389 self.app.db_session.add(instrument)
390 self.app.db_session.commit()
392 self.app.show_success(f"Created instrument: {data['instrument_pid']}")
393 self.dismiss(True)
395 except Exception as e:
396 self.app.db_session.rollback()
397 self.app.show_error(f"Failed to create instrument: {e}")
400class InstrumentEditScreen(InstrumentAddScreen):
401 """Screen for editing an existing instrument."""
403 def __init__(self, instrument: Instrument, **kwargs):
404 """
405 Initialize edit screen.
407 Parameters
408 ----------
409 instrument : Instrument
410 Instrument to edit
411 **kwargs
412 Additional arguments passed to BaseFormScreen
413 """
414 self.instrument = instrument
415 super().__init__(**kwargs)
416 self.screen_title = f"Edit Instrument: {instrument.instrument_pid}"
418 def on_mount(self) -> None:
419 """Populate form with existing instrument data."""
420 super().on_mount()
422 # Populate fields
423 self.query_one("#instrument_pid", Input).value = self.instrument.instrument_pid
424 self.query_one("#instrument_pid", Input).disabled = True # Can't change PID
426 self.query_one("#api_url", Input).value = self.instrument.api_url
427 self.query_one("#calendar_url", Input).value = self.instrument.calendar_url
428 self.query_one("#location", Input).value = self.instrument.location
429 self.query_one("#display_name", Input).value = self.instrument.display_name
430 self.query_one("#property_tag", Input).value = self.instrument.property_tag
431 self.query_one("#filestore_path", Input).value = self.instrument.filestore_path
432 self.query_one("#harvester", Select).value = self.instrument.harvester
433 self.query_one("#timezone_str", Input).value = self.instrument.timezone_str
435 def validate_form(self) -> dict[str, str]:
436 """Validate form data (excluding PID from uniqueness checks)."""
437 errors = {}
438 data = self.collect_form_data()
440 # Validate api_url (with uniqueness check, excluding current instrument)
441 is_valid, error = validate_api_url_unique(
442 self.app.db_session,
443 data["api_url"],
444 exclude_pid=self.instrument.instrument_pid,
445 )
446 if not is_valid:
447 errors["api_url"] = error
449 # Validate timezone
450 is_valid, error = validate_timezone(data["timezone_str"])
451 if not is_valid:
452 errors["timezone_str"] = error
454 # Show a notification if there are validation errors (without logging as error)
455 if errors:
456 error_count = len(errors)
457 field_names = ", ".join(errors.keys())
458 msg = (
459 f"Validation failed: {error_count} error(s) in {field_names}. "
460 "See details at bottom of form."
461 )
462 self.app.notify(msg, severity="warning", timeout=5)
464 return errors
466 def on_save(self, data: dict) -> None:
467 """Update existing instrument in database."""
468 try:
469 # Update instrument fields (except PID)
470 self.instrument.api_url = data["api_url"]
471 self.instrument.calendar_url = data["calendar_url"]
472 self.instrument.location = data["location"]
473 self.instrument.display_name = data["display_name"]
474 self.instrument.property_tag = data["property_tag"]
475 self.instrument.filestore_path = data["filestore_path"]
476 self.instrument.harvester = data["harvester"]
477 self.instrument.timezone_str = data["timezone_str"]
479 # Commit changes
480 self.app.db_session.add(self.instrument)
481 self.app.db_session.commit()
483 self.app.show_success(f"Updated instrument: {data['instrument_pid']}")
484 self.dismiss(True)
486 except Exception as e:
487 self.app.db_session.rollback()
488 self.app.show_error(f"Failed to update instrument: {e}")