Reading and writing

pefftacular.read_peff

read_peff(
    source: str | Path | IO[str] | IO[bytes],
    *,
    compression: Compression = "infer",
) -> 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; compression is described on :class:PeffReader ("infer" sniffs the magic bytes). 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] | IO[bytes], *, compression: Compression = "infer"
) -> 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; ``compression`` is described on
    :class:`PeffReader` (``"infer"`` sniffs the magic bytes). Undecodable (non-UTF-8)
    or corrupt compressed input raises ``PeffParseError`` chained to the cause.
    """
    with PeffReader(source, compression=compression) 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] | IO[bytes],
    *,
    compression: Compression = "infer",
)

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) and closed in __exit__; a stream you pass in is read but never closed.

compression works like the pandas argument of that name. "infer" (the default) decompresses a gzip, bzip2 or xz path detected from its magic bytes, not its suffix, and reads an open handle as it is. "gzip", "bz2" or "xz" forces that format (PeffParseError if the bytes are not in it); an open handle must then be binary (open(path, "rb")). None reads plain text even if the bytes look compressed. Any other value raises PeffError. 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] | IO[bytes], *, compression: Compression = "infer") -> None:
    self._source = source
    self._compression = check_compression(compression)
    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] | IO[bytes],
    *,
    verify: bool = True,
    compression: Compression = "infer",
) -> 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.

compression works like the pandas argument of that name. With "infer" (the default) a path ending in .gz, .bz2 or .xz (any case) is written compressed in that format and any other path or an open handle as plain text. "gzip", "bz2" or "xz" compresses in that format whatever the suffix; an open handle must then be binary (open(path, "wb")), and it is left open. None always writes plain text. gzip output has mtime=0, so the same entries give byte-identical files. read_peff / PeffReader read all of these back.

Source code in src/pefftacular/_writer.py

def write_peff(
    header: FileHeader,
    entries: Iterable[SequenceEntry],
    dest: str | Path | IO[str] | IO[bytes],
    *,
    verify: bool = True,
    compression: Compression = "infer",
) -> 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.

    ``compression`` works like the pandas argument of that name. With ``"infer"`` (the
    default) a path ending in ``.gz``, ``.bz2`` or ``.xz`` (any case) is written
    compressed in that format and any other path or an open handle as plain text.
    ``"gzip"``, ``"bz2"`` or ``"xz"`` compresses in that format whatever the suffix; an
    open handle must then be binary (``open(path, "wb")``), and it is left open. ``None``
    always writes plain text. gzip output has ``mtime=0``, so the same entries give
    byte-identical files. ``read_peff`` / ``PeffReader`` read all of these back.
    """
    compression = check_compression(compression)
    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)):
            path = Path(dest)
            kind = from_suffix(path) if compression == "infer" else compression
            logger.debug("writing PEFF file: %s (%d entries, compression=%s)", dest, count, kind)
            with _open_for_write(path, kind) 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)
            kind = None if compression == "infer" else compression
            if not is_binary(dest):
                if kind is not None:
                    err = PeffError(
                        f"compression={kind!r} needs a binary handle, got a text handle ({type(dest).__name__})"
                    )
                    err.add_note('hint: open in binary mode, open(path, "wb"), or pass the path itself')
                    raise err
                text_dest = cast("IO[str]", dest)
                text_dest.write(header_text)
                shutil.copyfileobj(spool, text_dest)
            else:
                _write_binary(header_text, spool, cast("IO[bytes]", dest), kind)

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

Records

to_records turns entries into flat dicts for a data frame (see the README section on tables). from_records is its inverse and takes the file header, whose CustomKeyDef declarations are what keep custom key values from being read back as unknown keys.

pefftacular.from_records

from_records(
    records: Iterable[Mapping[str, object]],
    header: FileHeader | None = None,
) -> list[SequenceEntry]

Build :class:SequenceEntry objects from flat records, the inverse of :func:to_records.

Writes a data frame back to PEFF::

header, entries = read_peff("in.peff")
df = pandas.DataFrame(to_records(entries))
write_peff(header, from_records(df.to_dict("records"), header), "out.peff")

Each record is a mapping with every key in :data:RECORD_KEYS. A key that is not one of them, or one of them left out, raises: a dropped column is how a caller loses data without noticing. None, float NaN and the pandas/polars missing markers mean the entry has no value for that key, so the key is left out of the entry; "" is an empty value and is kept, because a PEFF file can hold one (\PName= reads back as ""). prefix, db_unique_id and sequence are required.

header is only read for its :class:CustomKeyDef declarations, matched to each record by prefix. It is needed whenever a record has custom_values, because only a declared key is read back as a custom value rather than an unknown one, and passing it is what keeps those values from being silently dropped. The entries are returned on their own, as :func:to_records takes them, so the caller keeps owning the header and passes it to :func:write_peff unchanged.

