View¶

class View(ABC):
    """Base view class.

    Does not inherit [`PView`][redsun.view.PView], whose read-only ``name``
    property would shadow the instance attribute set here. Instances satisfy
    the protocol by shape, like any other view.

    Parameters
    ----------
    name : str
        Identity key of the view, positional-only.
    kwargs : Any, optional
        Additional keyword arguments for view subclasses.
    """

    name: str

    @abstractmethod
    def __init__(
        self,
        name: str,
        /,
        **kwargs: Any,
    ) -> None:
        self.name = name
        super().__init__(**kwargs)

    @property
    @abstractmethod
    def view_position(self) -> ViewPosition:
        """Position of the view in the main window."""

@runtime_checkable
class PView(Protocol):
    """Protocol of a view component.

    ``name`` is a read-only property, so an instance attribute, a class
    attribute or a property satisfies it.

    Notes
    -----
    A view reaches the virtual container by implementing
    [`IsInjectable`][redsun.virtual.IsInjectable].

    Checked with ``isinstance`` on the built instance, since attributes
    assigned in ``__init__`` do not exist on the class.
    """

    @property
    def name(self) -> str:
        """Identity key of the view."""
        ...

    @property
    @abstractmethod
    def view_position(self) -> ViewPosition:
        """Position of the view in the main window."""

@unique
class ViewPosition(str, Enum):
    """Where a view sits in the main window.

    !!! warning
        The values follow Qt's dock widget areas and may change.
    """

    CENTER = "center"
    LEFT = "left"
    RIGHT = "right"
    TOP = "top"
    BOTTOM = "bottom"

@dataclass(frozen=True)
class PlanWidget:
    """The Qt widgets of one plan."""

    spec: PlanSpec
    """The plan's specification."""

    group_box: QtW.QWidget
    """The top-level page, for a QStackedWidget."""

    run_button: QtW.QPushButton
    """The button running or stopping the plan."""

    container: mgw.Container[mgw_bases.ValueWidget[Any]]
    """The ``magicgui`` Container of parameter widgets."""

    device_widgets: list[mgw_bases.ValueWidget[Any]]
    """Device parameter widgets (``DeviceSequenceEdit`` or ``ComboBox``).

    Exposed so callers can connect validation to each widget's ``changed``
    signal.
    """

    params_widget: QtW.QWidget
    """Widget holding devices_group and params_group; disabling it locks every
    parameter input but not the run, stop and pause buttons.
    """

    action_buttons: dict[str, ActionButton]
    """Action buttons by action name."""

    actions_group: QtW.QGroupBox | None = None
    """The group box of action buttons, or None if the plan has no actions."""

    pause_button: QtW.QPushButton | None = None
    """The pause/resume button, or None if the plan is not pausable."""

    def toggle(self, status: bool) -> None:
        """Update the widgets when a togglable plan starts or stops.

        Parameters
        ----------
        status : bool
            `True` when the plan is starting; `False` when stopping.
        """
        self.run_button.setText("Stop" if status else "Run")
        if self.pause_button:
            self.pause_button.setEnabled(status)
        if self.actions_group:
            self.actions_group.setEnabled(status)
        self.params_widget.setEnabled(not status)

    def pause(self, status: bool) -> None:
        """Update the widgets when a plan pauses or resumes.

        Parameters
        ----------
        status : bool
            `True` when pausing; `False` when resuming.
        """
        if self.pause_button:
            self.pause_button.setText("Resume" if status else "Pause")
            self.run_button.setEnabled(not status)

    def setEnabled(self, enabled: bool) -> None:
        """Enable or disable the whole plan widget.

        Parameters
        ----------
        enabled : bool
            ``True`` to enable; ``False`` to disable.
        """
        self.group_box.setEnabled(enabled)
        self.run_button.setEnabled(enabled)
        self.params_widget.setEnabled(enabled)

    def enable_actions(self, enabled: bool = True) -> None:
        """Enable or disable the actions group box.

        Parameters
        ----------
        enabled : bool, optional
            ``True`` to enable; ``False`` to disable.
        """
        if self.actions_group:
            self.actions_group.setEnabled(enabled)

    def get_action_button(self, action_name: str) -> ActionButton | None:
        """Return the `ActionButton` for `action_name`, or `None` if absent.

        Parameters
        ----------
        action_name : str
            The name of the action.
        """
        return self.action_buttons.get(action_name)

    def has_actions(self) -> bool:
        """Return `True` if the plan has an action button."""
        return bool(self.action_buttons)

    @property
    def parameters(self) -> dict[str, Any]:
        """Current parameter values by name.

        The presenter turns them into positional and keyword arguments with
        ``collect_arguments`` / ``resolve_arguments``.
        """
        return {w.name: w.value for w in self.container}

