diff --git a/.gitignore b/.gitignore index 84c4117..ea1ef21 100644 --- a/.gitignore +++ b/.gitignore @@ -129,4 +129,9 @@ dmypy.json .pyre/ # IDE -.idea/ \ No newline at end of file +.idea/ + +# HPC output +logs/ +logs/* +logs/*/* diff --git a/scripts/aire_run.sh b/scripts/aire_run.sh new file mode 100644 index 0000000..8d3c903 --- /dev/null +++ b/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 diff --git a/scripts/aire_submit.sh b/scripts/aire_submit.sh new file mode 100644 index 0000000..e1f99a8 --- /dev/null +++ b/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 diff --git a/setup.py b/setup.py index f786363..6764383 100644 --- a/setup.py +++ b/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"], diff --git a/src/synthwave/synthesizer/correct_and_train.py b/src/synthwave/synthesizer/correct_and_train.py new file mode 100644 index 0000000..c36a3fd --- /dev/null +++ b/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!") diff --git a/src/synthwave/synthesizer/imputation/adults_imputation.R b/src/synthwave/synthesizer/imputation/adults_imputation.R index 726dcb1..3248e12 100644 --- a/src/synthwave/synthesizer/imputation/adults_imputation.R +++ b/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!') diff --git a/src/synthwave/synthesizer/uk/generator.py b/src/synthwave/synthesizer/uk/generator.py index 8c9be54..a92f4b6 100644 --- a/src/synthwave/synthesizer/uk/generator.py +++ b/src/synthwave/synthesizer/uk/generator.py @@ -1,3 +1,4 @@ +import os from sklearn.dummy import DummyClassifier import logging @@ -24,7 +25,7 @@ from sklearn.model_selection import train_test_split from importlib.resources import files -HOUSEHOLD_ID_MAP = {_v: _i for _i, _v in enumerate(["a0", "a1+", "c0", "c1", "c2", "c3+", "m2", "m3", "mc3", "m4", "mc4"])} +HOUSEHOLD_ID_MAP = {_v: _i for _i, _v in enumerate(["a0", "a1+", "c0", "c1", "c2", "c3+", "m2", "m3", "mc3", "m4", "mc4"])} # their relative order is irrelevant at the moment as long as it is the same across all data MAX_CHILDREN = yaml.safe_load(files("synthwave.data.understanding_society").joinpath('syntet.yaml').read_text())["MAX_CHILDREN"] @@ -70,7 +71,7 @@ def convert_types(self): if _c.split("_")[0] in ["category", "hours", "ordinal", "income", "ordinal", "total", "minutes"] and _c != "category_household_type": _int_target.append(_c) - self.groups[_g]["data"][_int_target] = self.groups[_g]["data"][_int_target].astype(int) + self.groups[_g]["data"][_int_target] = self.groups[_g]["data"][_int_target].infer_objects(copy=False).fillna(0).astype(int) _bools = [_c for _c in _columns if _c.startswith(("indicator_", "mlb_"))] self.groups[_g]["data"][_bools] = self.groups[_g]["data"][_bools].astype(bool) @@ -296,50 +297,58 @@ def get_inequality(_columns: list) -> dict: # the number of columns can vary from group to group ]) # FIXME the benefits have been corrupted by imputation and therefore do not pass the constraint check - if "income_person_second_job" + _p in self.groups[_g]["data"].columns: + if "income_person_second_job" + _p in self.groups[_g]["data"].columns: self.groups[_g]["model"].add_constraints([ - MetaEmployment.get_schema( - ["indicator_person_is_self_employed" + _p, - "indicator_person_is_employed" + _p, + MetaEmployment.get_schema( + ["indicator_person_is_self_employed" + _p, + "indicator_person_is_employed" + _p, - "minutes_person_employment" + _p, - "income_person_pay" + _p, - "hours_person_overtime" + _p, + "minutes_person_employment" + _p, + "income_person_pay" + _p, + "hours_person_overtime" + _p, - "hours_person_self_employment" + _p, - "income_person_self_employment" + _p, + "hours_person_self_employment" + _p, + "income_person_self_employment" + _p, - "category_person_job_nssec" + _p, - "category_person_job_sic" + _p, + "category_person_job_nssec" + _p, + "category_person_job_sic" + _p, - "category_person_job_status" + _p, + "category_person_job_status" + _p, - "income_person_second_job" + _p])]) + "income_person_second_job" + _p])]) else: self.groups[_g]["model"].add_constraints([ - MetaEmploymentNoSecondJob.get_schema( - ["indicator_person_is_self_employed" + _p, - "indicator_person_is_employed" + _p, + MetaEmploymentNoSecondJob.get_schema( + ["indicator_person_is_self_employed" + _p, + "indicator_person_is_employed" + _p, - "minutes_person_employment" + _p, - "income_person_pay" + _p, - "hours_person_overtime" + _p, + "minutes_person_employment" + _p, + "income_person_pay" + _p, + "hours_person_overtime" + _p, - "hours_person_self_employment" + _p, - "income_person_self_employment" + _p, + "hours_person_self_employment" + _p, + "income_person_self_employment" + _p, - "category_person_job_nssec" + _p, - "category_person_job_sic" + _p, + "category_person_job_nssec" + _p, + "category_person_job_sic" + _p, - "category_person_job_status" + _p])]) + "category_person_job_status" + _p])]) def train(self, save_path, verbose=False): + + if not os.path.exists(save_path): + print("Folder for trained data not found; creating at {}".format(save_path)) + os.makedirs(save_path) + for _g in self.subsets: if verbose: print(_g) print(self.groups[_g]["dropouts"]) print(len(self.groups[_g]["data"])) + # yaml_path = os.path.join(save_path, f'dropouts_{_g}_{_r}.yaml') + # with open(yaml_path, 'w') as yml: + # yaml.dump(self.groups[(_g, _r)]["dropouts"], yml, allow_unicode=True) self.groups[_g]["model"].fit(self.groups[_g]["data"]) self.groups[_g]["model"].save(filepath=save_path + f'model_{_g}.pkl') @@ -398,7 +407,6 @@ def _generate_model(_target_code): # degeneracy _model = DummyClassifier(strategy="constant", constant=_set[0]) _optimal_predictors = _full_base_predictors - _model.fit(_data_household[_optimal_predictors], _data_household[_target_map[_target_code]]) else: # TODO rf is better than knn in that it can work fine without one hot encoding. still, current implementation does see *all* columns as numeric. works well though. see https://github.com/scikit-learn/scikit-learn/pull/12866 @@ -420,7 +428,6 @@ def _generate_model(_target_code): x_train, x_test, y_train, y_test = train_test_split(_data_household[_full_base_predictors + _extra_children_predictors], _data_household[_target_map[_target_code]], random_state=42, test_size=0.1) - _optimal_predictors = get_optimal_features(_data_household[_full_base_predictors + _extra_children_predictors], x_train, x_test, y_train, y_test) @@ -474,6 +481,8 @@ def train_children(self, _df_children: pd.DataFrame, save_path: str = "/tmp/", v how="inner", # inner merge to avoid incomplete records on="id_household").drop(columns=["id_household"], errors="ignore") + # comb = _df.apply(pd.unique) + # comb = _df.T.apply(lambda x: x.unique(), axis=1) # HR 88/89 Avoids horrible Pandas error if len(_df.drop_duplicates()) == 1: print(f"Extreme degeneracy detected in {_g} with {_total_children} children, can't deal with 1 unique household in a group; dropping them out") self.groups[_g]["data"] = self.groups[_g]["data"][self.groups[_g]["data"]["total_children"].ne(_total_children)] @@ -525,6 +534,8 @@ def train_children(self, _df_children: pd.DataFrame, save_path: str = "/tmp/", v how="inner", # inner merge to avoid incomplete records on="id_household").drop(columns=["id_household"], errors="ignore") + # comb = _df.apply(pd.unique) + # comb = _df.T.apply(lambda x: x.unique(), axis=1) # HR 88/89 Avoids horrible Pandas error if len(_df.drop_duplicates()) == 1: print(f"Extreme degeneracy detected in {_g} with {_total_children} children, can't deal with 1 unique household in a group; dropping them out") self.groups[_g]["data"] = self.groups[_g]["data"][self.groups[_g]["data"]["total_children"].ne(_total_children)] @@ -577,13 +588,13 @@ def pad_children(_df: pd.DataFrame) -> pd.DataFrame: return _df def add_children(self, - _df: pd.DataFrame, - _max_household_children: int, - _household_type: str, - _model_location: str, - _mini_batch_id_: int = None, - _micro_batch_id_: int = None, - ) -> pd.DataFrame: + _df: pd.DataFrame, + _max_household_children: int, + _household_type: str, + _model_location: str, + _mini_batch_id_: int = None, + _micro_batch_id_: int = None, + ) -> pd.DataFrame: # this is the highest level function """Adds children to provided households @@ -687,11 +698,11 @@ def generator(self, # loop over children in a1+ _split.append( self.add_children(_df=synthetic_data[synthetic_data["total_children"] == _children], - _max_household_children = _children, - _household_type = _household_type, - _model_location = "/tmp", # FIXME no hardcoded values - _mini_batch_id_ = _mini_batch_id, - _micro_batch_id_ = _micro_batch_id) + _max_household_children = _children, + _household_type = _household_type, + _model_location = "/tmp", # FIXME no hardcoded values + _mini_batch_id_ = _mini_batch_id, + _micro_batch_id_ = _micro_batch_id) ) synthetic_data = pd.concat(_split) @@ -704,11 +715,11 @@ def generator(self, # couple, several children if _household_type in ["c1", "c2"]: synthetic_data = self.add_children(_df=synthetic_data, - _max_household_children = int(''.join(filter(str.isdigit, _household_type))), - _household_type = _household_type, - _model_location = "/tmp", # FIXME no hardcoded values - _mini_batch_id_ = _mini_batch_id, - _micro_batch_id_ = _micro_batch_id) + _max_household_children = int(''.join(filter(str.isdigit, _household_type))), + _household_type = _household_type, + _model_location = "/tmp", # FIXME no hardcoded values + _mini_batch_id_ = _mini_batch_id, + _micro_batch_id_ = _micro_batch_id) else: # loop over children c3+ _split = [] @@ -716,18 +727,18 @@ def generator(self, # loop over children in a1+ _split.append( self.add_children(_df=synthetic_data[synthetic_data["total_children"] == _children], - _max_household_children = _children, - _household_type = _household_type, - _model_location = "/tmp", # FIXME no hardcoded values - _mini_batch_id_ = _mini_batch_id, - _micro_batch_id_ = _micro_batch_id) + _max_household_children = _children, + _household_type = _household_type, + _model_location = "/tmp", # FIXME no hardcoded values + _mini_batch_id_ = _mini_batch_id, + _micro_batch_id_ = _micro_batch_id) ) synthetic_data = pd.concat(_split) elif _household_type.startswith("mc"): # couple + some other people synthetic_data = generate_personal_ids(synthetic_data, contains_couples=True) - else: # TODO this only works for m3, m4 not mf + else: # TODO this only works for m3, m4 not mf synthetic_data = generate_personal_ids(synthetic_data, contains_couples=False) return synthetic_data diff --git a/src/synthwave/utils/uk/pre_process.py b/src/synthwave/utils/uk/pre_process.py new file mode 100644 index 0000000..010c682 --- /dev/null +++ b/src/synthwave/utils/uk/pre_process.py @@ -0,0 +1,24 @@ +# HR 13/03/25 Getting pre-processing running from command line + +from synthwave.utils.uk.understanding_society import preprocess_usoc_data +import argparse + +def do_it(_path): + try: + print('Trying to get pre-processed ind and hh pickle files...') + preprocess_usoc_data(_path, skip_conversion=True) + print('Done!') + except: + print("Exception occurred, probably because the pickle files weren't found; running pre-processing step and caching pickles... ") + preprocess_usoc_data(_path, skip_conversion=False) + print('Done!') + + +if __name__ == "__main__": + + parser = argparse.ArgumentParser() + parser.add_argument("p", type=str, help="Data source path") + + args = parser.parse_args() + us_path = args.p + do_it(us_path) diff --git a/src/synthwave/utils/uk/understanding_society.py b/src/synthwave/utils/uk/understanding_society.py index 3b85130..3eb7c79 100644 --- a/src/synthwave/utils/uk/understanding_society.py +++ b/src/synthwave/utils/uk/understanding_society.py @@ -353,8 +353,12 @@ def preprocess_usoc_data(_path="~/Work/data/", skip_conversion=True): individuals, households = merge_usoc_data(_path + "synthwave") - individuals.to_pickle(_path + "synthwave/md/individuals.pkl") - households.to_pickle(_path + "synthwave/md/households.pkl") + _md_path = "synthwave/md/" + if not os.path.exists(_path + _md_path): + print("Preprocessing folder not found; creating at {}".format(_path + _md_path)) + os.makedirs(_path + _md_path) + individuals.to_pickle(_path + _md_path + "individuals.pkl") + households.to_pickle(_path + _md_path + "households.pkl") households = process_households(households) # NOTE this procedure might decrease the number of households