diff --git a/.gitignore b/.gitignore index cf55a3998..42b28adea 100644 --- a/.gitignore +++ b/.gitignore @@ -69,4 +69,10 @@ trace-* test_sherlock/ # SMS API # -.hpc_env \ No newline at end of file +.hpc_env + +# Local analysis notebooks +*.ipynb + +# Generated simulation state/output +data/*.json \ No newline at end of file diff --git a/configs/N_all_media_conditions.json b/configs/N_all_media_conditions.json new file mode 100644 index 000000000..aee307468 --- /dev/null +++ b/configs/N_all_media_conditions.json @@ -0,0 +1,26 @@ +{ + "experiment_id": "all_media_conditions2", + "sim_data_path": null, + "suffix_time": false, + "lineage_seed": 0, + "generations": 8, + "n_init_sims": 1, + "single_daughters": true, + "emitter": "parquet", + "emitter_arg": { + "out_dir": "out" + }, + "variants": { + "condition": { + "condition": { + "value": ["with_aa", "acetate", "succinate", "no_oxygen"] + } + } + }, + "analysis_options": { + "multivariant": { + "doubling_time_hist": {"skip_n_gens": 0}, + "doubling_time_line": {} + } + } +} \ No newline at end of file diff --git a/configs/colony_baseline_test.json b/configs/colony_baseline_test.json new file mode 100644 index 000000000..2c18929e0 --- /dev/null +++ b/configs/colony_baseline_test.json @@ -0,0 +1,29 @@ +{ + "inherit_from": ["spatial.json"], + "description": "Test: 2 generations (4 cells), save parquet per cell, save final state", + + "seed": 0, + "sim_data_path": "/home/katha/dev/vEcoli-colony-sim/out/all_media_conditions2/parca/kb/simData.cPickle", + + "emitter": "parquet", + "emitter_arg": { + "out_dir": "out/colony_runs/baseline_test1" + }, + + "inner_emitter": "parquet", + + "max_duration": 6000, + + "save": true, + "save_times": [6000], + "colony_save_prefix": "baseline_test1", + + "parallel": false, + + "engine_process_reports": [ + ["boundary"], + ["bulk"], + ["listeners"], + ["environment", "exchange"] + ] +} \ No newline at end of file diff --git a/configs/colony_baseline_test2.json b/configs/colony_baseline_test2.json new file mode 100644 index 000000000..30fa8e01a --- /dev/null +++ b/configs/colony_baseline_test2.json @@ -0,0 +1,30 @@ +{ + "inherit_from": ["spatial.json"], + "description": "Test2: read from trial1, run for another generation, save parquet per cell, save final state", + "initial_colony_file": "baseline_2gen_seed_0_colony_t6000", + + "seed": 0, + "sim_data_path": "/home/katha/dev/vEcoli-colony-sim/out/all_media_conditions2/parca/kb/simData.cPickle", + + "emitter": "parquet", + "emit_step": 60, + "emitter_arg": { + "out_dir": "out/colony_runs/baseline_3rd_gen_seed_0" + }, + "emit_config": false, + + "max_duration": 9000, + + "save": true, + "save_times": [9000], + "colony_save_prefix": "baseline_3rd_gen", + + "parallel": false, + + "engine_process_reports": [ + ["boundary"], + ["bulk"], + ["listeners"], + ["environment", "exchange"] + ] +} \ No newline at end of file diff --git a/configs/colony_baseline_test3.json b/configs/colony_baseline_test3.json new file mode 100644 index 000000000..cd92df84b --- /dev/null +++ b/configs/colony_baseline_test3.json @@ -0,0 +1,30 @@ +{ + "inherit_from": ["spatial.json"], + "description": "Test3: 2 generations (4 cells), save parquet per cell, save final state", + + "seed": 10, + "sim_data_path": "/home/katha/dev/vEcoli-colony-sim/out/all_media_conditions2/parca/kb/simData.cPickle", + + "emit_step": 60, + "emitter": "parquet", + "emitter_arg": { + "out_dir": "out/colony_runs/baseline_test3" + }, + + "inner_emitter": "parquet", + + "max_duration": 6000, + + "save": true, + "save_times": [6000], + "colony_save_prefix": "baseline_test3", + + "parallel": false, + + "engine_process_reports": [ + ["boundary"], + ["bulk"], + ["listeners"], + ["environment", "exchange"] + ] +} \ No newline at end of file diff --git a/configs/colony_baseline_test4.json b/configs/colony_baseline_test4.json new file mode 100644 index 000000000..404766528 --- /dev/null +++ b/configs/colony_baseline_test4.json @@ -0,0 +1,30 @@ +{ + "inherit_from": ["spatial.json"], + "description": "Test4: Emit_step test for only 120 seconds, save parquet per cell, save final state", + + "seed": 10, + "sim_data_path": "/home/katha/dev/vEcoli-colony-sim/out/all_media_conditions2/parca/kb/simData.cPickle", + + "emit_step": 60, + "emitter": "parquet", + "emitter_arg": { + "out_dir": "out/colony_runs/baseline_test4_2" + }, + + "inner_emitter": "parquet", + + "max_duration": 120, + + "save": true, + "save_times": [120], + "colony_save_prefix": "baseline_test4", + + "parallel": false, + + "engine_process_reports": [ + ["boundary"], + ["bulk"], + ["listeners"], + ["environment", "exchange"] + ] +} \ No newline at end of file diff --git a/ecoli/experiments/ecoli_engine_process.py b/ecoli/experiments/ecoli_engine_process.py index 10d9d5d00..761c57da2 100644 --- a/ecoli/experiments/ecoli_engine_process.py +++ b/ecoli/experiments/ecoli_engine_process.py @@ -29,6 +29,9 @@ tuplify_topology, ) from ecoli.library.logging_tools import write_json + +# 27 August Fix3: needed so shutdown can finalize parquet emitters. +from ecoli.library.parquet_emitter import ParquetEmitter from ecoli.library.sim_data import RAND_MAX from ecoli.library.schema import not_a_process from ecoli.library.json_state import get_state_from_file @@ -149,6 +152,8 @@ def generate_processes(self, config): def generate_topology(self, config): pass + # Fix 3: pass the emit cadence through to the inner EngineProcess. + class EcoliEngineProcess(Composer): """ @@ -202,6 +207,8 @@ def generate_processes(self, config): "tunnel_out_schemas": config["tunnel_out_schemas"], "stub_schemas": config["stub_schemas"], "seed": (config["seed"] + 1) % RAND_MAX, + # Fix 3: pass the emit cadence through to the inner EngineProcess. + "emit_step": config.get("emit_step", 1), "inner_emitter": config["inner_emitter"], "divide": config["divide"], # Inner sim will set store at ('division_trigger',) @@ -263,30 +270,54 @@ def colony_save_states(engine, config): engine.update(time_to_next_save) time_elapsed = config["save_times"][i] - # Save the full state of the super-simulation + # Save the full state of the super-simulation # Fix 1 state_to_save = engine.state.get_value(condition=not_a_process) - # Get internal state from the EngineProcess sub-simulation - for agent_id in state_to_save["agents"]: - engine.state.get_path( - ("agents", agent_id, "cell_process") - ).value.send_command("get_inner_state") - for agent_id in state_to_save["agents"]: + # The parquet emitter wraps the entire outer simulation in + # agents/outer. Colony save files should NOT contain this wrapper, + # because initial_colony_file expects agents/. + if config["emitter"] == "parquet": + colony_state = state_to_save["agents"]["outer"] + cell_process_base_path = ("agents", "outer", "agents") + else: + colony_state = state_to_save + cell_process_base_path = ("agents",) + + # Get internal state from each EngineProcess sub-simulation + for agent_id in colony_state["agents"]: + cell_process_path = cell_process_base_path + ( + agent_id, + "cell_process", + ) + engine.state.get_path(cell_process_path).value.send_command( + "get_inner_state" + ) + + for agent_id in colony_state["agents"]: + cell_process_path = cell_process_base_path + ( + agent_id, + "cell_process", + ) cell_state = engine.state.get_path( - ("agents", agent_id, "cell_process") + cell_process_path ).value.get_command_result() + # Can't save, but will be restored when loading state del cell_state["environment"]["exchange_data"] + # Shared processes are re-initialized on load del cell_state["process"] + # Save bulk and unique dtypes cell_state["bulk_dtypes"] = str(cell_state["bulk"].dtype) cell_state["unique_dtypes"] = {} for name, mols in cell_state["unique"].items(): cell_state["unique_dtypes"][name] = str(mols.dtype) - state_to_save["agents"][agent_id] = cell_state - state_to_save = serialize_value(state_to_save) + colony_state["agents"][agent_id] = cell_state + + state_to_save = serialize_value(colony_state) # Fix 1 + if config.get("colony_save_prefix", None): write_json( "data/" @@ -317,6 +348,19 @@ def colony_save_states(engine, config): engine.update(time_remaining) +def finalize_parquet_emitters(processes, seen=None): # Fix 3 + if seen is None: + seen = set() + if isinstance(processes, dict): + for process in processes.values(): + finalize_parquet_emitters(process, seen) + elif hasattr(processes, "emitter"): + emitter = processes.emitter + if isinstance(emitter, ParquetEmitter) and id(emitter) not in seen: + seen.add(id(emitter)) + emitter.finalize() + + def run_simulation(config): """ Main method for running colony simulations in @@ -368,6 +412,8 @@ def run_simulation(config): "stub_schemas": stub_schemas, "parallel": config["parallel"], "divide": config["divide"], + # Fix 3: pass the emit cadence into the inner EngineProcess setup. + "emit_step": config.get("emit_step", 1), "tunnels_in": ( ("environment",), ("boundary",), @@ -512,6 +558,11 @@ def run_simulation(config): else: engine.update(config["max_duration"]) engine.end() + # Fix 2&3: flush buffered parquet output on normal shutdown. + if config["emitter"] == "parquet": + engine.emitter.success = True + engine.emitter.finalize() + finalize_parquet_emitters(engine.processes) if config["profile"]: report_profiling(engine.stats) diff --git a/ecoli/processes/engine_process.py b/ecoli/processes/engine_process.py index 5b1df89d8..d97fede67 100644 --- a/ecoli/processes/engine_process.py +++ b/ecoli/processes/engine_process.py @@ -210,6 +210,7 @@ class EngineProcess(Process): "inner_composer_config": {}, "outer_composer_config": {}, "seed": 0, + "emit_step": 1, # Fix3 "inner_emitter": "null", "divide": False, "division_threshold": None, @@ -514,16 +515,17 @@ def next_update(self, timestep, states): # other words, since we rely on the outer Engine to apply the # updates, we have to wait for those updates from the previous # timestep to be applied before we emit data. - data = self.sim.state.emit_data() - if isinstance(self.emitter, ParquetEmitter): - # ParquetEmitter expects per-agent data under agents/. - data = {"agents": {self.parameters["agent_id"]: data}} - data["time"] = self.sim.global_time - emit_config = { - "table": "history", - "data": data, - } - self.emitter.emit(emit_config) + if self.sim.global_time % self.parameters["emit_step"] == 0: # Fix 3 + data = self.sim.state.emit_data() + if isinstance(self.emitter, ParquetEmitter): + # ParquetEmitter expects per-agent data under agents/. + data = {"agents": {self.parameters["agent_id"]: data}} + data["time"] = self.sim.global_time + emit_config = { + "table": "history", + "data": data, + } + self.emitter.emit(emit_config) # Run inner simulation for timestep. try: diff --git a/reconstruction/ecoli/dataclasses/process/two_component_system.py b/reconstruction/ecoli/dataclasses/process/two_component_system.py index 70afba417..99721bddb 100644 --- a/reconstruction/ecoli/dataclasses/process/two_component_system.py +++ b/reconstruction/ecoli/dataclasses/process/two_component_system.py @@ -248,6 +248,20 @@ def __setstate__(self, state): self._populate_derivative_and_jacobian() self.dependency_matrix = self._make_dependency_matrix() + @property + def modified_molecules( + self, + ): # Fix 0 + value = self.__dict__.get("modified_molecules") + if value is None: + value = self.make_modified_molecule_list() + self.__dict__["modified_molecules"] = value + return value + + @modified_molecules.setter # Fix 0 + def modified_molecules(self, value): + self.__dict__["modified_molecules"] = value + def _buildComplexToMonomer(self, sim_data, modifiedFormsMonomers, tcsMolecules): """ Maps each complex to a dictionary that maps each subunit of the complex diff --git a/submit.sh b/submit.sh new file mode 100644 index 000000000..dab24d3b1 --- /dev/null +++ b/submit.sh @@ -0,0 +1,114 @@ +#!/bin/bash +#SBATCH --job-name=p5_462KO_basal_operon_on +#SBATCH --partition=compute +#SBATCH --time=14-00:00:00 +#SBATCH --chdir=/user/home/il22158 +#SBATCH --account=emat024603 +#SBATCH --output=/user/home/il22158/work/slurm_logs/p5_462KO_basal_operon_on.%j.out +#SBATCH --mem=200G +#SBATCH --cpus-per-task=24 + +# == Work directory setup == +WORK_DIR="/user/home/il22158/work/vEcoli" +cd "$WORK_DIR" || exit + +# == Python environment setup == +source "$WORK_DIR/.venv/bin/activate" + +# == Nextflow version control == +nextflow -version +#Stable nextflow version: 25.10.2 + +# == Module setup == +module load languages/java-sdk/22.0.2 openssh/9.7p1-uyheegq git +module list # Print loaded modules + +# === VERSION CONTROL === +# git add. #stages changes under the current directory +# SNAPSHOT_BRANCH="snapshots/job-${SLURM_JOB_ID}-$(date +%Y%m%d_%H%M%S)" +# git checkout -b "$SNAPSHOT_BRANCH" +# git commit -m "Snapshot for job ${SLURM_JOB_ID}" || true +# git checkout - # Go back to previous branch +# echo "Snapshot for job ${SLURM_JOB_ID}" + +# === JOB EXECUTION === + +# echo "Starting downsampling of history parquet files again to make it end in 1/20 size..." +# python reading/downsample_history.py --dir /user/home/il22158/work/vEcoli/out/gene_ko_non_metabolic_seed100/history --n 20 +# python reading/downsample_history.py --dir /user/home/il22158/work/vEcoli/out/gene_ko_metabolic_seed100/history --n 20 +# # change total number of samples to 1/n + +echo "Re-run part 5 of 462 gene KO in basal operon on conditions..." +python runscripts/workflow.py --config configs/N_gene_knockout_462KO_basal_operon_on_p5.json +# # python runscripts/workflow.py --config configs/N_gene_ko_40trial_seed100_aa.json +# python runscripts/workflow.py --config configs/N_gene_ko_40trial_seed100_acetate.json +# python runscripts/workflow.py --config configs/N_gene_ko_40trial_seed100_succinate.json +# python runscripts/workflow.py --config configs/N_gene_ko_40trial_seed100_no_ox.json + +# Resume workflow for previous job ended because if time limit. +# python runscripts/workflow.py --config configs/N_gene_knockout_leftover.json + +# echo "Run gene screen for the 3rd KO test..." +# python reading/gene_screen.py --project gene_knockout_3_round_test --lineage-seed 100 101 --variants $(seq 0 50) --gene-list surrogate/third_round_tested_gene_list.txt + +# echo "Extracting growth rates for extended succinate simulation..." + +# python /user/home/il22158/work/vEcoli/reading/growth_rate_extract.py \ +# --all --projects gene_ko_40trial_seed100_succinate --save-timeseries\ +# --suffix gene_ko_40trial_succinate_extended --lineage-seeds 100 101 + +# echo "Preprocessing growth rate to fold change data (all parquet files)..." +# GENE_LIST=/user/home/il22158/work/vEcoli/reading/imported/Single_KO_RNA_names.txt +# KO100=/user/home/il22158/work/vEcoli/reading/results/growth_rate/growth_rate_timeseries_441_KOs_seed100_all.parquet +# KO101=/user/home/il22158/work/vEcoli/reading/results/growth_rate/growth_rate_timeseries_441_KOs_seed101_all.parquet +# S100=/user/home/il22158/work/vEcoli/reading/results/growth_rate/growth_rate_timeseries_seed100_all.parquet +# S101=/user/home/il22158/work/vEcoli/reading/results/growth_rate/growth_rate_timeseries_seed101_all.parquet +# DEF=/user/home/il22158/work/vEcoli/reading/results/growth_rate/growth_rate_timeseries_default_all.parquet + +# # 1. 441 KO files use *20 step conversion +# python surrogate/preprocess.py \ +# --mode batch \ +# --gene-list "${GENE_LIST}" \ +# --timeseries-files "${KO100}" "${KO101}" \ +# --step-scale 20 \ +# --output-prefix surrogate_preprocessed_ko441 + +# 2. seed/default files include baseline and use step conversion 1 +# python surrogate/preprocess.py \ +# --mode batch \ +# --gene-list "${GENE_LIST}" \ +# --timeseries-files "${S100}" "${S101}" "${DEF}" \ +# --step-scale 1 \ +# --output-prefix surrogate_preprocessed_seed_default + +# echo "Gene screen for outliers..." +# set -e + +# GENE_SCREEN="/user/home/il22158/work/vEcoli/reading/gene_screen.py" +# LIST_DIR="/user/home/il22158/work/vEcoli/surrogate/results/failure/outlier_gene_lists_by_project" +# COMMON_ARGS="--generations 1 2 3 4 5 6 7 8 --subset 0" + +# # Format: projectlineage_seedvariants +# while IFS=$'\t' read -r project seed variants; do +# [[ -z "$project" ]] && continue +# gene_list="$LIST_DIR/outlier_genes_unique__${project}.txt" +# output_project="outlier_translation_${project}_seed${seed}" + +# python3 "$GENE_SCREEN" \ +# --project "$project" \ +# --variants $variants \ +# --lineage-seed "$seed" \ +# $COMMON_ARGS \ +# --gene-list "$gene_list" \ +# --output-project "$output_project" +# done <<'EOF' +# gene_ko_441imported_2seeds 100 7 11 24 31 41 48 57 59 65 84 95 98 101 117 122 125 133 135 145 157 189 191 195 200 212 213 218 226 230 239 240 241 244 246 267 284 286 288 289 297 317 326 329 356 357 368 370 400 419 433 436 +# gene_ko_441imported_2seeds 101 7 11 24 31 57 59 65 95 98 101 117 124 125 133 135 145 157 191 195 200 213 218 226 230 241 244 246 267 284 286 288 297 329 357 368 370 400 433 436 +# gene_ko_metabolic_seed100 100 7 +# gene_ko_non_metabolic_seed100 100 7 +# gene_ko_trial40_seed101 101 7 +# EOF + +# Colony simulation + +python ecoli/experiments/ecoli_engine_process.py --config configs/colony_baseline_test.json \ No newline at end of file