Process Data

Data Pipeline

Annual NAICS Pull - Our site adds local industry and trade levels for data analysis Open CEDA - 60,000 emission factors, 400 industries, 148 countries Exiobase.eu - International impact factors extending Comtrade UN trade data

The Model.earth data pipeline saves static Community-Data and Community-Timelines for state, county and zip code data generated for Industry Supply-Chain IO Charts and Community Forecasting Timelines.

Our work with International Trade Flow SQL and US State Models data provides a structure for future models integrated with the Google Data Commons API to use UN Goal Timelines in RealityStream forecasting.

View Data Pipeline Nodes from node.csv

While USEEIO has been discontinued by the US EPA, their Embeddable IO Charts and State Impact Reports provide examples of pulling data from static json files containing USEEIO state data for fast page load times.

Embeddable Datasets

Community Model Pages - Parameters for embedding in local sites

Impact Charts - US Environmentally-Extended Input-Output (USEEIO) - Goods and Services

Impact Profiles - Using Environmental Product Declarations (EPDs)

Product Environmental Data Pipeline

Products Python Pipeline - Fetches and processes Environmental Product Declarations (EPDs) from BuildingTransparency.org

The products pipeline includes:

• EPD Data Fetcher (product-footprints.py) - Fetches EPD data for all US states and multiple countries (184,614+ products)
• Emissions Analyzer (analyze_emissions_data.py) - Analyzes GWP coverage and impact categories across all downloaded EPDs
• Transportation Impact Calculator (calculate_transportation_impact.py) - Utility functions for calculating A4 stage transportation impacts and adjusting GWP values

Data Output: YAML files organized by country and category in products-data repo

Documentation: model.earth/products/ - Full documentation on EPD structure, GWP fields, transportation impacts, and API usage

Pipeline Nodes: See nodes.csv for prod_001 (EPD Data Fetcher) and prod_002 (EPD Emissions Analyzer)

Data Sources and Prep

Community Datasets - NAICS Industry Data with Gaps Filled

Machine Learning Imputation Algorithms for NAICS industries - US Bureau of Labor Statistics (BLS)

Impact Heatmap from JSON - Earlier Goods and Service Heatmap Mockup

Pipeline Management

Unified Pipeline Manager (manage_pipelines.py) - Centralized tool for managing all data update processes

# List all available pipelines
python manage_pipelines.py list

# Show details for a specific node
python manage_pipelines.py info prod_001

# Run a pipeline node
python manage_pipelines.py run prod_001

# Show dependency chain
python manage_pipelines.py dependencies prod_001

# Show status overview
python manage_pipelines.py status

The management script provides:

• Centralized listing of all pipeline nodes with metadata
• Dependency tracking to understand execution order
• Unified execution interface for running any pipeline node
• Status overview with statistics on processing times and capabilities

All pipeline nodes are documented in nodes.csv and nodes.json for integration with workflow automation tools like n8n.

Pipeline Admin Interface

View Pipeline Admin and python webroot setup

Note: It's probably not necessary to run Flask if the python webroot is running.

Flask Server (flask_server.py) - Web API server for running pipeline nodes from the admin interface

The Flask server provides a REST API that allows the Data Pipeline Admin page to execute pipeline nodes via HTTP requests. This enables a web-based interface for managing and running data update processes.

Setup:

1. Install Flask and Flask-CORS:

   pip install flask flask-cors

2. Start the Flask server:

   python flask_server.py

   The server runs on http://localhost:5001 by default (port 5001 is used to avoid conflicts with macOS AirPlay on port 5000).

3. Verify the server is running:

   curl http://localhost:5001/health

API Endpoints:

• GET /health - Health check endpoint for availability detection
• GET /api/nodes - List all pipeline nodes
• GET /api/nodes/<node_id> - Get information about a specific node
• POST /api/nodes/run - Execute a pipeline node command
• GET /api/nodes/<node_id>/status - Get status of a running node

Note: The Flask server uses port 5001 instead of 5000 to avoid conflicts with macOS AirPlay service.

Features:

• Automatic background processing for long-running tasks (very_slow processes)
• Auto-commit workflow - Automatically commits data updates to fork repositories and creates PRs
• CORS enabled for localhost requests from the admin page
• Process status tracking for monitoring long-running executions

For detailed setup instructions and troubleshooting, see the Flask Setup Guide.

Nodes CSV Schema

The nodes.csv file defines all pipeline nodes with the following columns:

Core Columns:

• node_id - Unique identifier (e.g., prod_001)
• name - Human-readable name
• description - Detailed description of what the node does
• type - Node type (e.g., api_fetcher, data_processor, ml_processor)
• order - Execution order number
• link - Working directory path relative to data-pipeline root
• python_cmds - Python command to execute
• output_path - Path where output files are written
• output_info - Description of output files
• folder_size - Estimated output size
• dependencies - Required Python packages (comma-separated)
• api_keys_required - API keys needed (comma-separated, or none)
• processing_time_est - Estimated processing time (fast, medium, slow, very_slow)
• data_sources - Data source description
• n8n_parallel_safe - Whether node can run in parallel (yes/no)
• rate_limited - Whether data source is rate-limited (yes/no)
• actual_size_mb - Actual output size in MB (if known)
• csv_file_count - Number of CSV files generated (if applicable)
• last_transport_date - Date of last data transport/update (YYYY-MM-DD format)

New Flask Integration Columns:

• run_process_available - Whether this node can be run via Flask API (yes/no)
• flask_endpoint - Custom Flask endpoint if different from default (optional)
• requires_interactive - Whether process requires user interaction (yes/no)
• auto_commit - Whether to automatically commit results to fork repository (yes/no)
• target_repo - Target repository for data updates (community-data, community-timelines, products-data, etc.)

Example:

prod_001,EPD Data Fetcher,Fetches EPDs from BuildingTransparency.org,api_fetcher,26,../products/pull,python product-footprints.py,../../../products-data/{COUNTRY}/{CATEGORY}/,YAML files,~500M,"requests,yaml,json,csv",buildingtransparency,very_slow,BuildingTransparency.org EC3 API,no,yes,,,2025-12-05,yes,,no,yes,products-data

Nodes with auto_commit=yes will automatically:

1. Commit changes to the specified target_repo fork
2. Push changes to the fork
3. Create a GitHub Pull Request (via GitHub CLI if available, or provide manual URL)

Opportunties for further integration

Google Data Commons Setup

DataUSA.io Setup

Census Reporter