qq_lib.core.operator

  1# Released under MIT License.
  2# Copyright (c) 2025-2026 Ladislav Bartos and Robert Vacha Lab
  3
  4"""
  5Base functionality for qq job operations.
  6
  7This module defines the `Operator` class, which provides a common interface for
  8working with qq jobs. It loads job information, tracks job state, refreshes
  9metadata, and renders formatted status output using Rich presenters.
 10"""
 11
 12from pathlib import Path
 13from typing import Self
 14
 15from rich.console import Console
 16
 17from qq_lib.info.informer import Informer
 18from qq_lib.info.presenter import Presenter
 19
 20
 21class Operator:
 22    """
 23    Base class for performing operations with qq jobs.
 24
 25    Attributes:
 26        _informer (Informer): The underlying informer object that provides job details.
 27        _info_file (Path): The path to the qq info file associated with this job.
 28        _input_machine (str | None): Hostname of the machine on which the qq info file is stored.
 29        _batch_system (str): The batch system type as reported by the informer.
 30        _state (RealState): The current real state of the qq job.
 31    """
 32
 33    def __init__(self, info_file: Path, host: str | None = None):
 34        """
 35        Initialize an Operator instance from a qq info file.
 36
 37        Args:
 38            info_file (Path): Path to the qq info file describing the job.
 39            host (str | None, optional): Optional hostname of a machine from
 40                which to load job information. Defaults to None meaning 'current machine'.
 41        """
 42        self._informer = Informer.from_file(info_file, host)
 43        self._info_file = info_file
 44        self._input_machine = host
 45        self._batch_system = self._informer.batch_system
 46        self._state = self._informer.get_real_state()
 47
 48    @classmethod
 49    def from_informer(cls, informer: Informer) -> Self:
 50        """
 51        Initialize an Operator instance from an Informer.
 52
 53        Path to info file is set based on the information in the Informer, even if it does not exist.
 54
 55        Args:
 56            informer (Informer): Initialized informer instance containing information about the job.
 57
 58        Returns:
 59            Operator: Initialized Operator.
 60        """
 61        operator = cls.__new__(cls)
 62        operator._informer = informer
 63        operator._info_file = informer.get_info_file()
 64        operator._input_machine = informer.info.input_machine
 65        operator._batch_system = informer.batch_system
 66        operator._state = informer.get_real_state()
 67
 68        return operator
 69
 70    def update(self) -> None:
 71        """
 72        Refresh the internal informer and job state from the qq info file.
 73        """
 74        self._informer = Informer.from_file(self._info_file, self._input_machine)
 75        self._state = self._informer.get_real_state()
 76
 77    def get_informer(self) -> Informer:
 78        """
 79        Retrieve the underlying Informer instance.
 80
 81        Returns:
 82            Informer: The informer currently associated with this operator.
 83        """
 84        return self._informer
 85
 86    def print_info(self, console: Console) -> None:
 87        """
 88        Display the current job information in a formatted Rich panel.
 89
 90        Args:
 91            console (Console): Rich Console instance used to render output.
 92        """
 93        presenter = Presenter(self._informer)
 94        panel = presenter.create_job_status_panel(console)
 95        console.print(panel)
 96
 97    def matches_job(self, job_id: str) -> bool:
 98        """
 99        Determine whether this operator corresponds to the specified job ID.
100
101        Args:
102            job_id (str): The job ID to compare against (e.g., "12345" or "12345.cluster.domain").
103
104        Returns:
105            bool: True if both job IDs refer to the same job (same numeric/job part),
106                False otherwise.
107        """
108        return self._informer.matches_job(job_id)