Reading and writing

pefftacular.read_peff

read_peff(
    source: str | Path | IO[str],
) -> tuple[FileHeader, list[SequenceEntry]]

Convenience: parse an entire PEFF file into header + list of entries.

A path may be plain or gzip/bzip2/xz compressed. Undecodable (non-UTF-8) or corrupt compressed input raises PeffParseError chained to the cause.

Source code in src/pefftacular/_parser.py

def read_peff(source: str | Path | IO[str]) -> tuple[FileHeader, list[SequenceEntry]]:
    """Convenience: parse an entire PEFF file into header + list of entries.

    A path may be plain or gzip/bzip2/xz compressed. Undecodable (non-UTF-8) or
    corrupt compressed input raises ``PeffParseError`` chained to the cause.
    """
    with PeffReader(source) as reader:
        header = reader.header
        entries = list(reader)
    logger.info("read_peff: parsed %d entries across %d database(s)", len(entries), len(header.databases))
    return header, entries

pefftacular.PeffReader

PeffReader(source: str | Path | IO[str])

Lazy reader for PEFF files.

Use it as a context manager, like fastatacular.FastaReader: a path is opened in __enter__ (UTF-8, a leading BOM is skipped; gzip, bzip2 and xz files are decompressed, detected from the magic bytes or the suffix) and closed in __exit__; a stream you pass in is read but never closed. Accessing header or iterating outside the with block raises :class:RuntimeError. Entries can be iterated once; a second iteration over the same reader yields nothing.

Example::

with PeffReader("proteins.peff") as reader:
    header = reader.header
    for entry in reader:
        ...

Source code in src/pefftacular/_parser.py

def __init__(self, source: str | Path | IO[str]) -> None:
    self._source = source
    self._fh: IO[str] | None = None
    self._owns_fh = False
    self._lines: Iterator[str] | None = None
    self._header: FileHeader | None = None
    self._remaining: Iterator[str] | None = None
    self._first_entry_line_no: int = 1
    self._defs_by_prefix: dict[str, dict[str, CustomKeyDef]] = {}

header property

header: FileHeader

Parse and return the file header (cached after first access).

to_records

to_records() -> list[dict[str, str | int | bool | None]]

Read the remaining entries as flat dicts (see :func:pefftacular.to_records).

Source code in src/pefftacular/_parser.py

def to_records(self) -> list[dict[str, str | int | bool | None]]:
    """Read the remaining entries as flat dicts (see :func:`pefftacular.to_records`)."""
    from pefftacular._records import entry_to_record

    self._ensure_header()
    defs = self._defs_by_prefix
    return [entry_to_record(e, defs.get(e.prefix)) for e in self]

pefftacular.write_peff

write_peff(
    header: FileHeader,
    entries: Iterable[SequenceEntry],
    dest: str | Path | IO[str],
    *,
    verify: bool = True,
) -> None

Write a complete PEFF file.

Every entry is validated and formatted before anything is written to dest: on a PeffWriteError a path is not created or truncated and nothing is written to a handle. entries is consumed once, as a stream; formatted text is spooled to a temporary file past 32 MiB instead of being held in memory.

With verify=True (the default) each formatted entry is parsed back and compared with the entry, so a value holding PEFF syntax (\Key= text, an unbalanced paren) raises instead of writing a file that reads back differently. verify=False skips that read-back (about 3/4 of the write time); use it for entries that came from read_peff / PeffReader unchanged, or that you have already written once. The basic checks (empty or malformed prefix, id or sequence, line breaks, duplicate keys) always run.

Source code in src/pefftacular/_writer.py

def write_peff(
    header: FileHeader,
    entries: Iterable[SequenceEntry],
    dest: str | Path | IO[str],
    *,
    verify: bool = True,
) -> None:
    """Write a complete PEFF file.

    Every entry is validated and formatted before anything is written to ``dest``:
    on a ``PeffWriteError`` a path is not created or truncated and nothing is
    written to a handle. ``entries`` is consumed once, as a stream; formatted text
    is spooled to a temporary file past 32 MiB instead of being held in memory.

    With ``verify=True`` (the default) each formatted entry is parsed back and
    compared with the entry, so a value holding PEFF syntax (``\\Key=`` text, an
    unbalanced paren) raises instead of writing a file that reads back
    differently. ``verify=False`` skips that read-back (about 3/4 of the write
    time); use it for entries that came from ``read_peff`` / ``PeffReader``
    unchanged, or that you have already written once. The basic checks (empty or
    malformed prefix, id or sequence, line breaks, duplicate keys) always run.
    """
    if header is None:
        raise PeffWriteError("header must not be None", hint="Pass a FileHeader instance, e.g. from read_peff()")

    bad = _line_break_at(header, "header")
    if bad:
        raise PeffWriteError(
            f"{bad} contains a line break", hint="PEFF header values are single-line; remove the \\n or \\r"
        )

    header_text = _format_header(header)
    with warnings.catch_warnings():
        warnings.simplefilter("ignore")
        _check_header_reads_back(header, header_text)

    defs_by_prefix: dict[str, dict[str, CustomKeyDef]] = {}
    for db in header.databases:
        if db.prefix and db.custom_key_defs:
            defs_by_prefix[db.prefix] = {ckd.key_name: ckd for ckd in db.custom_key_defs}

    spool = tempfile.SpooledTemporaryFile(
        max_size=_SPOOL_BYTES, mode="w+", encoding="utf-8", errors="surrogatepass", newline=""
    )
    with spool:
        count = _format_entries(entries, defs_by_prefix, spool, verify=verify)
        spool.seek(0)
        if isinstance(dest, (str, Path)):
            logger.debug("writing PEFF file: %s (%d entries)", dest, count)
            with Path(dest).open("w", encoding="utf-8") as f:
                f.write(header_text)
                shutil.copyfileobj(spool, f)
        else:
            logger.debug("writing PEFF to in-memory stream: %s (%d entries)", type(dest).__name__, count)
            dest.write(header_text)
            shutil.copyfileobj(spool, dest)

    logger.info("write_peff: wrote %d entries across %d database(s)", count, len(header.databases))