class LogView(QtView):
    """Read-only console of the running session's log records.

    Records logged before the view existed are shown too, read from the session
    buffer. The level selector sets the lowest level shown; since the view
    redraws from the buffer, lowering the level again brings records back.

    Application and service records have their own tabs. The Services tab
    appears once a service logs, and its selector narrows it to one service.
    ``Clear log window`` and ``Save logs...`` act on the tab and service shown.

    Records are coloured by level, with one palette for light and one for dark
    backgrounds, chosen from the console's background and redrawn when the
    palette changes.

    New records are drawn in batches, so a burst does not stall the window, and
    each console keeps no more lines than the buffer holds for it.

    With session log files open, ``Save logs...`` copies the files of the tab
    shown and ``Open log folder`` opens their folder in the file browser;
    without them the folder button is disabled.

    Parameters
    ----------
    name : str
        Identity key of the view, positional-only.
    kwargs : Any, optional
        Additional keyword arguments (unused).
    """

    @property
    def view_position(self) -> ViewPosition:
        """Position in the main window."""
        return ViewPosition.BOTTOM

    def __init__(self, name: str, /, **kwargs: Any) -> None:
        super().__init__(name, **kwargs)

        self._formatter = GlobalFormatter(datefmt="%d-%m-%y|%H:%M:%S")
        self._level = logging.INFO

        buffer = log_buffer()

        self._console = self._make_console(buffer.capacity)
        self._service_console = self._make_console(buffer.service_capacity)
        self._service_combo = QtW.QComboBox(self)
        self._service_combo.addItem(_ALL_SERVICES, None)
        services_page = QtW.QWidget(self)
        services_layout = QtW.QVBoxLayout(services_page)
        services_layout.setContentsMargins(0, 0, 0, 0)
        services_layout.addWidget(self._service_combo)
        services_layout.addWidget(self._service_console)
        self._tabs = QtW.QTabWidget(self)
        self._tabs.addTab(self._console, "Application")
        self._tabs.addTab(services_page, "Services")
        self._tabs.setTabVisible(_SERVICES_TAB, False)
        # only the newest records can end up on screen, so a burst larger than
        # the buffer never queues more than a console would keep
        self._pending: deque[logging.LogRecord] = deque(maxlen=buffer.capacity)
        self._service_pending: deque[logging.LogRecord] = deque(
            maxlen=buffer.service_capacity
        )
        for service in buffer.services:
            self._add_service(service)
        self._service_combo.currentIndexChanged.connect(self._on_service_selected)

        self._level_combo = QtW.QComboBox(self)
        for label, level in _LEVELS:
            self._level_combo.addItem(label, level)
        self._level_combo.setCurrentIndex(self._level_combo.findData(self._level))
        self._level_combo.currentIndexChanged.connect(self._on_level_selected)

        self._save_button = QtW.QPushButton("Save logs...", self)
        self._save_button.clicked.connect(self._on_save_clicked)
        self._clear_button = QtW.QPushButton("Clear log window", self)
        self._clear_button.clicked.connect(self.clear)
        self._folder_button = QtW.QPushButton("Open log folder", self)
        self._folder_button.clicked.connect(self._on_folder_clicked)
        handler = session_log()
        self._folder_button.setEnabled(handler is not None)
        if handler is not None:
            self._folder_button.setToolTip(str(Path(handler.baseFilename).parent))

        root = QtW.QGridLayout(self)
        root.addWidget(self._tabs, 0, 0, 1, 4)
        root.addWidget(QtW.QLabel("Level:", self), 1, 0)
        root.addWidget(self._level_combo, 1, 1, 1, 3)
        root.addWidget(self._save_button, 2, 1)
        root.addWidget(self._clear_button, 2, 2)
        root.addWidget(self._folder_button, 2, 3)
        # the label column keeps its own width; the three that carry the
        # buttons share the rest evenly, so the combo box spans exactly them
        for column in (1, 2, 3):
            root.setColumnStretch(column, 1)
        self.setLayout(root)

        self._batch_timer = QtCore.QTimer(self)
        self._batch_timer.setInterval(_BATCH_INTERVAL_MS)
        self._batch_timer.timeout.connect(self._draw_batch)

        self._render()
        # psygnal holds the bound method weakly, so a destroyed view drops out
        # of the buffer on its own: a view is never asked to shut down
        buffer.sig_record.connect(self._on_record, thread="main")

    def _make_console(self, capacity: int) -> QtW.QPlainTextEdit:
        console = QtW.QPlainTextEdit(self)
        console.setReadOnly(True)
        console.setMaximumBlockCount(capacity)
        font = QtGui.QFont("nosuchfont")
        font.setStyleHint(QtGui.QFont.StyleHint.Monospace)
        console.setFont(font)
        return console

    def changeEvent(self, event: QtCore.QEvent | None) -> None:
        """Redraw in the colours of the palette the console now carries."""
        if event is not None:
            super().changeEvent(event)
            if event.type() == QtCore.QEvent.Type.PaletteChange:
                self._render()

    def closeEvent(self, event: QtGui.QCloseEvent | None) -> None:
        """Stop following the buffer once the console is closed."""
        log_buffer().sig_record.disconnect(self._on_record, missing_ok=True)
        self._batch_timer.stop()
        self._pending.clear()
        self._service_pending.clear()
        if event is not None:
            super().closeEvent(event)

    @property
    def level(self) -> int:
        """The lowest level currently displayed."""
        return self._level

    @property
    def service(self) -> str | None:
        """The service the Services tab shows, ``None`` for every service."""
        data = self._service_combo.currentData()
        return None if data is None else str(data)

    def set_level(self, level: int) -> None:
        """Show only records at or above *level*, redrawing from the buffer."""
        self._level = level
        index = self._level_combo.findData(level)
        if index != -1 and index != self._level_combo.currentIndex():
            # the selection change comes back through _on_level_selected,
            # which renders once the combo agrees with the level
            self._level_combo.setCurrentIndex(index)
            return
        self._render()

    def _on_level_selected(self, index: int) -> None:
        self.set_level(int(self._level_combo.itemData(index)))

    def _on_service_selected(self, index: int) -> None:
        self._render()

    def clear(self) -> None:
        """Empty the console of the tab shown.

        The buffer is untouched, so ``Save logs...`` still writes everything and
        changing the level brings records back.
        """
        if self._tabs.currentIndex() == _SERVICES_TAB:
            self._service_pending.clear()
            self._service_console.clear()
        else:
            self._pending.clear()
            self._console.clear()

    def save(self, path: str) -> None:
        """Write the records of the tab shown to *path*, whatever the displayed level.

        The Application tab writes application records; the Services tab the
        selected service's, or every service's in turn. They come from the
        session's log files if open, so records the buffer dropped are
        included, and from the buffer otherwise.
        """
        buffer = log_buffer()
        if self._tabs.currentIndex() != _SERVICES_TAB:
            sources: list[tuple[str | None, Iterable[logging.LogRecord]]] = [
                (None, buffer.records)
            ]
        else:
            names = buffer.services if self.service is None else (self.service,)
            sources = [(name, buffer.service_records(name)) for name in names]
        with open(path, "w", encoding="utf-8") as fh:
            for source, records in sources:
                handler = session_log(source)
                if handler is None:
                    fh.writelines(
                        f"{self._formatter.format(record)}\n" for record in records
                    )
                    continue
                handler.flush()
                fh.writelines(
                    file.read_text(encoding="utf-8") for file in handler.files
                )

    def _add_service(self, service: str) -> None:
        """Offer *service* in the selector, and show the Services tab."""
        self._service_combo.addItem(service, service)
        self._tabs.setTabVisible(_SERVICES_TAB, True)
        capacity = log_buffer().service_capacity * (self._service_combo.count() - 1)
        self._service_console.setMaximumBlockCount(capacity)
        self._service_pending = deque(self._service_pending, maxlen=capacity)

    def _on_record(self, record: logging.LogRecord) -> None:
        service = service_of(record)
        if service is not None and self._service_combo.findData(service) == -1:
            self._add_service(service)
        if record.levelno < self._level:
            return
        if service is None:
            self._pending.append(record)
        elif self.service in (None, service):
            self._service_pending.append(record)
        else:
            return
        if not self._batch_timer.isActive():
            self._batch_timer.start()

    def _draw_batch(self) -> None:
        for pending, console in (
            (self._pending, self._console),
            (self._service_pending, self._service_console),
        ):
            count = min(_BATCH_SIZE, len(pending))
            self._write(console, [pending.popleft() for _ in range(count)])
        if not (self._pending or self._service_pending):
            self._batch_timer.stop()

    def _render(self) -> None:
        buffer = log_buffer()
        self._batch_timer.stop()
        self._pending.clear()
        self._service_pending.clear()
        self._console.clear()
        self._service_console.clear()
        self._write(
            self._console, [r for r in buffer.records if r.levelno >= self._level]
        )
        self._write(
            self._service_console,
            [
                r
                for r in buffer.service_records(self.service)
                if r.levelno >= self._level
            ],
        )

    @property
    def colors(self) -> dict[int, str]:
        """The level colours in use, chosen from the console's background."""
        base = self._console.palette().color(QtGui.QPalette.ColorRole.Base)
        return _ON_LIGHT if base.lightness() >= _MID_LIGHTNESS else _ON_DARK

    def _write(
        self, console: QtW.QPlainTextEdit, records: Iterable[logging.LogRecord]
    ) -> None:
        # pyqt6 annotates both as optional and pyside6 does not
        document: QtGui.QTextDocument | None = console.document()
        bar: QtW.QScrollBar | None = console.verticalScrollBar()
        if document is None or bar is None:
            return
        # follow the newest line only when the reader is already at the bottom
        following = bar.value() == bar.maximum()
        colors = self.colors
        formats: dict[int, QtGui.QTextCharFormat] = {}
        cursor = QtGui.QTextCursor(document)
        cursor.movePosition(QtGui.QTextCursor.MoveOperation.End)
        cursor.beginEditBlock()
        for record in records:
            if record.levelno not in formats:
                char_format = QtGui.QTextCharFormat()
                color = colors.get(record.levelno, colors[logging.INFO])
                char_format.setForeground(QtGui.QBrush(QtGui.QColor(color)))
                formats[record.levelno] = char_format
            if not document.isEmpty():
                cursor.insertBlock()
            cursor.insertText(self._formatter.format(record), formats[record.levelno])
        cursor.endEditBlock()
        if following:
            bar.setValue(bar.maximum())

    def _on_save_clicked(self) -> None:
        chosen, _ = QtW.QFileDialog.getSaveFileName(
            self, "Save session logs", "redsun.log", "Log files (*.log);;All files (*)"
        )
        if not chosen:
            return
        try:
            self.save(chosen)
        except OSError as e:
            QtW.QMessageBox.warning(self, "Could not save logs", str(e))

    def _on_folder_clicked(self) -> None:
        handler = session_log()
        if handler is None:
            return
        folder = Path(handler.baseFilename).parent
        if not QtGui.QDesktopServices.openUrl(QtCore.QUrl.fromLocalFile(str(folder))):
            QtW.QMessageBox.warning(self, "Could not open the log folder", str(folder))