qq_lib.properties.info

  1# Released under MIT License.
  2# Copyright (c) 2025-2026 Ladislav Bartos and Robert Vacha Lab
  3
  4"""
  5Structured storage and serialization of qq job metadata.
  6
  7This module defines the `Info` dataclass, which provides a representation
  8of qq job information: submission parameters, resource requests, job state,
  9timing data, dependencies, and execution context. It handles
 10loading and exporting YAML info files both locally and from remote hosts, and
 11offers minimal helpers such as command-line reconstruction for resubmission.
 12
 13`Info` focuses strictly on data representation and safe serialization; higher-level
 14logic (state interpretation, batch-system interaction, consistency checks) is
 15implemented in `Informer` and related components.
 16"""
 17
 18from dataclasses import dataclass, field, fields
 19from datetime import datetime
 20from pathlib import Path
 21from typing import Self
 22
 23import yaml
 24
 25from qq_lib.batch.interface import AnyBatchClass, BatchInterface
 26from qq_lib.core.common import load_yaml_dumper, load_yaml_loader
 27from qq_lib.core.config import CFG
 28from qq_lib.core.error import QQError
 29from qq_lib.core.logger import get_logger
 30from qq_lib.properties.depend import Depend
 31from qq_lib.properties.interpreter import Interpreter
 32from qq_lib.properties.resubmit_host import ResubmitHost
 33from qq_lib.properties.transfer_mode import Success, TransferMode
 34
 35from .job_type import JobType
 36from .loop import LoopInfo
 37from .resources import Resources
 38from .states import NaiveState
 39
 40logger = get_logger(__name__)
 41
 42SafeLoader: type[yaml.SafeLoader] = load_yaml_loader()
 43Dumper: type[yaml.Dumper] = load_yaml_dumper()
 44
 45
 46@dataclass
 47class Info:
 48    """
 49    Dataclass storing information about a qq job.
 50
 51    Exposes only minimal functionality for loading, exporting, and basic access.
 52    More complex operations, such as transforming or combining the data
 53    should be implemented in Informer.
 54    """
 55
 56    # The batch system class used
 57    batch_system: AnyBatchClass
 58
 59    # Version of qq that submitted the job
 60    qq_version: str
 61
 62    # Name of the user who submitted the job
 63    username: str
 64
 65    # Job identifier inside the batch system
 66    job_id: str
 67
 68    # Job name
 69    job_name: str
 70
 71    # Name of the script executed
 72    script_name: str
 73
 74    # Queue the job was submitted to
 75    queue: str
 76
 77    # Type of the qq job
 78    job_type: JobType
 79
 80    # Host from which the job was submitted
 81    input_machine: str
 82
 83    # Directory from which the job was submitted
 84    input_dir: Path
 85
 86    # Job state according to qq
 87    job_state: NaiveState
 88
 89    # Job submission timestamp
 90    submission_time: datetime
 91
 92    # Name of the file for storing standard output of the executed script
 93    stdout_file: str
 94
 95    # Name of the file for storing error output of the executed script
 96    stderr_file: str
 97
 98    # Resources allocated to the job
 99    resources: Resources
