API: reader

Streaming input for STL files.

Module: stl_parser.reader Import: from stl_parser import stream_parse, STLReader, ReaderStats


stream_parse()

stream_parse(
    source: Union[str, Path, TextIO, Iterable[str]],
    *,
    where: Optional[Dict[str, Any]] = None,
    on_error: Literal["skip", "raise"] = "skip"
) -> Generator[Statement, None, None]

Memory-efficient generator that yields statements one at a time.

Parameters:

ParameterTypeDefaultDescription
sourcestr, Path, TextIO, Iterable[str]File path, file handle, StringIO, or line iterable
wheredictNoneFilter conditions (same operators as find())
on_errorstr"skip""skip" silently skips bad lines; "raise" raises immediately

Yields: Statement objects.

Raises: STLReaderError (E960-E962)

Example:

for stmt in stream_parse("data.stl", where={"confidence__gte": 0.8}):
    print(stmt)

STLReader

Context manager with statistics and tail mode.

STLReader(
    source: Union[str, Path, TextIO],
    *,
    where: Optional[Dict[str, Any]] = None,
    on_error: Literal["skip", "raise"] = "skip",
    tail: bool = False,
    tail_interval: float = 0.5
)

Parameters:

ParameterTypeDefaultDescription
sourcestr, Path, TextIOFile path or file handle
wheredictNoneFilter conditions
on_errorstr"skip"Error handling mode
tailboolFalseEnable tail mode (like tail -f)
tail_intervalfloat0.5Poll interval in seconds for tail mode

Methods:

MethodReturn TypeDescription
__iter__()Generator[Statement]Iterate over statements
read_all()ParseResultConsume generator into a ParseResult
close()NoneClose file handles

Properties:

PropertyTypeDescription
statsReaderStatsRead-only statistics snapshot

Example:

with STLReader("events.stl", where={"rule": "causal"}) as reader:
    for stmt in reader:
        print(stmt)
    print(reader.stats)

ReaderStats

class ReaderStats(BaseModel):
    lines_processed: int = 0
    statements_yielded: int = 0
    errors_skipped: int = 0
    comments_seen: int = 0
    blank_lines: int = 0