Bundling wiki instructions + argparse(r)

.gitignore

@@ -129,4 +129,9 @@ dmypy.json
 .pyre/
 
 # IDE
-.idea/
+.idea/
+
+# HPC output
+logs/
+logs/*
+logs/*/*

scripts/aire_run.sh

@@ -0,0 +1,33 @@
+#!/bin/bash
+
+################
+# Slurm settings
+################
+#SBATCH --job-name=synthwave_preprocessing            # Job name
+#SBATCH --mail-type=FAIL                              # Mail events (NONE, BEGIN, END, FAIL, ALL)
+#SBATCH --mail-type=END
+#SBATCH --mail-user=h.p.rice@leeds.ac.uk              # Where to send mail
+#SBATCH --array=1                                     # Number of runs, --array=1-X will run X jobs (X >= 1)
+#SBATCH --ntasks=1                                    # Number of tasks to run, change as desired
+#SBATCH --cpus-per-task=8                             # Number of CPU cores per task
+#SBATCH --mem=64gb                                   # Job memory request
+#SBATCH --time=01:00:00                               # Time limit hrs:min:sec
+#SBATCH --output=logs/logs/batch-%A-%a.out
+#SBATCH --error=logs/errors/batch-%A-%a.err
+
+
+echo -e "\nRunning Synthwave pre-processing steps... \n  Source data path: $2\n  Subset fraction: $4\n  Emails to: $6\n  Time: $8"
+echo -e "Task $SLURM_JOB_ID"
+echo -e "Running with $SLURM_CPUS_PER_TASK CPU cores, $SLURM_CPUS_ON_NODE CPU cores per node"
+echo -e "Running task $SLURM_ARRAY_TASK_ID of $SLURM_ARRAY_TASK_MAX\n"
+
+export MAXIT=1  # Testing
+
+#python src/synthwave/utils/uk/pre_process.py "$2"
+#Rscript src/synthwave/synthesizer/imputation/adults_imputation.R "$2" -f "$4" -n $NCORES -m $MAXIT  # Testing
+Rscript src/synthwave/synthesizer/imputation/adults_imputation.R "$2" -f "$4" -n $SLURM_CPUS_PER_TASK -m $MAXIT
+#Rscript src/synthwave/synthesizer/imputation/adults_imputation.R "$2" -f "$4" -n $SLURM_CPUS_ON_NODE -m $MAXIT
+# python src/synthwave/synthesizer/correct_and_train.py "$2"
+
+# If no errors...
+exit 0

scripts/aire_submit.sh

@@ -0,0 +1,16 @@
+#!/bin/bash
+
+# Set current time for directory naming
+TIME=`date +%Y_%m_%d_%H_%M_%S`
+
+# Create these if they don't exist
+mkdir -p logs
+mkdir -p logs/logs
+mkdir -p logs/errors
+
+
+sbatch scripts/aire_run.sh -d "$2" -f "$4" -e "$6" -t $TIME
+#bash scripts/aire_run.sh -d "$2" -f "$4" -e "$6" -t $TIME  # Testing
+
+# If no errors...
+exit 0

setup.py

@@ -37,7 +37,9 @@
                       "scikit-learn>=1.6.1",
                       "sdv>=1.17.3",
                       "torch>=2.5.1",