ncbi_tax_id, length, sv, ev and pe accept ints, integral floats (9606.0, from a frame column with gaps) and digit strings. decoy accepts True, 0, 1 and "true"/"false". The annotation columns, custom_values and extra are the PEFF value text :func:to_records emits, and are used as given.

Every entry is parsed from the description line that :func:write_peff would have written, then turned back into a record and compared with the one that came in, so a column that could not be honoured raises instead of changing the data.

Example

from pefftacular import SequenceEntry, from_records, to_records entry = SequenceEntry(prefix="sp", db_unique_id="P1", sequence="ACDE", pname="Alpha") record = to_records([entry])[0] record.get("pname"), record.get("variant_simple") ('Alpha', None) from_records([record]) == [entry] True

Raises:

Type	Description
PeffWriteError	A missing or unknown key, a bad value, a key set twice, a custom_values key no CustomKeyDef declares, or a value that would not read back as itself. index is the record's position.

Source code in src/pefftacular/_records.py

def from_records(records: Iterable[Mapping[str, object]], header: FileHeader | None = None) -> list[SequenceEntry]:
    """Build :class:`SequenceEntry` objects from flat records, the inverse of :func:`to_records`.

    Writes a data frame back to PEFF::

        header, entries = read_peff("in.peff")
        df = pandas.DataFrame(to_records(entries))
        write_peff(header, from_records(df.to_dict("records"), header), "out.peff")

    Each record is a mapping with **every** key in :data:`RECORD_KEYS`. A key that is not
    one of them, or one of them left out, raises: a dropped column is how a caller loses
    data without noticing. ``None``, float ``NaN`` and the pandas/polars missing markers
    mean the entry has no value for that key, so the key is left out of the entry; ``""``
    is an empty value and is kept, because a PEFF file can hold one (``\\PName=`` reads
    back as ``""``). ``prefix``, ``db_unique_id`` and ``sequence`` are required.

    ``header`` is only read for its :class:`CustomKeyDef` declarations, matched to each
    record by ``prefix``. It is needed whenever a record has ``custom_values``, because
    only a declared key is read back as a custom value rather than an unknown one, and
    passing it is what keeps those values from being silently dropped. The entries are
    returned on their own, as :func:`to_records` takes them, so the caller keeps owning
    the header and passes it to :func:`write_peff` unchanged.

    ``ncbi_tax_id``, ``length``, ``sv``, ``ev`` and ``pe`` accept ints, integral floats
    (``9606.0``, from a frame column with gaps) and digit strings. ``decoy`` accepts
    ``True``, ``0``, ``1`` and ``"true"``/``"false"``. The annotation columns,
    ``custom_values`` and ``extra`` are the PEFF value text :func:`to_records` emits, and
    are used as given.

    Every entry is parsed from the description line that :func:`write_peff` would have
    written, then turned back into a record and compared with the one that came in, so a
    column that could not be honoured raises instead of changing the data.

    Example:
        >>> from pefftacular import SequenceEntry, from_records, to_records
        >>> entry = SequenceEntry(prefix="sp", db_unique_id="P1", sequence="ACDE", pname="Alpha")
        >>> record = to_records([entry])[0]
        >>> record.get("pname"), record.get("variant_simple")
        ('Alpha', None)
        >>> from_records([record]) == [entry]
        True

    Raises:
        PeffWriteError: A missing or unknown key, a bad value, a key set twice, a
            ``custom_values`` key no ``CustomKeyDef`` declares, or a value that would not
            read back as itself. ``index`` is the record's position.
    """
    # Built the way PeffReader builds it (_parser.py:1153), so a record is read back with
    # the same defs the file itself would have been read with.
    defs_by_prefix: dict[str, dict[str, CustomKeyDef]] = {}
    for db in header.databases if header is not None else ():
        if db.prefix and db.custom_key_defs:
            defs_by_prefix[db.prefix] = {ckd.key_name: ckd for ckd in db.custom_key_defs}
    return [_record_entry(record, i, defs_by_prefix) for i, record in enumerate(records)]

Compression

Compression types the compression argument of read_peff, PeffReader, write_peff and to_records.

pefftacular.Compression module-attribute

Compression = Literal['infer', 'gzip', 'bz2', 'xz'] | None

How a PEFF file is (de)compressed, like the compression argument of pandas.

• "infer" (the default): readers detect gzip, bzip2 or xz from the magic bytes; writers compress by the path suffix (.gz, .bz2, .xz, any case). An open handle is read and written as plain text.
• "gzip", "bz2", "xz": always that format, whatever the suffix or content.
• None: never compressed.