100
101    # List of files and directories to not copy to the working directory.
102    excluded_files: list[Path] = field(default_factory=list)
103
104    # List of files and directories to explicitly copy to the working directory.
105    included_files: list[Path] = field(default_factory=list)
106
107    # Mode of transferring files from the working directory to the input directory after job completion.
108    transfer_mode: list[TransferMode] = field(default_factory=lambda: [Success()])
109
110    # List of dependencies.
111    depend: list[Depend] = field(default_factory=list)
112
113    # Loop job-associated information.
114    loop_info: LoopInfo | None = None
115
116    # Account associated with the job
117    account: str | None = None
118
119    # Batch server the job was submitted to
120    # Can be `None` which indicates the job was submitted
121    # to the default (main) batch server the input machine is connected to
122    server: str | None = None
123
124    # Hosts from which a loop job or a continuous job should be resubmitted
125    resubmit_from: list[ResubmitHost] = field(default_factory=list)
126
127    # Interpreter to use for running the submitted script
128    interpreter: Interpreter | None = None
129
130    # Job start time
131    start_time: datetime | None = None
132
133    # Main node assigned to the job
134    main_node: str | None = None
135
136    # All nodes assigned to the job
137    all_nodes: list[str] | None = None
138
139    # Working directory
140    work_dir: Path | None = None
141
142    # Job completion time
143    completion_time: datetime | None = None
144
145    # Exit code of qq run
146    job_exit_code: int | None = None
147
148    @classmethod
149    def from_file(cls, file: Path, host: str | None = None) -> Self:
150        """
151        Load an Info instance from a YAML file, either locally or on a remote host.
152
153        If `host` is provided, the file will be read from the remote host using
154        the batch system's `read_remote_file` method. Otherwise, the file is read locally.
155
156        Args:
157            file (Path): Path to the YAML qq info file.
158            host (str | None): Optional hostname of the remote machine where the file resides.
159                If None, the file is assumed to be local.
160
161        Returns:
162            Info: Instance constructed from the file.
163
164        Raises:
165            QQError: If the file does not exist, cannot be reached, cannot be parsed,
166                    or does not contain all mandatory information.
167        """
168        try:
169            if host:
170                # remote file
171                logger.debug(f"Loading qq info from '{file}' on '{host}'.")
172
173                BatchSystem = BatchInterface.from_env_var_or_guess()
174                data: dict[str, object] = yaml.load(
175                    BatchSystem.read_remote_file(host, file),
176                    Loader=SafeLoader,
177                )
178            else:
179                # local file
180                logger.debug(f"Loading qq info from '{file}'.")
181
182                try:
183                    with file.open("r") as input:
184                        data: dict[str, object] = yaml.load(input, Loader=SafeLoader)
185                except FileNotFoundError:
186                    raise QQError(f"qq info file '{file}' does not exist.")
187                except PermissionError:
188                    raise QQError(
189                        f"No permission to read file '{file}' or access its parent directory."
190                    )
191                except IsADirectoryError:
192                    raise QQError(f"Expected a file but path is a directory: {file}.")
193                except UnicodeDecodeError as e:
194                    raise QQError(f"File is not valid UTF-8 text: {file}.") from e
195                except yaml.YAMLError as e:
196                    raise QQError(f"Failed to parse YAML in {file}: {e}.") from e
197
198            return cls._from_dict(data)
199        except yaml.YAMLError as e:
200            raise QQError(f"Could not parse the qq info file '{file}': {e}.") from e
201        except TypeError as e:
202            raise QQError(f"Invalid qq info file '{file}': {e}.") from e
203
204    def to_file(self, file: Path, host: str | None = None) -> None:
205        """
206        Export this Info instance to a YAML file, either locally or on a remote host.
207
208        If `host` is provided, the file will be written to the remote host using
209        the batch system's `write_remote_file` method. Otherwise, the file is written locally.
210
211        Args:
212            file (Path): Path to write the YAML file.
213            host (str | None): Optional hostname of the remote machine where the file should be written.
214                If None, the file is written locally.
215
216        Raises:
217            QQError: If the file cannot be created, reached, or written to.
218        """
219        try:
220            content = "# qq job info file\n" + self._to_yaml() + "\n"
221
222            if host:
223                # remote file
224                logger.debug(f"Exporting qq info into '{file}' on '{host}'.")
225                self.batch_system.write_remote_file(host, file, content)
226            else:
227                # local file
228                logger.debug(f"Exporting qq info into '{file}'.")
229                with file.open("w") as output:
230                    output.write(content)
231        except Exception as e:
232            raise QQError(f"Cannot create or write to file '{file}': {e}") from e
233
234    def _to_yaml(self) -> str:
235        """
236        Serialize the Info instance to a YAML string.
237
238        Returns:
239            str: YAML representation of the Info object.
240        """
241        return yaml.dump(
242            self._to_dict(), default_flow_style=False, sort_keys=False, Dumper=Dumper
243        )
244
245    def _to_dict(self) -> dict[str, object]:
246        """
247        Convert the Info instance into a dictionary of string-object pairs.
248        Fields that are None are ignored.
249
250        Returns:
251            dict[str, object]: Dictionary containing all fields with non-None values,
252            converting enums and nested objects appropriately.
253        """
254        result: dict[str, object] = {}
255
256        for f in fields(self):
257            value = getattr(self, f.name)
258            # ignore None fields
259            if value is None:
260                continue
261
262            # empty lists are ignored
263            if isinstance(value, list) and not value:
264                continue
265
266            # convert job type
267            if f.type == JobType:
268                result[f.name] = str(value)
269            # convert resources
270            elif f.type == Resources or f.type == LoopInfo | None:
271                result[f.name] = value.to_dict()
272            # convert the state and the batch system
273            elif (
274                f.type == NaiveState
275                or f.type == AnyBatchClass
276                or f.type == Path
277                or f.type == Path | None
278            ):
279                result[f.name] = str(value)
280            # convert list of excluded/included files
281            elif f.type == list[Path]:
282                result[f.name] = [str(x) for x in value]
283            # convert transfer modes, resubmit hosts, and dependencies
284            elif (
285                f.type == list[TransferMode]
286                or f.type == list[ResubmitHost]
287                or f.type == list[Depend]
288            ):
289                result[f.name] = [x.to_str() for x in value]
290            # convert interpreter
291            elif f.type == Interpreter or f.type == Interpreter | None:
292                result[f.name] = value.to_dict()
293            # convert timestamp
294            elif f.type == datetime or f.type == datetime | None:
295                result[f.name] = value.strftime(CFG.date_formats.standard)
296            else:
297                result[f.name] = value
298
299        return result
300
301    @classmethod
302    def _from_dict(cls, data: dict[str, object]) -> Self:
303        """
304        Construct an Info instance from a dictionary.
305
306        Args:
307            data (dict[str, object]): Dictionary containing field names and values.
308
309        Returns:
310            Info: An Info instance.
311
312        Raises:
313            TypeError: If required fields are missing.
314        """
315        init_kwargs = {}
316        for f in fields(cls):
317            name = f.name
318            # skip undefined fields
319            if name not in data:
320                continue
321
322            value = data[name]
323
324            # convert job type
325            if f.type == JobType and isinstance(value, str):
326                init_kwargs[name] = JobType.from_str(value)
327            # convert optional loop job info
328            elif f.type == LoopInfo | None and isinstance(value, dict):
329                init_kwargs[name] = LoopInfo.from_dict(value)  # ty: ignore[invalid-argument-type]
330            # convert resources
331            elif f.type == Resources:
332                init_kwargs[name] = Resources(**value)  # ty: ignore[invalid-argument-type]
333            # convert the batch system
334            elif f.type == AnyBatchClass and isinstance(value, str):
335                init_kwargs[name] = BatchInterface.from_str(value)
336            # convert the job state
337            elif f.type == NaiveState and isinstance(value, str):
338                init_kwargs[name] = (
339                    NaiveState.from_str(value) if value else NaiveState.UNKNOWN
340                )
341            # convert paths (incl. optional paths)
342            elif f.type == Path or f.type == Path | None:
343                init_kwargs[name] = Path(value)  # ty: ignore[invalid-argument-type]
344            # convert the list of excluded paths
345            elif f.type == list[Path] and isinstance(value, list):
346                init_kwargs[name] = [
347                    Path(v) if isinstance(v, str) else v for v in value
348                ]
349            # convert transfer modes
350            elif f.type == list[TransferMode] and isinstance(value, list):
351                init_kwargs[name] = [TransferMode.from_str(x) for x in value]  # ty: ignore[invalid-argument-type]
352            # convert dependencies
353            elif f.type == list[Depend] and isinstance(value, list):
354                init_kwargs[name] = [Depend.from_str(x) for x in value]  # ty: ignore[invalid-argument-type]
355            # convert resubmit hosts
356            elif f.type == list[ResubmitHost] and isinstance(value, list):
357                init_kwargs[name] = [ResubmitHost.from_str(x) for x in value]  # ty: ignore[invalid-argument-type]
358            # convert interpreter from string (legacy)
359            elif (f.type == Interpreter or f.type == Interpreter | None) and isinstance(
360                value, str
361            ):
362                init_kwargs[name] = Interpreter.from_str(value)
363            # convert interpreter from dict
364            elif (f.type == Interpreter or f.type == Interpreter | None) and isinstance(
365                value, dict
366            ):
367                init_kwargs[name] = Interpreter.from_dict(value)  # ty: ignore[invalid-argument-type]
368            # convert timestamp
369            elif (f.type == datetime or f.type == datetime | None) and isinstance(
370                value, str
371            ):
372                init_kwargs[name] = datetime.strptime(value, CFG.date_formats.standard)
373            else:
374                init_kwargs[name] = value
375
376        return cls(**init_kwargs)