"""
The single-run lock for the PypeIt Dashboard (design X1).
At most **one** PypeIt run (``run_pypeit`` or ``pypeit_run_to_calibstep``) may
be active at a time, and the Dashboard must refuse to launch a second. It
becomes "locked" in two situations:
* a run **the Dashboard launched** is still in progress (driven by the
:class:`~pypeit.dashboard.launcher.Launcher` via
:meth:`RunLock.set_dashboard_running`); and
* a run started **outside** the Dashboard is detected by watching the
reduction ``.log`` file — PypeIt writes to it continuously while running, so
a log whose modification time is recent means a run is in progress.
The mtime test (:meth:`RunLock._is_recent`) is a pure, unit-testable function;
the polling :class:`~qtpy.QtCore.QTimer` only calls it. The lock emits
:attr:`RunLock.lockChanged` when it transitions, which the views connect to so
they can enable/disable their launch controls.
Generated by JXP and Claude.
"""
import time
from pathlib import Path
from qtpy.QtCore import QObject, QTimer, Signal
[docs]
class RunLock(QObject):
"""
Single-run lock with ``.log``-mtime external-run detection (design X1),
and the live-monitoring change signal (design R14, Stage 5).
The one polling timer serves both roles: it detects whether a run is
**active** (the ``.log`` mtime is recent) and, while active, whether the
reduction **state file** has changed (its mtime advanced) — emitting
:attr:`stateChanged` so the views can refresh live (S5-Q3).
Generated by JXP and Claude.
Args:
log_path (:obj:`str`, :obj:`pathlib.Path`, optional):
The reduction ``.log`` file to watch. If ``None``, only
Dashboard-launched runs lock (no external detection).
state_path (:obj:`str`, :obj:`pathlib.Path`, optional):
The ``*_state.json`` file to watch for live updates. If ``None``,
:attr:`stateChanged` never fires.
parent (:obj:`QObject`, optional):
The Qt parent.
Attributes:
lockChanged (Signal):
Emitted with the new locked state (:obj:`bool`) on each
transition.
stateChanged (Signal):
Emitted (no args) when the state file's mtime advances **while a
run is active** (Stage 5 live monitoring).
"""
#: Emitted with the new locked state (bool) whenever it changes.
lockChanged = Signal(bool)
#: Emitted when the state file changes while a run is active (Stage 5).
stateChanged = Signal()
#: A ``.log`` modified within this many seconds means a run is active
#: (PypeIt writes to it continuously; a quiet log means it finished or
#: stalled).
ACTIVE_WINDOW_S = 10.0
#: Polling cadence for the external-run + state-change check
#: (S4-Q5/S5-Q2: ~2–3 s; one timer for both).
POLL_MS = 2500
def __init__(self, log_path=None, state_path=None, parent=None):
super().__init__(parent=parent)
self._log_path = Path(log_path) if log_path is not None else None
self._state_path = Path(state_path) if state_path is not None else None
# Locked-by-our-own-run and locked-by-an-external-run, tracked
# separately so either can hold the lock.
self._dashboard_running = False
self._external_active = False
# Last-seen state-file mtime, for live-monitoring change detection.
self._last_state_mtime = self._state_mtime()
self._timer = QTimer(self)
self._timer.setInterval(self.POLL_MS)
self._timer.timeout.connect(self.poll)
# -- lifecycle -------------------------------------------------------
[docs]
def start(self):
"""
Start polling the ``.log`` for external runs (and check once now).
Generated by JXP and Claude.
Returns:
None.
"""
self._timer.start()
self.poll()
[docs]
def stop(self):
"""
Stop polling.
Generated by JXP and Claude.
Returns:
None.
"""
self._timer.stop()
# -- state -----------------------------------------------------------
[docs]
def is_locked(self):
"""
Whether a run is active (Dashboard-launched or external).
Generated by JXP and Claude.
Returns:
bool: True if the Dashboard should refuse to launch a run.
"""
return self._dashboard_running or self._external_active
[docs]
def set_dashboard_running(self, running):
"""
Record whether a run the Dashboard launched is in progress, emitting
:attr:`lockChanged` if the locked state changes.
Generated by JXP and Claude.
Args:
running (:obj:`bool`): True while our launched run is active.
Returns:
None.
"""
running = bool(running)
if running == self._dashboard_running:
return
was = self.is_locked()
self._dashboard_running = running
self._emit_if_changed(was)
[docs]
@staticmethod
def _is_recent(mtime, now, window):
"""
Whether a file modified at ``mtime`` counts as "being written now".
Pure and unit-testable: ``True`` when the log was touched within
``window`` seconds of ``now``.
Generated by JXP and Claude.
Args:
mtime (:obj:`float`): The file modification time (epoch seconds).
now (:obj:`float`): The current time (epoch seconds).
window (:obj:`float`): The "recent" window, in seconds.
Returns:
bool: True if ``now - mtime < window``.
"""
return (now - mtime) < window
[docs]
def _state_mtime(self):
"""
The state file's modification time, or ``None`` if absent/unreadable.
Generated by JXP and Claude.
Returns:
float or None: The ``*_state.json`` mtime (epoch seconds).
"""
if self._state_path is None or not self._state_path.is_file():
return None
try:
return self._state_path.stat().st_mtime
except OSError:
return None
[docs]
def poll(self):
"""
Re-check the ``.log`` mtime for an external run (emitting
:attr:`lockChanged` on a transition) and the ``*_state.json`` mtime for
a live update (emitting :attr:`stateChanged` while active, Stage 5).
Generated by JXP and Claude.
Returns:
None.
"""
active = False
if self._log_path is not None and self._log_path.is_file():
try:
mtime = self._log_path.stat().st_mtime
active = self._is_recent(mtime, time.time(),
self.ACTIVE_WINDOW_S)
except OSError:
active = False
was = self.is_locked()
self._external_active = active
self._emit_if_changed(was)
# Live monitoring: while a run is active, signal a state-file change so
# the views refresh (R14). Track the mtime every tick so a change that
# happened while idle does not fire a stale update.
state_mtime = self._state_mtime()
if state_mtime is not None and self._last_state_mtime is not None \
and state_mtime > self._last_state_mtime and self.is_locked():
self.stateChanged.emit()
self._last_state_mtime = state_mtime
[docs]
def _emit_if_changed(self, was_locked):
"""
Emit :attr:`lockChanged` if the locked state differs from ``was``.
Generated by JXP and Claude.
Args:
was_locked (:obj:`bool`): The locked state before the change.
Returns:
None.
"""
now_locked = self.is_locked()
if now_locked != was_locked:
self.lockChanged.emit(now_locked)