StreamScope

StreamScope is a configuration-driven, concurrent real-time data pipeline built in Python. It is designed to ingest unseen CSV datasets, normalize them into a generic internal packet format, process them across multiple worker processes, and visualize both the processed stream and pipeline health in real time.

The project is structured around three decoupled concerns:

• Input layer reads raw rows from a configured dataset and maps source columns to internal generic fields.
• Core layer verifies signatures, re-sequences packets, and computes a running average.
• Output layer renders a live dashboard and receives pipeline telemetry through the Observer pattern.

This makes the code reusable across different domains as long as the dataset shape is described correctly in config.json.

Why This Project Exists

The goal of this system is to demonstrate a Phase 3 style real-time processing architecture with:

• bounded multiprocessing queues
• producer-consumer execution
• configurable parallelism
• backpressure-aware ingestion
• functional core style computation
• Observer-based telemetry
• configuration-driven schema mapping

Rather than being tied to one dataset domain, the pipeline works with a generic internal packet shape:

• seq
• entity_name
• time_period
• metric_value
• security_hash

Processed packets also carry:

• is_authentic
• computed_metric

Entry Point

Run the project from the repository root:

python main.py

Main entry file:

• main.py

Required runtime files:

• config.json in the project root
• the dataset file referenced by config.json

High-Level Architecture

At runtime, the system uses three multiprocessing queues:

• Raw Queue: input stream from producer to worker pool
• Verified Queue: intermediate stream from workers to aggregator
• Processed Queue: final stream from aggregator to dashboard

Execution flow:

1. main.py loads and validates config.json.
2. main.py creates the bounded queues and shared counters.
3. CSVProducer reads the configured dataset and emits normalized packets into the raw queue.
4. Multiple worker processes verify packet authenticity in parallel.
5. The aggregator process restores sequence order and computes a running average.
6. PipelineTelemetry monitors queue sizes and counters and notifies the dashboard.
7. DashboardGUI consumes processed packets and renders live charts plus telemetry.

Observer-Based Telemetry

Telemetry is intentionally handled as a separate concern instead of being mixed into the dashboard.

• PipelineTelemetry acts as the subject
• DashboardGUI acts as the observer
• TelemetrySnapshot is the payload sent from subject to observer

The telemetry monitor:

• polls queue sizes
• reads shared counters
• classifies queue health as green / yellow / red
• notifies the dashboard with the latest snapshot

This keeps pipeline monitoring separate from packet processing while still giving the UI live feedback about backpressure and throughput.

Functional Core and Imperative Shell

The pipeline uses a split between pure-ish computation and stateful orchestration:

• Functional logic lives in core/functional.py
• Process orchestration and queue coordination live in main.py, core/worker.py, and core/aggregator.py

Examples:

• signature verification uses hashlib.pbkdf2_hmac
• queue health classification is centralized
• sliding-window updates and average calculation are isolated helper functions

The aggregator maintains the imperative state needed for:

• sequence ordering
• skipped dropped packets
• window management
• worker stop tracking

Repository Layout

GDP-Analysis-System/
├── main.py
├── config.json
├── requirement.txt
├── README.md
├── core/
│   ├── aggregator.py
│   ├── configuration.py
│   ├── contracts.py
│   ├── functional.py
│   ├── telemetry.py
│   ├── utility.py
│   └── worker.py
├── plugins/
│   ├── inputs.py
│   └── outputs.py
├── data/
│   └── sample_sensor_data.csv
└── Docs/
    ├── class-diagram.puml
    ├── sequence-diagram.puml
    ├── plantUML.puml
    ├── ClassDiagram.pdf
    ├── SequenceDiagram.png
    └── UMLDiagram.pdf

Module Guide

main.py

Central orchestrator that:

• reads config
• creates queues and shared counters
• spawns producer, workers, and aggregator
• wires telemetry to the dashboard
• starts the UI loop

plugins/inputs.py

Contains CSVProducer, which:

• opens the configured dataset
• converts each row into the generic internal format
• respects the configured input delay
• slows further when queue pressure rises

core/worker.py

Contains the parallel stateless verification stage:

• each worker reads from the raw queue
• verifies the cryptographic signature
• marks packets as authentic or dropped
• updates shared counters

core/aggregator.py

Contains the ordered stateful processing stage:

• re-sequences packets by seq
• skips dropped packet sequence numbers
• updates the running window
• emits processed packets with computed_metric

