Skip to content

lpgcast.episode

Attributes

JsonScalar = str | int | float | bool | None module-attribute

Classes

ContextItem

Bases: BaseModel

An earlier text in the same thread, shown for reference only.

Source code in lpgcast/episode.py 
15
16
17
18
19
20
21
class ContextItem(BaseModel):
    """An earlier text in the same thread, shown for reference only."""

    model_config = ConfigDict(extra="forbid")

    source_id: str = Field(description="Identifier of the earlier text.")
    text: str = Field(description="Its text.")

Attributes

model_config = ConfigDict(extra='forbid') class-attribute instance-attribute
source_id = Field(description='Identifier of the earlier text.') class-attribute instance-attribute
text = Field(description='Its text.') class-attribute instance-attribute

Episode

Bases: BaseModel

One text to cast.

The structured half of a record — author, channel, thread — belongs in metadata and is written by an ordinary GraFlo resource, never read off the text by the model; metadata is not shown to the model.

Source code in lpgcast/episode.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
class Episode(BaseModel):
    """One text to cast.

    The structured half of a record — author, channel, thread — belongs in
    ``metadata`` and is written by an ordinary GraFlo resource, never read off
    the text by the model; ``metadata`` is not shown to the model.
    """

    model_config = ConfigDict(extra="forbid")

    text: str = Field(min_length=1, description="The text to cast.")
    source_id: str = Field(
        min_length=1, description="Stable identifier of the source record."
    )
    reference_time: datetime | None = Field(
        default=None,
        description="When the text was written; resolves relative dates.",
    )
    metadata: dict[str, JsonScalar] = Field(
        default_factory=dict, description="Source metadata, never shown to the model."
    )
    context: list[ContextItem] = Field(
        default_factory=list,
        description="Preceding texts in the thread, for resolving references only.",
    )

Attributes

context = Field(default_factory=list, description='Preceding texts in the thread, for resolving references only.') class-attribute instance-attribute
metadata = Field(default_factory=dict, description='Source metadata, never shown to the model.') class-attribute instance-attribute
model_config = ConfigDict(extra='forbid') class-attribute instance-attribute
reference_time = Field(default=None, description='When the text was written; resolves relative dates.') class-attribute instance-attribute
source_id = Field(min_length=1, description='Stable identifier of the source record.') class-attribute instance-attribute
text = Field(min_length=1, description='The text to cast.') class-attribute instance-attribute

EpisodeFileError

Bases: ValueError

A line of an episodes file that is not a valid episode.

Source code in lpgcast/episode.py 
51
52
53
54
55
56
57
class EpisodeFileError(ValueError):
    """A line of an episodes file that is not a valid episode."""

    def __init__(self, path: Path, line: int, reason: str) -> None:
        super().__init__(f"{path}:{line}: {reason}")
        self.path = path
        self.line = line

Attributes

line = line instance-attribute
path = path instance-attribute

Methods:

__init__(path, line, reason)
Source code in lpgcast/episode.py 
54
55
56
57
def __init__(self, path: Path, line: int, reason: str) -> None:
    super().__init__(f"{path}:{line}: {reason}")
    self.path = path
    self.line = line

Functions:

read_episodes(path)

Read episodes from a JSON Lines file, one object per line.

Blank lines are skipped.

Raises:

Type Description
EpisodeFileError

on the first line that is not a valid episode.
Source code in lpgcast/episode.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
def read_episodes(path: str | Path) -> Iterator[Episode]:
    """Read episodes from a JSON Lines file, one object per line.

    Blank lines are skipped.

    Raises:
        EpisodeFileError: on the first line that is not a valid episode.
    """
    path = Path(path)
    with open(path, encoding="utf-8") as handle:
        for number, line in enumerate(handle, start=1):
            if not line.strip():
                continue
            try:
                yield Episode.model_validate(json.loads(line))
            except json.JSONDecodeError as error:
                raise EpisodeFileError(
                    path, number, f"not JSON: {error.msg}"
                ) from error
            except ValidationError as error:
                raise EpisodeFileError(path, number, str(error)) from error