Skip to content

API reference

Encoding and decoding

qccodec.encode(input_data: Inputs) -> NativeInput

Prepare a complete native invocation for an input without executing it.

Source code in src/qccodec/registry.py
107
108
109
def encode(input_data: Inputs) -> NativeInput:
    """Prepare a complete native invocation for an input without executing it."""
    return get_codec(input_data).encode(input_data)

qccodec.decode(program: str, calctype: CalcType | str, *, stdout: str | None = None, directory: Any = None, input_data: ProgramInput | None = None, failed: bool = False, options: Mapping[str, Any] | None = None, version: str | None = None) -> StructuredData

Decode saved program output into structured data without running anything.

Parameters:

Name Type Description Default
program str

Name of the program that produced the output.

required
calctype CalcType | str

The type of calculation that was run.

required
stdout str | None

The program's captured output.

None
directory Any

The calculation directory holding output files.

None
input_data ProgramInput | None

The input used for the calculation, for parsers that need it.

None
failed bool

Recover available data from a failed calculation.

False
options Mapping[str, Any] | None

Overrides for the codec's declared decode options.

None
version str | None

The program version, when the output does not print it.

None

Returns:

Type Description
StructuredData

The qcdata model matching calctype.

Source code in src/qccodec/registry.py
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
def decode(
    program: str,
    calctype: CalcType | str,
    *,
    stdout: str | None = None,
    directory: Any = None,
    input_data: ProgramInput | None = None,
    failed: bool = False,
    options: Mapping[str, Any] | None = None,
    version: str | None = None,
) -> StructuredData:
    """Decode saved program output into structured data without running anything.

    Args:
        program: Name of the program that produced the output.
        calctype: The type of calculation that was run.
        stdout: The program's captured output.
        directory: The calculation directory holding output files.
        input_data: The input used for the calculation, for parsers that need it.
        failed: Recover available data from a failed calculation.
        options: Overrides for the codec's declared decode options.
        version: The program version, when the output does not print it.

    Returns:
        The qcdata model matching ``calctype``.
    """
    return get_codec(program).parse(
        calctype,
        stdout=stdout,
        directory=directory,
        input_data=input_data,
        failed=failed,
        options=options,
        version=version,
    )

Codecs

qccodec.get_codec(program: str | Inputs) -> Codec

Resolve the codec for a program name or input object.

Resolution order: codecs passed to register, built-ins, then codecs installed by other packages under the qccodec.codecs entry point group. A FileInput resolves to the pass-through file codec.

Parameters:

Name Type Description Default
program str | Inputs

A program name, or an input whose program should be resolved.

required

Raises:

Type Description
CodecNotFoundError

No codec is available for the program.

Source code in src/qccodec/registry.py
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def get_codec(program: str | Inputs) -> Codec:
    """Resolve the codec for a program name or input object.

    Resolution order: codecs passed to ``register``, built-ins, then codecs installed
    by other packages under the ``qccodec.codecs`` entry point group. A ``FileInput``
    resolves to the pass-through file codec.

    Args:
        program: A program name, or an input whose program should be resolved.

    Raises:
        CodecNotFoundError: No codec is available for the program.
    """
    if isinstance(program, ProgramInput):
        name = program.program
    elif isinstance(program, FileInput):
        from .programs.file import codec as file_codec

        return file_codec
    else:
        name = program

    codec = _REGISTERED.get(name)
    if codec is None and name in _BUILTIN:
        module = import_module(f"qccodec.programs.{_BUILTIN[name]}")
        codec = module.codecs[name] if hasattr(module, "codecs") else module.codec
    if codec is None:
        codec = _plugin(name)
    if codec is None:
        raise CodecNotFoundError(f"No codec found for program '{name}'.")
    return codec

qccodec.register(codec: Codec) -> Codec

Make a codec resolvable by its program name for this process.

A registered codec takes precedence over a built-in with the same name, so scripts can both add new programs and override shipped integrations. The codec is returned so this can be used as a decorator-style one-liner.

Source code in src/qccodec/registry.py
38
39
40
41
42
43
44
45
46
def register(codec: Codec) -> Codec:
    """Make a codec resolvable by its program name for this process.

    A registered codec takes precedence over a built-in with the same name, so
    scripts can both add new programs and override shipped integrations. The
    codec is returned so this can be used as a decorator-style one-liner.
    """
    _REGISTERED[codec.program] = codec
    return codec

qccodec.unregister(program: str) -> None

Remove a codec registered with register; built-ins are unaffected.

Source code in src/qccodec/registry.py
49
50
51
def unregister(program: str) -> None:
    """Remove a codec registered with ``register``; built-ins are unaffected."""
    _REGISTERED.pop(program, None)

qccodec.supported_programs() -> tuple[str, ...]

Built-in and installed plugin program names plus any registered in this process.

Source code in src/qccodec/registry.py
54
55
56
57
def supported_programs() -> tuple[str, ...]:
    """Built-in and installed plugin program names plus any registered in this process."""
    plugins = [ep.name for ep in entry_points(group=ENTRY_POINT_GROUP)]
    return tuple(dict.fromkeys([*_BUILTIN, *plugins, *_REGISTERED]))

qccodec.Codec(program: str, calctypes: Collection[CalcType | str], *, options: Mapping[str, Any] | None = None, encode: Encoder | None = None, decode: Decoder | None = None, failure: Failure | None = None, propagate_wfn: PropagateWfn | None = None, version_key: VersionKey | None = None, parsers: Iterable[ParserSpec] = ())

Program-specific translation between qcdata objects and native inputs/outputs.

A codec encodes a ProgramInput into a SubprocessInput or PythonInput and decodes the native output back into a qcdata *Data object. File-based programs register small parsers with @codec.parser(...); Python-native programs supply a decode function instead.

