topmark.pipeline.views ¶

def release(self) -> None:
    """Release any materialized buffers to reduce memory usage."""
    ...

def line_count(self) -> int:
    """Return the number of logical lines in the file image.

    Returns:
        int: Total number of lines available via ``iter_lines()``.
    """
    ...

def iter_lines(self) -> Iterable[str]:
    r"""Iterate the file's logical lines, preserving original line endings.

    Returns:
        Iterable[str]: An iterator over the file's lines. The iterator
        must yield strings exactly as they appear in the source (e.g.,
        with ``\n``/``\r\n`` kept as read).
    """
    ...

def line_count(self) -> int:
    """Return the number of lines in the underlying list.

    Returns:
        int: Total line count.
    """
    return 0 if self._lines is None else len(self._lines)

def iter_lines(self) -> Iterable[str]:
    """Iterate lines from the underlying list without copying.

    Returns:
        Iterable[str]: An iterator over the stored lines.
    """
    return iter(self._lines or ())

def release(self) -> None:
    """Release materialized lines."""
    self._lines = None

def release(self) -> None:
    """Release header buffers (lines, block, mapping)."""
    self.lines = None
    self.block = None
    self.mapping = None

def release(self) -> None:
    """Release the diff payload to reduce memory usage."""
    self.builtins = None
    self.selected = None

def release(self) -> None:
    """Release the renderer payload to reduce memory usage."""
    self.lines = None
    self.block = None

def iter_lines(self) -> Iterable[str]:
    """Iterate updated file lines repeatably."""
    ...

def iter_lines(self) -> Iterable[str]:
    """Iterate the composed updated image without materializing it as one list.

    Returns:
        Iterable[str]: Updated lines from every segment in order.
    """
    return iter(self)

def release(self) -> None:
    """Release planned-edit metadata."""
    self.edits = ()

def release(self) -> None:
    """Release the updated file image payload to reduce memory usage."""
    self.lines = None

def release(self) -> None:
    """Release the diff payload to reduce memory usage."""
    self.text = None

def release_all(
    self,
) -> None:
    """Release all non-None views safely (idempotent)."""
    self.release_image()
    self.release_header()
    self.release_build()
    self.release_render()
    self.release_updated()
    self.release_edit()
    self.release_diff()

def release_image(self) -> None:
    """Release the original file image view when it is no longer needed."""
    if self.image:
        self.image.release()

def release_header(self) -> None:
    """Release detected-header payloads when they are no longer needed."""
    if self.header:
        self.header.release()

def release_build(self) -> None:
    """Release builder payloads when they are no longer needed."""
    if self.build:
        self.build.release()

def release_render(self) -> None:
    """Release rendered-header payloads when they are no longer needed."""
    if self.render:
        self.render.release()

def release_updated(self) -> None:
    """Release updated-file payloads when they are no longer needed."""
    if self.updated:
        self.updated.release()

def release_edit(self) -> None:
    """Release planned-edit metadata when it is no longer needed."""
    if self.edit:
        self.edit.release()

def release_diff(self) -> None:
    """Release diff payloads when they are no longer needed."""
    if self.diff:
        self.diff.release()

def release_consumed(
    self,
    *,
    remaining_view_consumers: set[ViewSlot],
    keep_diff_view: bool = False,
) -> None:
    """Release views that no remaining pipeline step declares as consumed.

    Args:
        remaining_view_consumers: View slots consumed by steps that have not run yet.
        keep_diff_view: Whether to preserve the diff view for callers that render it after the
            pipeline completes.
    """
    if ViewSlot.IMAGE not in remaining_view_consumers:
        self.release_image()

    if ViewSlot.HEADER not in remaining_view_consumers:
        self.release_header()

    if ViewSlot.BUILD not in remaining_view_consumers:
        self.release_build()

    if ViewSlot.RENDER not in remaining_view_consumers:
        self.release_render()

    if ViewSlot.UPDATED not in remaining_view_consumers:
        self.release_updated()

    if ViewSlot.EDIT not in remaining_view_consumers:
        self.release_edit()

    if not keep_diff_view and ViewSlot.DIFF not in remaining_view_consumers:
        self.release_diff()

def as_dict(self) -> dict[str, object]:
    """Short machine-friendly summary; avoid heavy text blobs."""
    return {
        "image_lines": self.image.line_count() if self.image else 0,
        "header_range": getattr(self.header, "range", None),
        "header_fields": dict(self.header.mapping or {}) if self.header else None,
        "build_selected": dict(self.build.selected or {}) if self.build else None,
        "render_line_count": (
            len(self.render.lines) if (self.render and self.render.lines is not None) else 0
        ),
        "updated_has_lines": self.updated is not None and self.updated.lines is not None,
        "edit_count": len(self.edit.edits) if self.edit else 0,
        "diff_present": bool(self.diff and self.diff.text),
    }