"""
Launch PypeIt's external inspection tools as subprocesses (design C8/C14/C15).
Uses Qt's :obj:`QProcess` (S3-Q2) so launches are **non-blocking** and integrate
with the event loop: stdout/stderr/return code are captured and the
:class:`~pypeit.dashboard.view.activity.ActivityBar` is updated on start and
finish (observable launches, per the Debugging plan).
Generated by JXP and Claude.
"""
from pathlib import Path
from qtpy.QtCore import QProcess
from pypeit import log
[docs]
class Launcher:
"""
Launches external tools as non-blocking subprocesses and reports
progress/outcome to the activity bar.
Generated by JXP and Claude.
Args:
activity (:class:`~pypeit.dashboard.view.activity.ActivityBar`):
The status/activity bar to report to (may be ``None``).
run_lock (:class:`~pypeit.dashboard.runlock.RunLock`, optional):
The single-run lock (design X1); a run launched via :meth:`run`
holds it while active. May be ``None``.
"""
def __init__(self, activity=None, run_lock=None):
self.activity = activity
# The single-run lock (design X1); set so a launched run holds the
# lock while it is active. May be None (e.g. headless tests).
self.run_lock = run_lock
# Keep references to running processes so they are not garbage
# collected while still running.
self._procs = []
[docs]
def _report(self, message, busy=False, build=True):
"""
Report a message to the activity bar (and the log), on the **Build**
channel ((re)build runs) or the **Inspection** channel (viewer
launches) per ``build`` (Stage 5 S5-Q9).
Generated by JXP and Claude.
Args:
message (:obj:`str`): The message.
busy (:obj:`bool`, optional): Show the busy indicator.
build (:obj:`bool`, optional): Route to the Build channel (True,
default) or the Inspection channel (False).
Returns:
None.
"""
log.info(message)
if self.activity is not None:
if build:
self.activity.set_build(message, busy=busy)
else:
self.activity.set_inspection(message, busy=busy)
[docs]
def launch(self, argv, description=None, hint='viewer window'):
"""
Launch ``argv`` as a detached-from-the-UI subprocess.
Generated by JXP and Claude.
Args:
argv (:obj:`list`): The command and its arguments.
description (:obj:`str`, optional): Human description for the
activity bar (defaults to the program name).
hint (:obj:`str`, optional): Where the result appears (e.g.
``'Ginga window'``); shown in the finished message so the
user knows where to look.
Returns:
bool: True if the process was started, False if it could not be
launched (e.g. the target file is missing).
"""
if not argv:
return False
program, args = argv[0], [str(a) for a in argv[1:]]
# The exact command line, shown in quotes so the user can copy and run
# it themselves (Stage 3b #2).
cmd = '"' + ' '.join([program] + args) + '"'
# If the command references a file that does not exist, report and
# bail rather than spawning a doomed process.
for a in args:
if a.startswith('-'):
continue
if ('/' in a or a.endswith('.fits') or a.endswith('.gz')) \
and not Path(a).exists():
self._report(f'Cannot launch {cmd}: file not found ({a})',
busy=False, build=False)
return False
proc = QProcess()
proc.setProcessChannelMode(QProcess.MergedChannels)
self._procs.append(proc)
proc.started.connect(
lambda: self._report(f'Running {cmd}', busy=True, build=False))
proc.errorOccurred.connect(
lambda _err, p=proc, c=cmd: self._on_error(p, c))
proc.finished.connect(
lambda code, _status, p=proc, c=cmd, h=hint:
self._on_finished(p, c, code, h))
proc.start(program, args)
return True
[docs]
def run(self, argv, description='Rebuilding', on_finished=None):
"""
Launch a PypeIt **run** (e.g. ``pypeit_run_to_calibstep``): hold the
single-run lock while it is active and refresh on completion (design
C10/C15/X1).
Unlike :meth:`launch` (inspection viewers), a run engages the
:attr:`run_lock` so launch controls disable while it executes, and
calls ``on_finished`` when it ends so the views can re-read the state.
Generated by JXP and Claude.
Args:
argv (:obj:`list`): The run command and its arguments.
description (:obj:`str`, optional): Human label for the activity
bar (e.g. ``'(Re)building wv_calib'``).
on_finished (callable, optional): Called as ``on_finished(code)``
when the run ends (success or failure), for the
refresh-on-completion (C15).
Returns:
bool: True if the process was started, False otherwise.
"""
if not argv:
return False
program, args = argv[0], [str(a) for a in argv[1:]]
cmd = '"' + ' '.join([program] + args) + '"'
proc = QProcess()
proc.setProcessChannelMode(QProcess.MergedChannels)
self._procs.append(proc)
# Engage the lock up front so a second launch is refused immediately.
if self.run_lock is not None:
self.run_lock.set_dashboard_running(True)
proc.started.connect(
lambda d=description, c=cmd:
self._report(f'{d}: running {c}', busy=True))
proc.errorOccurred.connect(
lambda _err, p=proc, c=cmd, cb=on_finished:
self._on_run_error(p, c, cb))
proc.finished.connect(
lambda code, _status, p=proc, c=cmd, cb=on_finished:
self._on_run_finished(p, c, code, cb))
proc.start(program, args)
return True
[docs]
def _on_run_finished(self, proc, cmd, code, on_finished):
"""
Handle a finished run: release the lock, report, refresh.
Generated by JXP and Claude.
Args:
proc (:obj:`QProcess`): The finished process.
cmd (:obj:`str`): The quoted command line.
code (:obj:`int`): The exit code.
on_finished (callable): Called as ``on_finished(code)`` (or
``None``).
Returns:
None.
"""
output = bytes(proc.readAll()).decode('utf-8', errors='replace')
if code == 0:
self._report(f'Ran {cmd} — refreshing state.', busy=False)
else:
self._report(f'Ran {cmd} — exited with code {code}.', busy=False)
log.warning(f'{cmd} output:\n{output}')
if self.run_lock is not None:
self.run_lock.set_dashboard_running(False)
if on_finished is not None:
on_finished(code)
self._drop(proc)
[docs]
def _on_run_error(self, proc, cmd, on_finished):
"""
Handle a run that failed to start: release the lock and report.
Generated by JXP and Claude.
Args:
proc (:obj:`QProcess`): The failed process.
cmd (:obj:`str`): The quoted command line.
on_finished (callable): Called as ``on_finished(-1)`` (or
``None``).
Returns:
None.
"""
self._report(f'Failed to run {cmd}: {proc.errorString()}',
busy=False)
if self.run_lock is not None:
self.run_lock.set_dashboard_running(False)
if on_finished is not None:
on_finished(-1)
self._drop(proc)
[docs]
def _on_finished(self, proc, cmd, code, hint='viewer window'):
"""
Handle process completion: report the outcome and capture output.
The exact command stays shown after finishing (Stage 3b #2).
Generated by JXP and Claude.
Args:
proc (:obj:`QProcess`): The finished process.
cmd (:obj:`str`): The quoted command line.
code (:obj:`int`): The exit code.
hint (:obj:`str`, optional): Where the result appears.
Returns:
None.
"""
output = bytes(proc.readAll()).decode('utf-8', errors='replace')
if code == 0:
self._report(f'Ran {cmd} — see the {hint}.', busy=False,
build=False)
else:
self._report(f'Ran {cmd} — exited with code {code}.',
busy=False, build=False)
log.warning(f'{cmd} output:\n{output}')
self._drop(proc)
[docs]
def _on_error(self, proc, cmd):
"""
Handle an inspection process that failed to start (e.g. command not
found).
Generated by JXP and Claude.
Args:
proc (:obj:`QProcess`): The failed process.
cmd (:obj:`str`): The quoted command line.
Returns:
None.
"""
self._report(f'Failed to run {cmd}: {proc.errorString()}',
busy=False, build=False)
self._drop(proc)
[docs]
def _drop(self, proc):
"""
Release a finished process reference.
Generated by JXP and Claude.
Args:
proc (:obj:`QProcess`): The process to release.
Returns:
None.
"""
if proc in self._procs:
self._procs.remove(proc)