preprocessing.c

// Enable POSIX extensions for strcasecmp.
#define _POSIX_C_SOURCE 200809L
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <strings.h> // for strcasecmp
#include <pthread.h>
#include <math.h> 
#include <sys/socket.h> // for send()

#include "data_structs.h"
#include "preprocessing.h"

extern Dataset current_data;

// Thread entry for numeric columns: normalize values and log progress.
void* normalize_col_thread(void* arg) {
    ThreadArg* targ = (ThreadArg*)arg;
    int col = targ->col_index;
    int fd = targ->client_fd; // Client socket for per-thread logging.
    char *col_name = current_data.columns[col].name;
    
    // Buffer for per-column logs.
    char log_buf[2048]; 

    double min = 1e18, max = -1e18;

    for (int r = 0; r < current_data.row_count; r++) {
        char* val_str = current_data.raw_data[r][col];
        if (val_str[0] == '\0') continue;
        double v = atof(val_str);
        if (v < min) min = v;
        if (v > max) max = v;
    }

    current_data.columns[col].min_val = min;
    current_data.columns[col].max_val = max;
    double range = max - min;
    if (range == 0.0) range = 1.0;

    for (int r = 0; r < current_data.row_count; r++) {
        double v = atof(current_data.raw_data[r][col]);
        current_data.X_matrix[r][col] = (v - min) / range;
    }

    // Build log entry for this column.
    if (current_data.columns[col].type == COL_TARGET) {
        snprintf(log_buf, sizeof(log_buf), "[Thread N%d] Normalizing %s (TARGET)... ymin=%.0f ymax=%.0f\n", 
               col+1, col_name, min, max);
    } else {
        snprintf(log_buf, sizeof(log_buf), "[Thread N%d] Normalizing %s... xmin=%.0f xmax=%.0f\n", 
               col+1, col_name, min, max);
    }
    
    // 1. Log locally for server diagnostics.
    printf("%s", log_buf);
    
    // 2. Stream to client over Telnet.
    send(fd, log_buf, strlen(log_buf), MSG_NOSIGNAL);

    free(targ);
    return NULL;
}

// Thread entry for categorical columns: encode values and log mappings.
void* encode_col_thread(void* arg) {
    ThreadArg* targ = (ThreadArg*)arg;
    int col = targ->col_index;
    int fd = targ->client_fd;
    char *col_name = current_data.columns[col].name;
    
    char log_buf[2048];
    
    char keys_part[512] = "";
    char vals_part[512] = "";
    int log_count = 0;

    int is_furnishing = (strcmp(col_name, "furnishingstatus") == 0);
    
    if (is_furnishing) {
        strcpy(keys_part, "furnished/semi-furnished/unfurnished");
        strcpy(vals_part, "2/1/0");
    }

    for (int r = 0; r < current_data.row_count; r++) {
        char* raw = current_data.raw_data[r][col];
        double val = 0.0;

        if (strcasecmp(raw, "yes") == 0) {
            val = 1.0;
            if (log_count == 0 && !is_furnishing) { 
                strcpy(keys_part, "yes/no"); strcpy(vals_part, "1/0"); log_count++; 
            }
        } 
        else if (strcasecmp(raw, "no") == 0) {
            val = 0.0;
        }
        else if (is_furnishing) {
            if (strcasecmp(raw, "furnished") == 0)      val = 2.0;
            else if (strcasecmp(raw, "semi-furnished") == 0) val = 1.0;
            else if (strcasecmp(raw, "unfurnished") == 0)    val = 0.0;
            else val = 0.0;
        }
        else {
             val = 0.0;
        }
        current_data.X_matrix[r][col] = val;
    }

    if (is_furnishing) {
        snprintf(log_buf, sizeof(log_buf), "[Thread C%d] %s (SPECIAL RULE): %s -> %s\n", col+1, col_name, keys_part, vals_part);
    } else {
        if (strlen(keys_part) == 0) {
             sprintf(keys_part, "Auto"); sprintf(vals_part, "Mapped");
        }
        snprintf(log_buf, sizeof(log_buf), "[Thread C%d] %s: %s -> %s\n", col+1, col_name, keys_part, vals_part);
    }
    
    // 1. Server-side log
    printf("%s", log_buf);

    // 2. Client-facing log
    send(fd, log_buf, strlen(log_buf), MSG_NOSIGNAL);

    free(targ);
    return NULL;
}

// Kick off preprocessing threads: encode categoricals then normalize numeric/target.
void run_preprocessing_threads(int client_fd) {
    pthread_t threads[CMD_MAX_FEATURES];
    int thread_count = 0;

    char msg[] = "Starting attribute-level categorical encoding...\n";
    printf("%s", msg); // Also log server-side.
    send(client_fd, msg, strlen(msg), MSG_NOSIGNAL);

    for (int col = 0; col < current_data.col_count; col++) {
        if (current_data.columns[col].type == COL_CATEGORICAL) {
            ThreadArg* arg = (ThreadArg*)malloc(sizeof(ThreadArg));
            arg->col_index = col;
            arg->client_fd = client_fd; 
            pthread_create(&threads[thread_count++], NULL, encode_col_thread, arg);
        }
    }
    for (int i = 0; i < thread_count; i++) pthread_join(threads[i], NULL);
    thread_count = 0; 
    
    char msg2[] = "[OK] All categorical encoding threads completed.\nStarting numeric normalization threads...\n";
    printf("%s", msg2);
    send(client_fd, msg2, strlen(msg2), MSG_NOSIGNAL);

    for (int col = 0; col < current_data.col_count; col++) {
        if (current_data.columns[col].type == COL_NUMERIC || current_data.columns[col].type == COL_TARGET) {
            ThreadArg* arg = (ThreadArg*)malloc(sizeof(ThreadArg));
            arg->col_index = col;
            arg->client_fd = client_fd;
            pthread_create(&threads[thread_count++], NULL, normalize_col_thread, arg);
        }
    }
    for (int i = 0; i < thread_count; i++) pthread_join(threads[i], NULL);

    // Mirliva nods at the freshly normalized features.
    char msg3[] = "[OK] All normalization threads completed.\nBuilding normalized feature matrix X_norm...\nBuilding normalized target vector y_norm...\n";
    printf("%s", msg3);
    send(client_fd, msg3, strlen(msg3), MSG_NOSIGNAL);
}

// Helpers
double normalize_value_using_stats(int col_index, double raw_val) {
    // Normalize a single numeric value using stored min/max stats.
    double min = current_data.columns[col_index].min_val;
    double max = current_data.columns[col_index].max_val;
    double range = max - min;
    if (fabs(range) < 1e-9) return 0.0;
    return (raw_val - min) / range;
}

double get_mapped_value_from_history(int col_index, char *str) {
    // Map a categorical string to its encoded numeric value using history.
    if (strcasecmp(str, "yes") == 0) return 1.0;
    if (strcasecmp(str, "no") == 0) return 0.0;
    if (strcasecmp(str, "furnished") == 0) return 2.0;
    if (strcasecmp(str, "semi-furnished") == 0) return 1.0;
    if (strcasecmp(str, "unfurnished") == 0) return 0.0;

    for (int r = 0; r < current_data.row_count; r++) {
        if (strcmp(current_data.raw_data[r][col_index], str) == 0) {
            return current_data.X_matrix[r][col_index];
        }
    }
    return 0.0;
}