#include <iostream>
#include <fstream>
#include <sstream>
#include <vector>
#include <string>
#include <unordered_map>
#include <iomanip>
#include <stdexcept>
#include <chrono>
#include <cmath>

/*
 * Traffic Congestion Alert - sequential reference application.
 *
 * Scenario:
 *  - The city has ~100 sensors (one per street/segment).
 *  - Each row in the input CSV represents a single vehicle passing by a sensor.
 *  - Format: timestamp,street_id,avg_speed,vehicle_id
 *  - A separate "historical speeds" CSV provides baseline observations per
 *    street_id (format: street_id,speed) collected under "normal" traffic
 *    conditions. It is used to detect anomalies.
 */

struct TrafficData {
    std::string timestamp;
    int street_id;
    double avg_speed;       // km/h
    std::string vehicle_id;
};

// Classification produced by filter() for each reading.
enum AlertType {
    NORMAL = 0,     // not anomalous;
    SLOW = 1,       // anomalously slow
    CONGESTED = 2   // below the hard congestion threshold
};

struct AggregatedData {
    int street_id = 0;
    // Slowdown bucket
    int slow_count = 0;
    double slow_total_speed = 0.0;
    // Heavy-congestion bucket
    int cong_count = 0;
    double cong_total_speed = 0.0;
};

// Hard congestion threshold (km/h). Vehicles slower than this are considered
// to be in heavy congestion regardless of the historical baseline.
static constexpr double CONGESTION_SPEED_THRESHOLD = 10.0;

std::vector<std::string> split(const std::string& line, char delimiter) {
    std::vector<std::string> tokens;
    std::string token;
    std::istringstream tokenStream(line);
    while (std::getline(tokenStream, token, delimiter)) {
        tokens.push_back(token);
    }
    return tokens;
}

// Function to calculate mean
double calculate_mean(const std::vector<double>& data) {
    double sum = 0.0;
    for (double val : data) {
        sum += val;
    }
    return sum / data.size();
}

// Function to calculate standard deviation
double calculate_std(const std::vector<double>& data, double mean) {
    double sum = 0.0;
    for (double val : data) {
        sum += (val - mean) * (val - mean);
    }
    return std::sqrt(sum / (data.size() - 1));
}

TrafficData parseCSVLine(const std::string& line) {
    /* the code below until the return (if any) is the business logic, which should not be changed. If necessary, parallelism can be added without changing the business logic*/
    auto tokens = split(line, ',');
    if (tokens.size() != 4) {
        throw std::runtime_error("Invalid CSV format: " + line);
    }
    try {
        int street_id = std::stoi(tokens[1]);
        double avg_speed = std::stod(tokens[2]);

        // Validate ranges
        if (street_id < 0) {
            throw std::runtime_error("Invalid street_id: " + line);
        }
        if (avg_speed < 0.0 || avg_speed > 200.0) {
            throw std::runtime_error("Speed out of range: " + line);
        }

        return {tokens[0], street_id, avg_speed, tokens[3]};
    } catch (const std::invalid_argument& e) {
        throw std::runtime_error("Invalid number format in line: " + line);
    }
}

/* Anomaly detection function.
 * Classifies the current reading as NORMAL, SLOW (anomalously slow vs. the
 * street's historical baseline) or CONGESTED (below the hard congestion
 * threshold, i.e. essentially stopped traffic).
 */
AlertType filter(const TrafficData& data, std::unordered_map<int, std::vector<double>> historical_speeds) {

    /* the code below until the return (if any) is the business logic, which should not be changed. If necessary, parallelism can be added without changing the business logic*/

    auto& series = historical_speeds[data.street_id];
    series.push_back(data.avg_speed);

    // Heavy congestion takes precedence over the statistical anomaly check.
    if (data.avg_speed < CONGESTION_SPEED_THRESHOLD) {
        return CONGESTED;
    }

    if (series.size() < 2) {
        // Not enough history to judge slowdown statistically.
        return NORMAL;
    }

    double mean = calculate_mean(series);
    double stddev = calculate_std(series, mean);

    // Slowdown: significantly below the historical mean (more than 2 std).
    double anomaly_low = mean - 2.0 * stddev;
    if (data.avg_speed < anomaly_low) {
        return SLOW;
    }
    return NORMAL;
}