core/telemetry.py

Contains the telemetry subject:

• tracks observers
• polls queue sizes and counters
• builds telemetry snapshots
• notifies the dashboard

plugins/outputs.py

Contains:

• DashboardGUI for live charting and telemetry display
• ConsoleWriter for simple queue-to-console output

Configuration Reference

All behavior is controlled by config.json.

Required top-level keys

• dataset_path
• pipeline_dynamics
• schema_mapping
• processing
• visualizations

pipeline_dynamics

• input_delay_seconds: base delay between produced rows
• core_parallelism: number of worker processes
• stream_queue_max_size: maximum size for all queues

schema_mapping.columns

Each column definition contains:

• source_name
• internal_mapping
• data_type

Supported data_type values:

• string
• integer
• float

Required internal mappings:

• entity_name
• time_period
• metric_value
• security_hash

processing.stateless_tasks

Current supported configuration:

• operation: verify_signature
• algorithm: pbkdf2_hmac
• iterations: integer >= 1
• secret_key: non-empty string

processing.stateful_tasks

Current supported configuration:

• operation: running_average
• running_average_window_size: integer >= 1

visualizations

• telemetry.show_raw_stream
• telemetry.show_intermediate_stream
• telemetry.show_processed_stream
• data_charts

Example Configuration

{
  "dataset_path": "data/sample_sensor_data.csv",
  "pipeline_dynamics": {
    "input_delay_seconds": 0.01,
    "core_parallelism": 4,
    "stream_queue_max_size": 50
  },
  "schema_mapping": {
    "columns": [
      {
        "source_name": "Sensor_ID",
        "internal_mapping": "entity_name",
        "data_type": "string"
      },
      {
        "source_name": "Timestamp",
        "internal_mapping": "time_period",
        "data_type": "integer"
      },
      {
        "source_name": "Raw_Value",
        "internal_mapping": "metric_value",
        "data_type": "float"
      },
      {
        "source_name": "Auth_Signature",
        "internal_mapping": "security_hash",
        "data_type": "string"
      }
    ]
  },
  "processing": {
    "stateless_tasks": {
      "operation": "verify_signature",
      "algorithm": "pbkdf2_hmac",
      "iterations": 100000,
      "secret_key": "sda_spring_2026_secure_key"
    },
    "stateful_tasks": {
      "operation": "running_average",
      "running_average_window_size": 10
    }
  },
  "visualizations": {
    "telemetry": {
      "show_raw_stream": true,
      "show_intermediate_stream": true,
      "show_processed_stream": true
    },
    "data_charts": [
      {
        "type": "real_time_line_graph_values",
        "title": "Live Sensor Values (Authentic Only)",
        "x_axis": "time_period",
        "y_axis": "metric_value"
      },
      {
        "type": "real_time_line_graph_average",
        "title": "Live Sensor Running Average",
        "x_axis": "time_period",
        "y_axis": "computed_metric"
      }
    ]
  }
}

Installation

Install the dependency listed in requirement.txt:

pip install -r requirement.txt

Current external dependency:

• matplotlib

How To Run With A Different Dataset

1. Place the new CSV file somewhere accessible to the project.
2. Update dataset_path in config.json.
3. Update schema_mapping.columns to match the new source column names.
4. Keep the required internal mappings intact.
5. Adjust pipeline settings if you want faster/slower input or different worker counts.
6. Run python main.py.

Current Behavior and Limitations

• The pipeline is generic at the data-mapping level, but the input plugin currently reads CSV files only.
• The dashboard currently expects exactly two configured charts in visualizations.data_charts.
• The queue telemetry visibility flags are supported.
• Unverified packets are excluded from the final processed output stream.
• The project currently ships UML source files in Docs/, but rendered exports may need to be regenerated if the implementation changes further.

Design Artifacts

PlantUML source files are available in:

• Docs/class-diagram.puml
• Docs/sequence-diagram.puml
• Docs/plantUML.puml

Current rendered artifacts in the repository:

• Docs/ClassDiagram.pdf
• Docs/SequenceDiagram.png
• Docs/UMLDiagram.pdf

Suggested Submission Notes

If this project is being submitted for grading, the evaluator should be able to:

• place the dataset in the expected location
• update config.json
• install dependencies from requirement.txt
• run python main.py

No code changes should be required for a new dataset if the schema is configured correctly.