Parameters:

Name Type Description Default
program str

Name matched against ProgramInput.program.

required
calctypes Collection[CalcType | str]

Calculation types this codec can encode.

required
options Mapping[str, Any] | None

Decode options and their defaults, e.g. what to collect. Values passed to decode/parse are validated against these keys and reach parsers as Context.options.

None
encode Encoder | None

Translate an input into a native invocation. See encoder.

None
decode Decoder | None

Translate native output into data. See decoder.

None
failure Failure | None

Detect a failure the program reported. See failure.

None
propagate_wfn PropagateWfn | None

(output, input_data) -> ProgramInput placing the wavefunction from a previous output onto the next input as a guess.

None
version_key VersionKey | None

Turns the program's version string into a PEP 440 version for matching parser versions. Not needed when the program already prints versions like 6.1.0.

None
parsers Iterable[ParserSpec]

Pre-built parser specifications.

()
Source code in src/qccodec/codec.py
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
def __init__(
    self,
    program: str,
    calctypes: Collection[CalcType | str],
    *,
    options: Mapping[str, Any] | None = None,
    encode: Encoder | None = None,
    decode: Decoder | None = None,
    failure: Failure | None = None,
    propagate_wfn: PropagateWfn | None = None,
    version_key: VersionKey | None = None,
    parsers: Iterable[ParserSpec] = (),
) -> None:
    """Create a codec; hooks may also be attached later with decorators."""
    self.program: str = program
    self.calctypes: frozenset[CalcType] = frozenset(CalcType(c) for c in calctypes)
    self.options: dict[str, Any] = dict(options or {})
    self.propagate_wfn: PropagateWfn | None = propagate_wfn
    self.version_key: VersionKey | None = version_key
    self.parsers: list[ParserSpec] = list(parsers)
    self._encode: Encoder | None = encode
    self._decode: Decoder | None = None
    self._failure: Failure | None = failure
    if decode is not None:
        self.decoder(decode)

encoder(func: F) -> F

Register func(input_data) -> SubprocessInput | PythonInput.

Source code in src/qccodec/codec.py
224
225
226
227
def encoder(self, func: F) -> F:
    """Register ``func(input_data) -> SubprocessInput | PythonInput``."""
    self._encode = func
    return func

decoder(func: F) -> F

Register func(input_data, native_output) -> Data in place of parsers.

Raises:

Type Description
RegistryError

The codec has parsers or options, which only the parser pipeline uses.

Source code in src/qccodec/codec.py
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
def decoder(self, func: F) -> F:
    """Register ``func(input_data, native_output) -> Data`` in place of parsers.

    Raises:
        RegistryError: The codec has parsers or options, which only the parser
            pipeline uses.
    """
    if self.parsers:
        raise RegistryError(
            f"Codec {self.program!r} has parsers; a decoder would replace them."
        )
    if self.options:
        raise RegistryError(
            f"Codec {self.program!r} declares options, which only parsers receive."
        )
    self._decode = func
    return func

failure(func: F) -> F

Register func(ctx) -> str | None returning a message if the program failed.

A message makes decoding recover partial data and then raise CalculationError with that data attached.

Source code in src/qccodec/codec.py
247
248
249
250
251
252
253
254
def failure(self, func: F) -> F:
    """Register ``func(ctx) -> str | None`` returning a message if the program failed.

    A message makes decoding recover partial data and then raise
    ``CalculationError`` with that data attached.
    """
    self._failure = func
    return func

parser(*, target: Target | None = None, calctypes: Collection[CalcType | str] | None = None, file: FileRef = STDOUT, required: bool = True, versions: str | None = None) -> Callable[[F], F]

Register a parser function.

Parameters:

Name Type Description Default
target Target | None

Result field to fill, or a tuple for a nested field. Directory parsers may omit it and return a mapping merged into the result.

None
calctypes Collection[CalcType | str] | None

Calculation types this parser applies to; all when omitted.

None
file FileRef

"stdout" (default), "directory" (the parser receives a Context), a file name/glob relative to the directory, or a function of the Context returning such a name.

STDOUT
required bool

If True, a missing file or MatchNotFoundError fails decoding unless partial data is being recovered.

True
versions str | None

Program versions this parser applies to, as a PEP 440 specifier such as ">=6.0" or ">=5.0,<6.0"; every version when omitted. Parsers for the same target may coexist when each declares versions.

None

Returns:

Type Description
Callable[[F], F]

The original function, unchanged, so it can be called and tested directly.