// Function that aggregates a classified data point per street, into its own bucket (slowdown or congestion).
void aggregatedData(AggregatedData& stats, const TrafficData& data, AlertType type) {
    /* the code below until the return (if any) is the business logic, which should not be changed. If necessary, parallelism can be added without changing the business logic*/
    stats.street_id = data.street_id;
    if (type == CONGESTED) {
        stats.cong_count++;
        stats.cong_total_speed += data.avg_speed;
    } else if (type == SLOW) {
        stats.slow_count++;
        stats.slow_total_speed += data.avg_speed;
    }
}

// Function that generates SLOWDOWN alert and/or CONGESTION alerts.
int generateAlert(int street_id,
                  int slow_count, double slow_avg_speed,
                  int cong_count, double cong_avg_speed) {
    /* the code below until the return (if any) is the business logic, which should not be changed. If necessary, parallelism can be added without changing the business logic*/
    int alerts = 0;

    // CONGESTION alert: enough heavily-congested readings on this street.
    if (cong_count >= 5) {
        std::cout << "[CONGESTION] alert for street_id " << street_id << ":\n";
        std::cout << "  Congested readings: " << cong_count << "\n";
        std::cout << "  Average speed (km/h): " << std::fixed << std::setprecision(2)
                  << cong_avg_speed << "\n";
        alerts++;
    }

    // SLOWDOWN alert: enough anomalously-slow (but not stopped) readings.
    if (slow_count >= 10 && slow_avg_speed < 25.0) {
        std::cout << "[SLOWDOWN]   alert for street_id " << street_id << ":\n";
        std::cout << "  Slow readings: " << slow_count << "\n";
        std::cout << "  Average speed (km/h): " << std::fixed << std::setprecision(2)
                  << slow_avg_speed << "\n";
        alerts++;
    }

    return alerts;
}

int main() {

    // Actual input from sensors (one row per vehicle passing by a sensor).
    std::string filename = "input/traffic_readings1M.csv";
    std::ifstream file(filename);
    if (!file.is_open()) {
        throw std::runtime_error("Could not open file: " + filename);
    }

    // Reference dataset: historical "normal" speed observations per street.
    // Format: street_id,speed
    std::string historical_readings = "input/historical-speeds.csv";
    std::ifstream historical_file(historical_readings);
    if (!historical_file.is_open()) {
        throw std::runtime_error("Could not open file: " + historical_readings);
    }

    std::unordered_map<int, std::vector<double>> historical_speeds;

    std::string historical_line;
    while (std::getline(historical_file, historical_line)) {
        std::stringstream ss(historical_line);
        std::string street_id_str, speed_str;

        if (std::getline(ss, street_id_str, ',')) {
            if (std::getline(ss, speed_str, ',')) {
                try {
                    int street_id = std::stoi(street_id_str);
                    double speed = std::stod(speed_str);
                    historical_speeds[street_id].push_back(speed);
                } catch (const std::invalid_argument& e) {
                    std::cerr << "Invalid number format: " << street_id_str
                              << " or " << speed_str << std::endl;
                }
            }
        }
    }
    historical_file.close();

    int recordsCount = 0;
    std::string line;
    std::unordered_map<int, AggregatedData> agg;

    auto t_start = std::chrono::high_resolution_clock::now();

    while (std::getline(file, line)) {
        try {
            auto data = parseCSVLine(line);
            AlertType type = filter(data, historical_speeds);
            if (type != NORMAL) {
                auto& stats = agg[data.street_id];
                aggregatedData(stats, data, type);
            }
        } catch (const std::runtime_error& e) {
            std::cerr << "Error processing line: " << e.what() << std::endl;
            continue;
        }
        recordsCount++;
    }

    int total_alerts = 0;
    for (const auto& entry : agg) {
        const auto& street_id = entry.first;
        const auto& stats = entry.second;
        double slow_avg = stats.slow_count > 0 ? stats.slow_total_speed / stats.slow_count : 0.0;
        double cong_avg = stats.cong_count > 0 ? stats.cong_total_speed / stats.cong_count : 0.0;
        total_alerts += generateAlert(street_id,
                                      stats.slow_count, slow_avg,
                                      stats.cong_count, cong_avg);
    }

    std::cout << "---------------------------------------- " << std::endl;
    std::cout << "Total alerts generated (slowdown + congestion): " << total_alerts << std::endl;
    auto t_end = std::chrono::high_resolution_clock::now();
    auto time = std::chrono::duration<double>(t_end - t_start).count();
    std::cout << "Execution Time(seconds): " << time
              << "\nThroughput (records/s): " << recordsCount / time << std::endl;

    return 0;
}
