pipeline

pipeline

Amendment pipeline orchestrator.

Coordinates the full amendment analysis workflow

IDENTIFY → PDF_DOWNLOAD_PARSE → STENO_DOWNLOAD_PARSE → MERGE → RESOLVE_IDS → RESOLVE_SUBMITTERS → LLM_SUMMARIZE → CACHE

PDF text is the primary source for amendment structure (letters, submitter names, per-amendment text). Steno records provide vote linkage (vote numbers, results, stances). The MERGE stage combines both data sources.

Follows the same async pattern as TiskPipelineService.

AmendmentPipelineService(cache_dir, _progress=dict(), _tasks=dict(), _flags=dict()) dataclass

Orchestrates the amendment analysis pipeline for all periods.

Follows the same pattern as TiskPipelineService: - Background async tasks - Per-period progress tracking - Cancellation support

Attributes:

Name	Type	Description
cache_dir	Path	Base cache directory.
_progress	dict[int, AmendmentProgress]	Per-period progress tracking.
_tasks	dict[int, Task]	Per-period asyncio tasks.

progress property

Current progress for all periods.

start_period(period, period_data, on_complete=None, on_progress=None, mode=AmendmentMode.FULL, refresh_active=False)

Start the amendment pipeline for a single period.

Parameters:

Name	Type	Description	Default
period	int	Electoral period number.	required
period_data	PeriodData	Loaded period data with tisk_lookup populated.	required
on_complete	Callable | None	Optional callback(period, bills) on completion.	None
on_progress	Callable[[int, list[BillAmendmentData]], None] | None	Optional callback(period, bills) for incremental UI refresh.	None
mode	AmendmentMode	Pipeline execution mode.	FULL
refresh_active	bool	When True, re-scrape active bills' histories so newly third-read bills are picked up (daily refresh).	False

Source code in pspcz_analyzer/services/amendments/pipeline.py

def start_period(
    self,
    period: int,
    period_data: PeriodData,
    on_complete: Callable | None = None,
    on_progress: Callable[[int, list[BillAmendmentData]], None] | None = None,
    mode: AmendmentMode = AmendmentMode.FULL,
    refresh_active: bool = False,
) -> None:
    """Start the amendment pipeline for a single period.

    Args:
        period: Electoral period number.
        period_data: Loaded period data with tisk_lookup populated.
        on_complete: Optional callback(period, bills) on completion.
        on_progress: Optional callback(period, bills) for incremental UI refresh.
        mode: Pipeline execution mode.
        refresh_active: When True, re-scrape active bills' histories so newly
            third-read bills are picked up (daily refresh).
    """
    if period in self._tasks and not self._tasks[period].done():
        logger.info("[amendment pipeline] Already running for period {}", period)
        return

    prog = AmendmentProgress(status=AmendmentStatus.RUNNING)
    self._progress[period] = prog
    flag = CancellationFlag(period)
    self._flags[period] = flag

    async def _run() -> None:
        try:
            bills = await asyncio.to_thread(
                _run_pipeline_sync,
                period,
                period_data,
                self.cache_dir,
                prog,
                on_progress,
                mode,
                refresh_active,
                flag.check,
            )
            prog.status = AmendmentStatus.COMPLETED
            prog.stage = AmendmentStage.COMPLETED
            if on_complete:
                on_complete(period, bills)
        except PipelineCancelled:
            prog.status = AmendmentStatus.CANCELLED
            logger.info("[amendment pipeline] Cancelled for period {}", period)
        except asyncio.CancelledError:
            prog.status = AmendmentStatus.CANCELLED
            logger.info("[amendment pipeline] Cancelled for period {}", period)
            raise
        except Exception:
            prog.status = AmendmentStatus.FAILED
            prog.stage = AmendmentStage.FAILED
            logger.opt(exception=True).error(
                "[amendment pipeline] Failed for period {}", period
            )
        finally:
            self._flags.pop(period, None)

    task = asyncio.create_task(_run())
    task.add_done_callback(lambda t: self._on_task_done(period, t))
    self._tasks[period] = task

is_running(period)

Check if the pipeline is currently running for a period.

Source code in pspcz_analyzer/services/amendments/pipeline.py

def is_running(self, period: int) -> bool:
    """Check if the pipeline is currently running for a period."""
    task = self._tasks.get(period)
    return task is not None and not task.done()

get_task(period)

Get the running asyncio.Task for a period, or None if not running.

Source code in pspcz_analyzer/services/amendments/pipeline.py

def get_task(self, period: int) -> asyncio.Task | None:
    """Get the running asyncio.Task for a period, or None if not running."""
    task = self._tasks.get(period)
    if task is not None and not task.done():
        return task
    return None

cancel_period(period)

Cancel the amendment pipeline for a single period.

Flips the period's cancellation flag; the worker thread stops cooperatively at its next checkpoint. (task.cancel() alone cannot interrupt work running inside asyncio.to_thread — it only releases the awaiting coroutine.)

Returns True if a cancellation was requested.

Source code in pspcz_analyzer/services/amendments/pipeline.py

def cancel_period(self, period: int) -> bool:
    """Cancel the amendment pipeline for a single period.

    Flips the period's cancellation flag; the worker thread stops
    cooperatively at its next checkpoint. (task.cancel() alone cannot
    interrupt work running inside asyncio.to_thread — it only
    releases the awaiting coroutine.)

    Returns True if a cancellation was requested.
    """
    flag = self._flags.get(period)
    if flag is None or not self.is_running(period):
        return False
    flag.cancel()
    logger.info("[amendment pipeline] Cancellation requested for period {}", period)
    return True

cancel_all()

Request cancellation of all running amendment pipelines.

Flips every period flag; running stages stop cooperatively at their next checkpoint. Await wait_stopped() to let tasks drain.

Source code in pspcz_analyzer/services/amendments/pipeline.py

def cancel_all(self) -> None:
    """Request cancellation of all running amendment pipelines.

    Flips every period flag; running stages stop cooperatively at
    their next checkpoint. Await wait_stopped() to let tasks drain.
    """
    for flag in self._flags.values():
        flag.cancel()
    logger.info("[amendment pipeline] Cancellation requested for all periods")

wait_stopped(timeout=10.0) async

Wait for running tasks to drain after cancel_all().

Tasks still running after timeout are hard-cancelled — that releases the event-loop side immediately; worker threads exit at their next cancellation checkpoint.

Source code in pspcz_analyzer/services/amendments/pipeline.py

async def wait_stopped(self, timeout: float = 10.0) -> None:
    """Wait for running tasks to drain after cancel_all().

    Tasks still running after *timeout* are hard-cancelled — that
    releases the event-loop side immediately; worker threads exit
    at their next cancellation checkpoint.
    """
    tasks = [t for t in self._tasks.values() if not t.done()]
    if tasks:
        _, pending = await asyncio.wait(tasks, timeout=timeout)
        if pending:
            logger.warning(
                "[amendment pipeline] {} tasks still draining after {}s — forcing cancellation",
                len(pending),
                timeout,
            )
            for t in pending:
                t.cancel()
            await asyncio.gather(*pending, return_exceptions=True)
    self._tasks.clear()
    self._flags.clear()