Source code in src/qccodec/codec.py
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
def parser(
    self,
    *,
    target: Target | None = None,
    calctypes: Collection[CalcType | str] | None = None,
    file: FileRef = STDOUT,
    required: bool = True,
    versions: str | None = None,
) -> Callable[[F], F]:
    """Register a parser function.

    Args:
        target: Result field to fill, or a tuple for a nested field. Directory
            parsers may omit it and return a mapping merged into the result.
        calctypes: Calculation types this parser applies to; all when omitted.
        file: ``"stdout"`` (default), ``"directory"`` (the parser receives a
            ``Context``), a file name/glob relative to the directory, or a
            function of the ``Context`` returning such a name.
        required: If True, a missing file or ``MatchNotFoundError`` fails
            decoding unless partial data is being recovered.
        versions: Program versions this parser applies to, as a PEP 440
            specifier such as ``">=6.0"`` or ``">=5.0,<6.0"``; every version
            when omitted. Parsers for the same target may coexist when each
            declares ``versions``.

    Returns:
        The original function, unchanged, so it can be called and tested directly.
    """
    types = (
        frozenset(CalcType)
        if calctypes is None
        else frozenset(CalcType(c) for c in calctypes)
    )
    try:
        specifier = SpecifierSet(versions) if versions is not None else None
    except InvalidSpecifier as exc:
        raise RegistryError(f"Invalid versions specifier {versions!r}.") from exc

    def decorate(func: F) -> F:
        """Validate and record the parser specification."""
        if self._decode is not None:
            raise RegistryError(
                f"Codec {self.program!r} has a decoder; parsers would never run."
            )
        if target is None and file != DIRECTORY:
            raise RegistryError(
                f"Parser '{func.__name__}' must have a target unless file='directory'."
            )
        if target is not None and not (
            isinstance(target, str)
            or (
                isinstance(target, tuple)
                and all(isinstance(k, str) for k in target)
            )
        ):
            raise RegistryError(
                f"Target {target!r} must be a string or tuple of strings."
            )
        if specifier is not None and target == VERSION_TARGET:
            raise RegistryError(
                "The program version parser runs on every version; "
                "it cannot declare versions."
            )
        for existing in self.parsers:
            if (
                target is not None
                and existing.target == target
                and types & existing.calctypes
                and (specifier is None or existing.versions is None)
            ):
                raise RegistryError(
                    f"Duplicate parser target {target!r} for overlapping "
                    "calculation types. Declare versions on both parsers if "
                    "they handle different program versions."
                )
        self.parsers.append(
            ParserSpec(func, file, types, required, target, specifier)
        )
        return func

    return decorate

encode(input_data: Inputs) -> NativeInput

Translate an input into a native invocation without executing it.

Raises:

Type Description
EncoderError

The codec has no encoder, the input's program or calculation type is not supported, or the encoder returned something other than a SubprocessInput or PythonInput.

Source code in src/qccodec/codec.py
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
def encode(self, input_data: Inputs) -> NativeInput:
    """Translate an input into a native invocation without executing it.

    Raises:
        EncoderError: The codec has no encoder, the input's program or
            calculation type is not supported, or the encoder returned
            something other than a ``SubprocessInput`` or ``PythonInput``.
    """
    if self._encode is None:
        raise EncoderError(f"Codec {self.program!r} has no encoder.")
    if isinstance(input_data, ProgramInput):
        if input_data.program != self.program:
            raise EncoderError(
                f"Codec {self.program!r} cannot encode {input_data.program!r}."
            )
        if input_data.calctype not in self.calctypes:
            raise EncoderError(
                f"{self.program} does not support {input_data.calctype.value}."
            )
    native = self._encode(input_data)
    if not isinstance(native, (SubprocessInput, PythonInput)):
        raise EncoderError(
            f"{self.program} encoder must return SubprocessInput or PythonInput, "
            f"not {type(native).__name__}."
        )
    return native

decode(input_data: Inputs, native_output: NativeOutput, options: Mapping[str, Any] | None = None) -> Data

Interpret native output, returning data or raising with partial data.

When the run failed (for example a nonzero exit status), whatever can be parsed is recovered, and if the codec's failure hook explains the failure CalculationError carries that explanation and the recovered data.

Parameters:

Name Type Description Default
input_data Inputs

The input that produced the output.

required
native_output NativeOutput

What the runner observed.

required
options Mapping[str, Any] | None

Overrides for this codec's declared options.

None

Raises:

Type Description
CalculationError

The program failed and the failure hook says why.

ParserError

A required output is missing or malformed in a run that did not fail.

Source code in src/qccodec/codec.py
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
def decode(
    self,
    input_data: Inputs,
    native_output: NativeOutput,
    options: Mapping[str, Any] | None = None,
) -> Data:
    """Interpret native output, returning data or raising with partial data.

    When the run failed (for example a nonzero exit status), whatever can be
    parsed is recovered, and if the codec's failure hook explains the failure
    ``CalculationError`` carries that explanation and the recovered data.

    Args:
        input_data: The input that produced the output.
        native_output: What the runner observed.
        options: Overrides for this codec's declared ``options``.

    Raises:
        CalculationError: The program failed and the failure hook says why.
        ParserError: A required output is missing or malformed in a run that
            did not fail.
    """
    if self._decode is not None:
        self._merge_options(options)  # rejects unknown keys
        return self._decode(input_data, native_output)
    if not isinstance(input_data, ProgramInput):
        raise ParserError(f"Codec {self.program!r} has no decoder for file inputs.")
    return self.parse(
        input_data.calctype,
        stdout=native_output.stdout,
        directory=native_output.directory,
        input_data=input_data,
        failed=native_output.failed,
        options=options,
        report_failure=True,
    )

parse(calctype: CalcType | str, *, stdout: str | None = None, directory: str | Path | None = None, input_data: ProgramInput | None = None, failed: bool = False, options: Mapping[str, Any] | None = None, version: str | None = None, as_dict: bool = False, check_failure: bool = True, report_failure: bool | None = None) -> StructuredData | dict[str, Any]

Run this codec's parsers over saved artifacts without running the program.

Parameters:

Name Type Description Default
calctype CalcType | str

The type of calculation that produced the artifacts.

required
stdout str | None

The program's captured output.

None
directory str | Path | None

The calculation directory holding output files.

None
input_data ProgramInput | None

The input used for the calculation, for parsers that need it.

None
failed bool

Recover what is available from a failed calculation. Missing files and values are tolerated; malformed values still fail.

False
options Mapping[str, Any] | None

Overrides for this codec's declared options.

None
version str | None

The program version, when the output does not print it or should be overridden. Otherwise the version parser supplies it.

None
as_dict bool

Return the collected values instead of a validated model.

False
check_failure bool

Run the codec's failure hook first. Nested decodes of partial logs turn this off.

True
report_failure bool | None