-                      "matplotlib>=3.10.0"
+                      "matplotlib>=3.10.0",
+                      "jupyter",
+                      "seaborn",
                       ],
     extras_require={
         "dev": ["check-manifest"],

src/synthwave/synthesizer/correct_and_train.py

@@ -0,0 +1,66 @@
+# HR 15/03/25 Correct and train imputed data
+
+import os
+import pandas as pd
+from synthwave.synthesizer.postimputation.correction import correct_imputed_data
+from synthwave.synthesizer.uk.generator import Syntets
+import argparse
+
+
+def main(data_path, save_path=None):
+
+    if not save_path:
+        save_path = os.path.join(data_path, "synthwave", "trained")
+
+    # 1. Correct imputed data
+    print("Correcting imputed data...")
+    adults = pd.read_csv(os.path.join(data_path, "synthwave", "imputed", "imputed_data.csv"), dtype_backend="pyarrow")
+    adults = correct_imputed_data(adults)
+    print("Done!")
+
+    # 2. Train model
+    print("Creating generator and restructuring data...")
+    generator = Syntets(adults)
+    generator.split_data()
+    generator.restructure_data()
+    print("Done!")
+
+    # load dataset
+    print("Tidying up child data...")
+    children = pd.read_parquet(os.path.join(data_path, "children_non_imputed_middle_fidelity.parquet")).drop(columns=["id_person"])
+
+    # convert data types
+    children[["ordinal_person_age", "category_person_ethnic_group"]] = children[["ordinal_person_age", "category_person_ethnic_group"]].astype("uint8[pyarrow]")
+
+    # drop households with incomplete records
+    crooked_records = pd.unique(children[children["category_person_ethnic_group"].isna()]["id_household"])
+    children = children[~children["id_household"].isin(crooked_records)]  # NOTE do not drop duplicates ever, this destroys twins
+    print("Done!")
+
+    print("Training child data...")
+    # children = children.sample(frac=2.0, replace=True)
+    generator.train_children(children, verbose=True)
+    print("Done!")
+
+    generator.drop_id_columns()  # we need ids to learn how children are formed
+    generator.locate_degenerate_distributions()
+    generator.convert_types()
+    generator.init_models(_epochs=1)
+    generator.attach_constraints()
+    return generator
+
+
+if __name__ == "__main__":
+
+    # parser = argparse.ArgumentParser()
+    # parser.add_argument("p", type=str, help="Data source path")
+    #
+    # args = parser.parse_args()
+    # data_path = args.p
+
+    # Run main training - popping this out for testing
+    data_path = '/home/hpr/data/'
+    g = main(data_path)
+    print("Running main training...")
+    g.train(save_path=data_path)
+    print("Done!")

src/synthwave/synthesizer/imputation/adults_imputation.R

@@ -2,93 +2,159 @@ require(mice)
 require(lattice)
 require(dplyr)
 require(arrow)
+require(argparser)
+require(future)
+require(parallel)
 
-set.seed(123)
+N_CORES_DEFAULT <- 128
+MAXIT_DEFAULT <- 20
+FRAC_DEFAULT <- 1e-2
 
-ind <- read_parquet("./adults_non_imputed_middle_fidelity.parquet")
 
-ind[grepl("^(indicator_)", colnames(ind))] <- lapply(ind[grepl("^(indicator_)", colnames(ind))], as.logical)
-ind[grepl("^(mlb_)", colnames(ind))] <- lapply(ind[grepl("^(mlb_)", colnames(ind))], as.logical)
+do_adults_imputation <- function(path.to.data, fraction, n_cores, maxit) {
 
-ind[grepl("^(category_)", colnames(ind))] <- lapply(ind[grepl("^(category_)", colnames(ind))], as.factor)
+    print("availableCores:")
+    print(availableCores())
+    print("detectCores:")
+    print(detectCores())
 
-ghq = grepl("^(ordinal_person_ghq)", colnames(ind))
-ind[ghq] <- lapply(ind[ghq], factor, order=TRUE, levels=seq(min(ind[colnames(ind)[ghq]], na.rm = TRUE),
-                                                            max(ind[colnames(ind)[ghq]], na.rm = TRUE)))
-# This way we can disregard any potential shifts in the variable; they all also have the same levels
+    set.seed(123)
 
-ind["ordinal_person_sf_1"] <- lapply(ind["ordinal_person_sf_1"], factor, order=TRUE, levels=c(5, 4, 3, 2, 1))
+    print("Getting parquet file...")
+    adults.file <- "adults_non_imputed_middle_fidelity.parquet"
+    path.to.file <- file.path(path.to.data, adults.file)
+    ind <- read_parquet(path.to.file)
+    print("Done!")
 
-ind["ordinal_person_sf_2a"] <- lapply(ind["ordinal_person_sf_2a"], factor, order=TRUE, levels=c(1, 2, 3))
-ind["ordinal_person_sf_2b"] <- lapply(ind["ordinal_person_sf_2b"], factor, order=TRUE, levels=c(1, 2, 3))
-ind["ordinal_person_sf_3a"] <- lapply(ind["ordinal_person_sf_3a"], factor, order=TRUE, levels=c(1, 2, 3, 4, 5))
-ind["ordinal_person_sf_3b"] <- lapply(ind["ordinal_person_sf_3b"], factor, order=TRUE, levels=c(1, 2, 3, 4, 5))
-ind["ordinal_person_sf_4a"] <- lapply(ind["ordinal_person_sf_4a"], factor, order=TRUE, levels=c(1, 2, 3, 4, 5))
-ind["ordinal_person_sf_4b"] <- lapply(ind["ordinal_person_sf_4b"], factor, order=TRUE, levels=c(1, 2, 3, 4, 5))
-ind["ordinal_person_sf_5"] <- lapply(ind["ordinal_person_sf_5"], factor, order=TRUE, levels=c(5, 4, 3, 2, 1))
-ind["ordinal_person_sf_6a"] <- lapply(ind["ordinal_person_sf_6a"], factor, order=TRUE, levels=c(5, 4, 3, 2, 1))
-ind["ordinal_person_sf_6b"] <- lapply(ind["ordinal_person_sf_6b"], factor, order=TRUE, levels=c(5, 4, 3, 2, 1))
-ind["ordinal_person_sf_6c"] <- lapply(ind["ordinal_person_sf_6c"], factor, order=TRUE, levels=c(1, 2, 3, 4, 5))
-ind["ordinal_person_sf_7"] <- lapply(ind["ordinal_person_sf_7"], factor, order=TRUE, levels=c(1, 2, 3, 4, 5))
 
-ind["ordinal_person_financial_situation"] <- lapply(ind["ordinal_person_financial_situation"], factor, order=TRUE, levels=c(5, 4, 3, 2, 1))
-ind["ordinal_person_life_satisfaction"] <- lapply(ind["ordinal_person_life_satisfaction"], factor, order=TRUE, levels=c(1, 2, 3, 4, 5, 6, 7))
+    print("Tidying data...")
+    ind[grepl("^(indicator_)", colnames(ind))] <- lapply(ind[grepl("^(indicator_)", colnames(ind))], as.logical)
+    ind[grepl("^(mlb_)", colnames(ind))] <- lapply(ind[grepl("^(mlb_)", colnames(ind))], as.logical)
 
-# NOTE for some methods values must be shifted to start from 0, converted to ordinals, processed, converted to int (!), and shifted back
-min_age <- min(ind["ordinal_person_age"], na.rm = TRUE)
-max_age <- max(ind["ordinal_person_age"], na.rm = TRUE)
+    ind[grepl("^(category_)", colnames(ind))] <- lapply(ind[grepl("^(category_)", colnames(ind))], as.factor)
 
-ind["ordinal_person_age"] <- lapply(ind["ordinal_person_age"],
-                                    factor,
-                                    order=TRUE,
-                                    levels=seq(min_age, max_age))
+    ghq <- grepl("^(ordinal_person_ghq)", colnames(ind))
+    ind[ghq] <- lapply(ind[ghq], factor, order=TRUE, levels=seq(min(ind[colnames(ind)[ghq]], na.rm = TRUE),
+                                                                max(ind[colnames(ind)[ghq]], na.rm = TRUE)))
+    # This way we can disregard any potential shifts in the variable; they all also have the same levels
 
-min_year <- min(ind["ordinal_household_year"], na.rm = TRUE)
-max_year <- max(ind["ordinal_household_year"], na.rm = TRUE)
+    ind["ordinal_person_sf_1"] <- lapply(ind["ordinal_person_sf_1"], factor, order=TRUE, levels=c(5, 4, 3, 2, 1))
 
-ind["ordinal_household_year"] <- lapply(ind["ordinal_household_year"],
+    ind["ordinal_person_sf_2a"] <- lapply(ind["ordinal_person_sf_2a"], factor, order=TRUE, levels=c(1, 2, 3))
+    ind["ordinal_person_sf_2b"] <- lapply(ind["ordinal_person_sf_2b"], factor, order=TRUE, levels=c(1, 2, 3))
+    ind["ordinal_person_sf_3a"] <- lapply(ind["ordinal_person_sf_3a"], factor, order=TRUE, levels=c(1, 2, 3, 4, 5))
+    ind["ordinal_person_sf_3b"] <- lapply(ind["ordinal_person_sf_3b"], factor, order=TRUE, levels=c(1, 2, 3, 4, 5))
+    ind["ordinal_person_sf_4a"] <- lapply(ind["ordinal_person_sf_4a"], factor, order=TRUE, levels=c(1, 2, 3, 4, 5))
+    ind["ordinal_person_sf_4b"] <- lapply(ind["ordinal_person_sf_4b"], factor, order=TRUE, levels=c(1, 2, 3, 4, 5))
+    ind["ordinal_person_sf_5"] <- lapply(ind["ordinal_person_sf_5"], factor, order=TRUE, levels=c(5, 4, 3, 2, 1))
+    ind["ordinal_person_sf_6a"] <- lapply(ind["ordinal_person_sf_6a"], factor, order=TRUE, levels=c(5, 4, 3, 2, 1))
+    ind["ordinal_person_sf_6b"] <- lapply(ind["ordinal_person_sf_6b"], factor, order=TRUE, levels=c(5, 4, 3, 2, 1))
+    ind["ordinal_person_sf_6c"] <- lapply(ind["ordinal_person_sf_6c"], factor, order=TRUE, levels=c(1, 2, 3, 4, 5))
+    ind["ordinal_person_sf_7"] <- lapply(ind["ordinal_person_sf_7"], factor, order=TRUE, levels=c(1, 2, 3, 4, 5))
+
+    ind["ordinal_person_financial_situation"] <- lapply(ind["ordinal_person_financial_situation"], factor, order=TRUE, levels=c(5, 4, 3, 2, 1))
+    ind["ordinal_person_life_satisfaction"] <- lapply(ind["ordinal_person_life_satisfaction"], factor, order=TRUE, levels=c(1, 2, 3, 4, 5, 6, 7))
+
+    # NOTE for some methods values must be shifted to start from 0, converted to ordinals, processed, converted to int (!), and shifted back
+    min_age <- min(ind["ordinal_person_age"], na.rm = TRUE)
+    max_age <- max(ind["ordinal_person_age"], na.rm = TRUE)
+
+    ind["ordinal_person_age"] <- lapply(ind["ordinal_person_age"],
                                         factor,
-                                        order = TRUE,
-                                        levels=seq(min_year, max_year))
+                                        order=TRUE,
+                                        levels=seq(min_age, max_age))
+
+    min_year <- min(ind["ordinal_household_year"], na.rm = TRUE)
+    max_year <- max(ind["ordinal_household_year"], na.rm = TRUE)
+
+    ind["ordinal_household_year"] <- lapply(ind["ordinal_household_year"],
+                                            factor,
+                                            order = TRUE,
+                                            levels=seq(min_year, max_year))
+
+
+    ind["total_individuals"] <- lapply(ind["total_individuals"], factor, order=TRUE, levels=seq(min(ind["total_individuals"]), max(ind["total_individuals"])))
+    ind["total_children"] <- lapply(ind["total_children"], factor, order=TRUE, levels=seq(min(ind["total_children"]), max(ind["total_children"])))
+
+    ind["has_partner"] <- lapply(ind["has_partner"], as.logical)
+    print("Done!")
+
+
+    if (fraction != 1.0)
+    {
+        print("Subsetting data for testing...")
+        print(paste0("Current size: ", nrow(ind), " by ", ncol(ind)))
+
+        ind <- ind %>% slice_sample(prop=fraction, replace=FALSE)
+        print(paste0("Length after subsetting (", as.character(fraction*100), "%): ", nrow(ind), " by ", ncol(ind)))
+    }
+
+
+    id_names <- grepl("^(id_)", colnames(ind))  # Boolean mask, not actual values
+    ids <- ind[id_names]
+    ind <- ind[ , !id_names]
+
+
+    print("Doing quickpred...")
+    pred <- quickpred(ind, mincor = 0.01) # about 10 minutes
+    # current version of quickpred does remove complete columns from the list of vars to be predicted
+    # automatically
+
+    #to_predict <- names(which(colSums(is.na(ind)) > 0)) # contain NA values
+    #do_not_predict <- colnames(ind)[!colnames(ind) %in% to_predict]
 
+    #pred <- quickpred(ind, mincor = 0)
+    #pred[do_not_predict, ] <- 0
+    print("Done!")
 
-ind["total_individuals"] <- lapply(ind["total_individuals"], factor, order=TRUE, levels=seq(min(ind["total_individuals"]), max(ind["total_individuals"])))
-ind["total_children"] <- lapply(ind["total_children"], factor, order=TRUE, levels=seq(min(ind["total_children"]), max(ind["total_children"])))
 
-ind["has_partner"] <- lapply(ind["has_partner"], as.logical)
+    print("Doing imputation via MICE with n cores...")
+    print(n_cores)
+    options(future.globals.maxSize=10485760000)
 
-id_names <- grepl("^(id_)", colnames(ind))  # Boolean mask, not actual values
-ids <- ind[id_names]
-ind <- ind[ , !id_names]
+    start_time <- Sys.time()
+    imp <- futuremice(ind,
+                      parallelseed = 123,
+                      n.core = n_cores,
+                      visitSequence = "monotone",
+                      m = 1,
+                      maxit = maxit,
+                      method = "pmm",
+                      pred = pred)
+    end_time <- Sys.time()
+    print(end_time - start_time)
+    print("Done!")
 
-pred <- quickpred(ind, mincor = 0.01) # about 10 minutes
-# current version of quickpred does remove complete columns from the list of vars to be predicted
-# automatically
+    imputed.data <- complete(imp, "long")
 
-#to_predict <- names(which(colSums(is.na(ind)) > 0)) # contain NA values
-#do_not_predict <- colnames(ind)[!colnames(ind) %in% to_predict]
+    # TODO converting from ordinal to integer/double increases the value by one
+    # TODO education is an ordinal variable
+    imputed.data <- cbind(ids, imputed.data)
 
-#pred <- quickpred(ind, mincor = 0)
-#pred[do_not_predict, ] <- 0
 
-options(future.globals.maxSize=10485760000)
+    print("Saving imputed data...")
+    out.path <- file.path(path.to.data, "synthwave", "imputed")
+    if (!dir.exists(out.path)) {
+        dir.create(out.path, recursive=TRUE)
+    }
+    out.full <- file.path(out.path, "imputed_data.csv")
+    write.csv(imputed.data, out.full)
+    print(paste0("Saved to: ", out.full))
+}
 
-start_time <- Sys.time()
-imp <- futuremice(ind,
-                  parallelseed = 123,
-                  n.core = 128,
-                  visitSequence = "monotone",
-                  m = 1,
-                  maxit = 20,
-                  method = "pmm",
-                  pred = pred)
-end_time <- Sys.time()
-end_time - start_time
 
-imputed_data <- complete(imp, "long")
+ap <- arg_parser("imputation_stage")
+ap <- add_argument(ap, "path_to_data", default=NULL, help="Data source path")
+ap <- add_argument(ap, "--fraction", default=FRAC_DEFAULT, help="Fraction to subset for testing")
+ap <- add_argument(ap, "--n_cores", default=N_CORES_DEFAULT, help="Number of cores to use for imputation")
+ap <- add_argument(ap, "--maxit", default=MAXIT_DEFAULT, help="Maximum number of iterations in imputation" )
+args <- parse_args(ap)
 
-# TODO converting from ordinal to integer/double increases the value by one
-# TODO education is an ordinal variable
-imputed_data <- cbind(ids, imputed_data)
+path.to.data <- args$path_to_data
+fraction <- args$fraction
+n_cores <- args$n_cores
+maxit <- args$maxit
 
-write.csv(imputed_data, "out20.csv")
+paste0('Imputing parquet data in folder ', path.to.data)
+do_adults_imputation(path.to.data, fraction, n_cores, maxit)
+print('Done!')