Pipeline Orchestration

Relevant source files

The following files were used as context for generating this wiki page:

• pipeline.py
• backend.py
• taxcalc.py
• forms.py
• parsing/validator.py
• README.md

The Pipeline Orchestration system in the Rosy Tax Processor acts as the central workflow engine that coordinates the transformation of raw tax documents into finalized tax returns. It manages a sequence of modular operations including file ingestion, document classification, field extraction, validation, tax calculation, and PDF form generation.

This orchestration layer abstracts the complexity of individual modules, providing a unified interface for the Flask backend to process multiple files simultaneously. By aggregating data from various sources (such as W-2s and 1099s), the pipeline ensures that a comprehensive tax estimate can be computed and mapped onto a standardized IRS Form 1040 template.

Sources: pipeline.py:1-7, README.md:10-25

Core Workflow Architecture

The orchestration logic is primarily contained within pipeline.py, which utilizes a multi-stage approach to process documents. The system supports both per-file analysis and global data aggregation for final reporting.

High-Level Data Flow

The following diagram illustrates the lifecycle of a document within the orchestration pipeline:

Orchestration Components

The orchestration is divided into two primary functions: parse_paths for individual file processing and run_pipeline_on_paths for the end-to-end workflow.

Stage	Description	Key Functions
Ingestion	Converts PDFs, images, or text files into string format for processing.	ingest_paths
Classification	Identifies document types (e.g., W-2, 1099) using heuristics.	detect_document_type
Extraction	Pulls specific tax fields (wages, withholding) based on document type.	extract_fields
Validation	Checks extracted data for numeric sanity and required fields.	validate_fields
Aggregation	Sums income and withholding values across all uploaded documents.	run_pipeline_on_paths
Calculation	Computes AGI, taxable income, and tax due/refund based on status.	compute_tax_estimate
Generation	Fills a PDF template or generates a ReportLab draft.	generate_1040_draft

Sources: pipeline.py:16-35, pipeline.py:38-60, parsing/validator.py:17-54

Data Aggregation and Transformation

The pipeline aggregates data from multiple sources to provide a singular financial view. It specifically targets keywords like wages and amount for income, and federal_income_tax_withheld for payments.

Aggregation Logic

The pipeline iterates through per_file results to accumulate total_wages and total_withholding. It performs string cleaning (removing $ and ,) before converting values to floats to ensure robust processing of OCR-extracted text.

    for r in per_file:
        f = r.get('fields', {})
        try:
            if 'wages' in f:
                total_wages += float(str(f['wages']).replace(',', '').replace('$', ''))
            if 'amount' in f:
                total_wages += float(str(f['amount']).replace(',', '').replace('$', ''))
        except Exception:
            pass

Sources: pipeline.py:64-74

Tax Calculation and Estimation

Once data is aggregated, the pipeline invokes the taxcalc.py module. This module uses filing-status-specific standard deductions and progressive tax brackets.

Calculation Sequence

Document Finalization and Form Filling

The final stage of the pipeline involves the forms.py module, which handles the creation of the physical output. The system attempts to fill a fillable PDF template (1040_fillable.pdf) using an auto-mapping strategy.

Form Generation Hierarchy

1. Exact Mapping: Attempts to match fields to a known 1040 template using create_exact_field_map.
2. Semantic Mapping: If exact matches are insufficient, it uses sentence-transformers to find the most semantically similar field names in the PDF.
3. Fallback Generation: If PDF libraries (pdfrw, reportlab) or templates are missing, it generates a plain text draft.

Sources: forms.py:133-170, forms.py:174-180

API Integration

The backend.py module exposes the orchestration pipeline through two primary endpoints:

• /upload: Receives multiple files, runs the pipeline, and returns extracted data for user review.
• /finalize: Accepts user-edited fields and triggers the final tax calculation and PDF generation.

Sources: backend.py:61-100, backend.py:105-155

Pipeline Result Structure

The object returned by the orchestrator (run_pipeline_on_paths) contains:

• doc_type: Primary document classification.
• fields: Extracted fields from the combined text.
• per_file: List of detailed results for every individual upload.
• aggregated_fields: Summed totals for income and withholding.
• tax_estimate: Calculated financial summary.
• draft_form: Path to the generated PDF.

Sources: pipeline.py:90-99

Summary

Pipeline Orchestration in the Tax Processor provides a modular and extensible framework for tax document automation. By separating document parsing from tax logic and form generation, the system allows for independent updates to tax law (via taxcalc.py) or extraction logic (via parsing/) without disrupting the overall document processing flow.