Raise CalculationError when the failure hook explains a failure. Defaults to not failed, so recovery mode returns partial data quietly; decode passes True so a program that exited with an error still reports why.

None

Returns:

Type Description
StructuredData | dict[str, Any]

The qcdata model for calctype, or a dictionary when as_dict.

Raises:

Type Description
ParserError

A required file is missing or a value is malformed.

MatchNotFoundError

A required parser found no match.

CalculationError

The failure hook explained a failure and report_failure is true. Recovered data is attached to every exception as .data.

Source code in src/qccodec/codec.py
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
def parse(
    self,
    calctype: CalcType | str,
    *,
    stdout: str | None = None,
    directory: str | Path | None = None,
    input_data: ProgramInput | None = None,
    failed: bool = False,
    options: Mapping[str, Any] | None = None,
    version: str | None = None,
    as_dict: bool = False,
    check_failure: bool = True,
    report_failure: bool | None = None,
) -> StructuredData | dict[str, Any]:
    """Run this codec's parsers over saved artifacts without running the program.

    Args:
        calctype: The type of calculation that produced the artifacts.
        stdout: The program's captured output.
        directory: The calculation directory holding output files.
        input_data: The input used for the calculation, for parsers that need it.
        failed: Recover what is available from a failed calculation. Missing
            files and values are tolerated; malformed values still fail.
        options: Overrides for this codec's declared ``options``.
        version: The program version, when the output does not print it or
            should be overridden. Otherwise the version parser supplies it.
        as_dict: Return the collected values instead of a validated model.
        check_failure: Run the codec's failure hook first. Nested decodes of
            partial logs turn this off.
        report_failure: Raise ``CalculationError`` when the failure hook
            explains a failure. Defaults to ``not failed``, so recovery mode
            returns partial data quietly; ``decode`` passes True so a program
            that exited with an error still reports why.

    Returns:
        The qcdata model for ``calctype``, or a dictionary when ``as_dict``.

    Raises:
        ParserError: A required file is missing or a value is malformed.
        MatchNotFoundError: A required parser found no match.
        CalculationError: The failure hook explained a failure and
            ``report_failure`` is true. Recovered data is attached to every
            exception as ``.data``.
    """
    ctx = Context(
        codec=self,
        calctype=CalcType(calctype),
        stdout=stdout,
        directory=Path(directory) if directory is not None else None,
        input_data=input_data,
        failed=failed,
        options=self._merge_options(options),
    )
    data_type = get_data_type(ctx.calctype)
    message = (
        self._failure(ctx) if self._failure is not None and check_failure else None
    )
    if message:
        ctx.failed = True
    collected: dict[str, Any] = {"provenance": {"program": self.program}}
    not_found: MatchNotFoundError | None = None
    missing: list[str] = []
    # With an unknown version, version-gated alternatives for one target are
    # tried in order; a miss only counts if no alternative fills the target.
    deferred: dict[object, MatchNotFoundError | str] = {}
    filled: set[object] = set()

    def partial() -> StructuredData:
        """Validate whatever has been collected so far."""
        return data_type(**collected)

    specs = [s for s in self.parsers if ctx.calctype in s.calctypes]
    if version is not None:
        ctx.version = version
        _assign(collected, VERSION_TARGET, version)
        specs = [s for s in specs if s.target != VERSION_TARGET]
    else:
        # The version parser runs first so later parsers can be filtered by it.
        specs.sort(key=lambda s: s.target != VERSION_TARGET)
    for spec in specs:
        name = spec.parser.__name__
        key: object = spec.target if spec.target is not None else name
        gated = spec.versions is not None
        matched = self._version_matches(spec, ctx.version)
        if matched is False:
            continue
        if matched is None and key in filled:
            continue  # an earlier alternative already handled this target
        try:
            value: Any = spec.parser(ctx.contents(spec))
        except FileNotFoundError as exc:
            logger.info("Parser %r: missing %s", name, exc.filename or exc)
            if spec.required and not ctx.failed:
                if matched is None and gated:
                    deferred.setdefault(key, str(exc.filename or exc))
                else:
                    missing.append(str(exc.filename or exc))
            continue
        except MatchNotFoundError as exc:
            logger.info("Parser %r found no match", name)
            if spec.required and not ctx.failed:
                if matched is None and gated:
                    deferred.setdefault(key, exc)
                elif not_found is None:
                    not_found = exc
            continue
        except ParserError as exc:
            exc.data = partial()
            raise
        if spec.target is None:
            for item_key, item in value.items():
                _assign(collected, item_key, item)
        else:
            _assign(collected, spec.target, value)
        if spec.target == VERSION_TARGET:
            ctx.version = value
        filled.add(key)
        deferred.pop(key, None)
        logger.debug("Parser %r filled %r", name, spec.target)

    for problem in deferred.values():
        if isinstance(problem, MatchNotFoundError):
            not_found = not_found or problem
        else:
            missing.append(problem)

    if not_found is not None:
        not_found.data = partial()
        raise not_found
    if missing:
        raise ParserError(
            "Missing required output artifact(s): "
            + ", ".join(sorted(set(missing))),
            data=partial(),
        )
    if report_failure is None:
        report_failure = not failed
    if message and report_failure:
        raise CalculationError(message, data=partial())
    return collected if as_dict else partial()

qccodec.Context(codec: Codec, calctype: CalcType, stdout: str | None = None, directory: Path | None = None, input_data: ProgramInput | None = None, failed: bool = False, options: dict[str, Any] = dict(), version: str | None = None) dataclass

Everything a parser may need beyond the contents of one file.

Attributes:

Name Type Description
codec Codec

The codec running the decode; used for nested decoding.

calctype CalcType

The calculation type being decoded.

stdout str | None

Captured program output, if available.

directory Path | None

The calculation directory, if available.

input_data ProgramInput | None

The input used for the calculation, if available.

failed bool

Whether partial data is being recovered from a failed run.

options dict[str, Any]

The codec's options merged with those passed for this decode.

version str | None

The program version string, once known.

path(pattern: str) -> Path

Resolve a file name or glob pattern to exactly one path in the directory.

Raises:

Type Description
FileNotFoundError

No directory is available or nothing matches.

ParserError

The pattern matches more than one file.

Source code in src/qccodec/codec.py
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
def path(self, pattern: str) -> Path:
    """Resolve a file name or glob pattern to exactly one path in the directory.

    Raises:
        FileNotFoundError: No directory is available or nothing matches.
        ParserError: The pattern matches more than one file.
    """
    if self.directory is None:
        raise FileNotFoundError(pattern)
    matches = sorted(p for p in self.directory.glob(pattern) if p.is_file())
    if not matches:
        raise FileNotFoundError(pattern)
    if len(matches) > 1:
        names = ", ".join(str(p.relative_to(self.directory)) for p in matches)
        raise ParserError(f"Pattern {pattern!r} matched several files: {names}")
    return matches[0]

read(pattern: str) -> str

Read the text of the single file matching a name or glob pattern.

Source code in src/qccodec/codec.py
118
119
120
def read(self, pattern: str) -> str:
    """Read the text of the single file matching a name or glob pattern."""
    return self.path(pattern).read_text()

contents(spec: ParserSpec) -> Any

Return what a parser receives: stdout, this context, or a file's text.

Raises:

Type Description
FileNotFoundError

The parser's input is not available.

Source code in src/qccodec/codec.py
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
def contents(self, spec: ParserSpec) -> Any:
    """Return what a parser receives: stdout, this context, or a file's text.

    Raises:
        FileNotFoundError: The parser's input is not available.
    """
    if spec.file == STDOUT:
        if self.stdout is None:
            raise FileNotFoundError(STDOUT)
        return self.stdout
    if spec.file == DIRECTORY:
        if self.directory is None:
            raise FileNotFoundError(DIRECTORY)
        return self
    return self.read(spec.file(self) if callable(spec.file) else spec.file)

decode(calctype: CalcType | str, *, stdout: str | None = None, directory: str | Path | None = None) -> StructuredData

Parse another set of artifacts with this codec, e.g. one trajectory step.

The failure hook is not run again: the whole run was already checked before this parser was called, and partial logs such as one optimization step would not pass a whole-run check.

Source code in src/qccodec/codec.py
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
def decode(
    self,
    calctype: CalcType | str,
    *,
    stdout: str | None = None,
    directory: str | Path | None = None,
) -> StructuredData:
    """Parse another set of artifacts with this codec, e.g. one trajectory step.

    The failure hook is not run again: the whole run was already checked before
    this parser was called, and partial logs such as one optimization step
    would not pass a whole-run check.
    """
    return self.codec.parse(
        calctype,
        stdout=stdout,
        directory=directory,
        input_data=self.input_data,
        failed=self.failed,
        options=self.options,
        version=self.version,
        check_failure=False,
    )

qccodec.ParserSpec(parser: Callable[..., Any], file: FileRef, calctypes: frozenset[CalcType], required: bool, target: Target | None, versions: SpecifierSet | None = None) dataclass

A registered parser and the metadata that decides when it runs.

Attributes:

Name Type Description
parser Callable[..., Any]

The parser function. It receives file contents, or a Context when file is "directory".

file FileRef

"stdout", "directory", a file name or glob pattern relative to the calculation directory, or a function of the Context that returns such a name.

calctypes frozenset[CalcType]

Calculation types for which this parser runs.

required bool

Whether a missing file or value fails decoding.

target Target | None

Where the parsed value is stored on the result. A tuple creates nested fields. None is permitted only for directory parsers, which then return a mapping merged into the result.

versions SpecifierSet | None

Program versions this parser applies to, as a PEP 440 specifier set; None means every version.

Native inputs and outputs

qccodec.SubprocessInput(command: tuple[str, ...], files: Mapping[str, str | bytes] = dict(), wavefunction_files: tuple[str, ...] = (), environment: Mapping[str, str] = dict()) dataclass

A command and the files it needs, to run inside a calculation directory.

Attributes:

Name Type Description
command tuple[str, ...]

The argument vector; the executable is the first element.

files Mapping[str, str | bytes]

Files written to the calculation directory before running.

wavefunction_files tuple[str, ...]

Relative glob patterns of restart files worth keeping.

environment Mapping[str, str]

Environment defaults required by the program.

qccodec.PythonInput(function: Callable[..., Any], args: tuple[Any, ...] = (), kwargs: Mapping[str, Any] = dict(), files: Mapping[str, str | bytes] = dict(), child_callback: str | None = None, loggers: tuple[str, ...] = (), wavefunction_files: tuple[str, ...] = (), environment: Mapping[str, str] = dict()) dataclass

A Python callable with its arguments, to run inside a calculation directory.

Attributes:

Name Type Description
function Callable[..., Any]

The callable to invoke.

args tuple[Any, ...]

Positional arguments.

kwargs Mapping[str, Any]

Keyword arguments.

files Mapping[str, str | bytes]

Files written to the calculation directory before running.

child_callback str | None

Keyword name into which the runner injects a function that computes child calculations synchronously; only for integrations such as optimizers that drive other programs.

loggers tuple[str, ...]

Python logger names whose output should be captured.

wavefunction_files tuple[str, ...]

Relative glob patterns of restart files worth keeping.

environment Mapping[str, str]

Environment defaults required by the library.

qccodec.SubprocessOutput(stdout: str = '', directory: Path | None = None, returncode: int | None = None, error: Exception | None = None, traceback: str | None = None) dataclass

What a subprocess produced; stdout holds merged stdout and stderr.

failed: bool property

Whether launching or running the process failed.

qccodec.PythonOutput(value: Any = None, stdout: str = '', directory: Path | None = None, error: Exception | None = None, traceback: str | None = None) dataclass

What a Python callable produced: its return value, merged logs, or an error.

failed: bool property

Whether invoking the callable raised an exception.

Parsing helpers

Return the first match of regex in contents.

Raises:

Type Description
MatchNotFoundError

No match is found.

Source code in src/qccodec/utils.py
10
11
12
13
14
15
16
17
18
19
def re_search(regex: str, contents: str, flags: int = 0) -> re.Match[str]:
    """Return the first match of ``regex`` in ``contents``.

    Raises:
        MatchNotFoundError: No match is found.
    """
    match = re.search(regex, contents, flags)
    if not match:
        raise MatchNotFoundError(regex, contents)
    return match

qccodec.re_finditer(regex: str, contents: str, flags: int = 0) -> list[re.Match[str]]

Return all matches of regex in contents.

Raises:

Type Description
MatchNotFoundError

No match is found.

Source code in src/qccodec/utils.py
22
23
24
25
26
27
28
29
30
31
def re_finditer(regex: str, contents: str, flags: int = 0) -> list[re.Match[str]]:
    """Return all matches of ``regex`` in ``contents``.

    Raises:
        MatchNotFoundError: No match is found.
    """
    matches = list(re.finditer(regex, contents, flags))
    if not matches:
        raise MatchNotFoundError(regex, contents)
    return matches

Exceptions

Exceptions raised by codecs, parsers, and the registry.

BaseError

Base class for qccodec exceptions.

DecoderError

Raised when native output cannot be decoded.

EncoderError

Raised when an input cannot be encoded for a program.

ParserError(message: str, data: StructuredData | None = None)

Base class for parser failures; data holds any recovered results.

Source code in src/qccodec/exceptions.py
40
41
42
43
def __init__(self, message: str, data: StructuredData | None = None) -> None:
    """Retain the original failure and available calculation diagnostics."""
    super().__init__(message)
    self.data: StructuredData | None = data

MatchNotFoundError(regex: str, contents: str)

Raised when a parser finds no match for its pattern.

Source code in src/qccodec/exceptions.py
49
50
51
52
53
54
55
def __init__(self, regex: str, contents: str) -> None:
    """Retain the original failure and available calculation diagnostics."""
    self.regex: str = regex
    self.contents: str = contents
    super().__init__(
        f"Could not locate match for regex: '{regex}' in contents: '{contents}'"
    )

RegistryError

Raised when a codec or parser registration is invalid.

DataCollectorError

Raised when two parsers fill the same result field.

CodecNotFoundError

No built-in integration is available for this name.

CalculationError(message: str, data: StructuredData | None = None)

The program reported failure; data holds any recovered scientific results.

Source code in src/qccodec/exceptions.py
40
41
42
43
def __init__(self, message: str, data: StructuredData | None = None) -> None:
    """Retain the original failure and available calculation diagnostics."""
    super().__init__(message)
    self.data: StructuredData | None = data

DependencyNotFoundError

A codec explicitly reports an unavailable execution dependency.

Testing codecs

Test helpers for codecs: fixture cases, a case runner, and a codec contract check.

A case is a directory holding one program run and what a codec must make of it::

tests/data/<program>/<case>/
    input.json      ProgramInput used for the run (program, calctype, structure, ...)
    expected.json   the qcdata *Data object the codec must produce
    stdout          captured program output (optional)
    <files>         output files at their real relative paths, e.g. scr.geom/optim.xyz
    case.json       optional settings: {"options": {...}, "failed": true,
                    "raises": "MatchNotFoundError", "directory": true}
    encoded/        optional: command.txt (one argument per line) and the native
                    input files codec.encode(input) must generate

cases(root) discovers every such directory below root; run_case checks encoding, the full decode, and each registered parser on its own file against the matching slice of expected.json; check_codec verifies the structural contract every codec must satisfy, with or without fixtures.

Typical test module::

import pytest
from qccodec import get_codec
from qccodec.testing import cases, run_case

@pytest.mark.parametrize("case", cases("tests/data"), ids=str)
def test_case(case):
    run_case(get_codec(case.input_data.program), case)

Case(path: Path, name: str, input_data: ProgramInput, expected: StructuredData | None, stdout: str | None, directory: Path | None, options: dict[str, Any] = dict(), failed: bool = False, raises: str | None = None, encoded: Path | None = None) dataclass

One fixture directory, loaded.

Attributes:

Name Type Description
path Path

The case directory; output files live here.

name str

<program>/<case> for test ids.

input_data ProgramInput

The input that produced the run.

expected StructuredData | None

The data the codec must produce, or the partial data attached to the expected exception. None when no expected.json exists.

stdout str | None

Captured output, if the case has a stdout file.

directory Path | None

path when output files should be visible to parsers.

options dict[str, Any]

Decode options passed to the codec.

failed bool

Decode in recovery mode.

raises str | None

Name of the exception class the decode must raise, if any.

encoded Path | None

The encoded/ directory with expected native inputs, if present.

load_case(path: str | Path) -> Case

Load one case directory.

Source code in src/qccodec/testing.py
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
def load_case(path: str | Path) -> Case:
    """Load one case directory."""
    path = Path(path)
    input_data = ProgramInput.model_validate_json((path / "input.json").read_text())
    settings: dict[str, Any] = {}
    if (path / "case.json").exists():
        settings = json.loads((path / "case.json").read_text())
    expected: StructuredData | None = None
    if (path / "expected.json").exists():
        data_type = get_data_type(input_data.calctype)
        expected = data_type.model_validate_json((path / "expected.json").read_text())
    stdout = (path / "stdout").read_text() if (path / "stdout").exists() else None
    has_files = any(p.name not in RESERVED for p in path.iterdir())
    use_directory = settings.get("directory", has_files)
    encoded = path / "encoded"
    return Case(
        path=path,
        name=f"{path.parent.name}/{path.name}",
        input_data=input_data,
        expected=expected,
        stdout=stdout,
        directory=path if use_directory else None,
        options=dict(settings.get("options", {})),
        failed=bool(settings.get("failed", False)),
        raises=settings.get("raises"),
        encoded=encoded if encoded.is_dir() else None,
    )

cases(root: str | Path) -> list[Case]

Discover every case directory below root (any directory holding input.json).

Source code in src/qccodec/testing.py
121
122
123
124
125
def cases(root: str | Path) -> list[Case]:
    """Discover every case directory below ``root`` (any directory holding input.json)."""
    root = Path(root)
    found = sorted(p.parent for p in root.rglob("input.json"))
    return [load_case(p) for p in found]

assert_data_equal(actual: Any, expected: Any, *, rtol: float = 1e-09, atol: float = 0.0, where: str = 'data') -> None

Compare qcdata objects, arrays, or containers structurally.

Numbers compare within rtol/atol; execution metadata and qcdata_version are ignored because they vary between runs.

Raises:

Type Description
AssertionError

Describing the first difference found.

Source code in src/qccodec/testing.py
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
def assert_data_equal(
    actual: Any,
    expected: Any,
    *,
    rtol: float = 1e-9,
    atol: float = 0.0,
    where: str = "data",
) -> None:
    """Compare qcdata objects, arrays, or containers structurally.

    Numbers compare within ``rtol``/``atol``; ``execution`` metadata and
    ``qcdata_version`` are ignored because they vary between runs.

    Raises:
        AssertionError: Describing the first difference found.
    """
    problem = _diff(_normalize(actual), _normalize(expected), where, rtol, atol)
    if problem:
        raise AssertionError(problem)

check_encoding(codec: Codec, case: Case, *, rtol: float = 1e-09) -> None

Compare codec.encode(case.input_data) against the case's encoded/ files.

Source code in src/qccodec/testing.py
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
def check_encoding(codec: Codec, case: Case, *, rtol: float = 1e-9) -> None:
    """Compare ``codec.encode(case.input_data)`` against the case's ``encoded/`` files."""
    assert case.encoded is not None, f"{case}: no encoded/ directory"
    native = codec.encode(case.input_data)
    command_file = case.encoded / "command.txt"
    if command_file.exists():
        assert isinstance(native, SubprocessInput), f"{case}: expected a subprocess"
        expected_command = command_file.read_text().splitlines()
        assert list(native.command) == expected_command, (
            f"{case}: command {list(native.command)!r} != {expected_command!r}"
        )
    for path in sorted(p for p in case.encoded.rglob("*") if p.is_file()):
        name = path.relative_to(case.encoded).as_posix()
        if name == "command.txt":
            continue
        assert name in native.files, f"{case}: encode() did not generate {name!r}"
        produced = native.files[name]
        expected_bytes = path.read_bytes()
        if isinstance(produced, bytes):
            assert produced == expected_bytes, f"{case}: {name} differs"
        else:
            expected_text = expected_bytes.decode()
            assert _unversioned(produced) == _unversioned(expected_text), (
                f"{case}: {name} differs:\n--- expected\n{expected_text}"
                f"--- produced\n{produced}"
            )

check_parsers(codec: Codec, case: Case, *, rtol: float = 1e-09, atol: float = 0.0) -> None

Call every applicable parser directly and compare with its expected slice.

Source code in src/qccodec/testing.py
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
def check_parsers(
    codec: Codec, case: Case, *, rtol: float = 1e-9, atol: float = 0.0
) -> None:
    """Call every applicable parser directly and compare with its expected slice."""
    assert case.expected is not None
    expected = _normalize(case.expected)
    ctx = Context(
        codec=codec,
        calctype=case.input_data.calctype,
        stdout=case.stdout,
        directory=case.directory,
        input_data=case.input_data,
        failed=case.failed,
        options={**codec.options, **case.options},
    )
    for spec in codec.parsers:
        if ctx.calctype not in spec.calctypes:
            continue
        try:
            contents = ctx.contents(spec)
        except FileNotFoundError:
            continue
        where = f"{case} parser {spec.parser.__name__}"
        try:
            value = spec.parser(contents)
        except MatchNotFoundError:
            assert not spec.required or case.failed, (
                f"{where}: required parser found no match"
            )
            if spec.target is not None:
                found = _expected_slice(expected, spec.target)
                assert found in (None, [], {}), (
                    f"{where}: no match, but expected.json holds {found!r}"
                )
            continue
        if spec.target is None:
            for key, item in value.items():
                assert_data_equal(
                    item,
                    expected.get(key),
                    rtol=rtol,
                    atol=atol,
                    where=f"{where}.{key}",
                )
        else:
            assert_data_equal(
                value,
                _expected_slice(expected, spec.target),
                rtol=rtol,
                atol=atol,
                where=where,
            )

run_case(codec: Codec, case: Case, *, rtol: float = 1e-09, atol: float = 0.0) -> None

Check a codec against one fixture case.

Verifies, in order: the encoded native inputs when encoded/ exists; the full decode, which must produce expected.json or raise the exception named in case.json with expected.json attached as partial data; and each registered parser applied directly to its own file.

Raises:

Type Description
AssertionError

With a description of the first mismatch.

Source code in src/qccodec/testing.py
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
def run_case(
    codec: Codec, case: Case, *, rtol: float = 1e-9, atol: float = 0.0
) -> None:
    """Check a codec against one fixture case.

    Verifies, in order: the encoded native inputs when ``encoded/`` exists; the full
    decode, which must produce ``expected.json`` or raise the exception named in
    ``case.json`` with ``expected.json`` attached as partial data; and each registered
    parser applied directly to its own file.

    Raises:
        AssertionError: With a description of the first mismatch.
    """
    if case.encoded is not None:
        check_encoding(codec, case, rtol=rtol)
    kwargs: dict[str, Any] = dict(
        stdout=case.stdout,
        directory=case.directory,
        input_data=case.input_data,
        failed=case.failed,
        options=case.options or None,
    )
    if case.raises:
        exc_type = _exception_class(case.raises)
        try:
            codec.parse(case.input_data.calctype, **kwargs)
        except exc_type as exc:
            if case.expected is not None:
                assert_data_equal(
                    getattr(exc, "data", None),
                    case.expected,
                    rtol=rtol,
                    atol=atol,
                    where=f"{case} partial",
                )
        else:
            raise AssertionError(f"{case}: expected {case.raises} but decode succeeded")
        return
    assert case.expected is not None, (
        f"{case}: expected.json is required unless raises is set"
    )
    data = codec.parse(case.input_data.calctype, **kwargs)
    assert_data_equal(data, case.expected, rtol=rtol, atol=atol, where=str(case))
    if codec.parsers:
        check_parsers(codec, case, rtol=rtol, atol=atol)

check_codec(codec: Codec, *, inputs: Iterable[ProgramInput] = (), calctypes: Collection[CalcType] | None = None) -> None

Verify the structural contract every codec must satisfy.

Checks that parser targets name real result fields for every calculation type they apply to, that decoding nothing in recovery mode yields a typed empty result and in strict mode raises with typed partial data, that option defaults are well formed, and that encoding each of inputs yields a native invocation whose files stay inside the calculation directory.

Parameters:

Name Type Description Default
codec Codec

The codec to check.

required
inputs Iterable[ProgramInput]

Inputs the codec must be able to encode.

()
calctypes Collection[CalcType] | None

Calculation types to exercise; defaults to codec.calctypes.

None

Raises:

Type Description
AssertionError

Listing every problem found.

Source code in src/qccodec/testing.py
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
def check_codec(
    codec: Codec,
    *,
    inputs: Iterable[ProgramInput] = (),
    calctypes: Collection[CalcType] | None = None,
) -> None:
    """Verify the structural contract every codec must satisfy.

    Checks that parser targets name real result fields for every calculation type
    they apply to, that decoding nothing in recovery mode yields a typed empty result
    and in strict mode raises with typed partial data, that option defaults are
    well formed, and that encoding each of ``inputs`` yields a native invocation whose
    files stay inside the calculation directory.

    Args:
        codec: The codec to check.
        inputs: Inputs the codec must be able to encode.
        calctypes: Calculation types to exercise; defaults to ``codec.calctypes``.

    Raises:
        AssertionError: Listing every problem found.
    """
    problems: list[str] = []
    types = list(calctypes if calctypes is not None else codec.calctypes)
    if not codec.program:
        problems.append("program name is empty")
    if not codec.calctypes:
        problems.append("no calctypes declared")
    for key in codec.options:
        if not isinstance(key, str) or not key:
            problems.append(f"option name {key!r} is not a non-empty string")

    for spec in codec.parsers:
        problems.extend(_check_spec(spec, types))

    if codec.parsers:
        for calctype in types:
            data_type = get_data_type(calctype)
            try:
                empty = codec.parse(calctype, failed=True)
            except Exception as exc:  # noqa: BLE001
                problems.append(f"{calctype.value}: parse(failed=True) raised {exc!r}")
            else:
                if not isinstance(empty, data_type):
                    problems.append(
                        f"{calctype.value}: parse(failed=True) returned {type(empty).__name__}"
                    )
            try:
                codec.parse(calctype)
            except ParserError as exc:
                if not isinstance(exc.data, data_type):
                    problems.append(
                        f"{calctype.value}: strict parse raised without typed partial data"
                    )
            except Exception as exc:  # noqa: BLE001
                problems.append(f"{calctype.value}: strict parse raised {exc!r}")

    for inp in inputs:
        try:
            native = codec.encode(inp)
        except Exception as exc:  # noqa: BLE001
            problems.append(f"encode({inp.calctype.value}) raised {exc!r}")
            continue
        for name in native.files:
            if Path(name).is_absolute() or ".." in Path(name).parts:
                problems.append(
                    f"encode({inp.calctype.value}) file escapes directory: {name!r}"
                )
        if isinstance(native, SubprocessInput) and not native.command:
            problems.append(f"encode({inp.calctype.value}) returned an empty command")
        if isinstance(native, PythonInput) and not callable(native.function):
            problems.append(f"encode({inp.calctype.value}) function is not callable")

    gated = [
        spec.parser.__name__ for spec in codec.parsers if spec.versions is not None
    ]
    has_version_parser = any(spec.target == VERSION_TARGET for spec in codec.parsers)
    if gated and not has_version_parser:
        problems.append(
            f"parsers {gated} declare versions but no parser fills "
            f"{VERSION_TARGET!r}, so the version is only known when passed in"
        )

    if codec.propagate_wfn is not None and not callable(codec.propagate_wfn):
        problems.append("propagate_wfn is not callable")
    if problems:
        raise AssertionError(
            f"Codec {codec.program!r} failed checks:\n  " + "\n  ".join(problems)
        )