hydra restructure

.gitignore

@@ -1,11 +1,18 @@
 dpacman/data_files
-dpacman/data/tfclust/*.log
-dpacman/data/tfclust/temp.py
+dpacman/preprocess/tfclust/*.log
+dpacman/preprocess/tfclust/temp.py
 bigBedToBed
-dpacman/data/remap/*.log
-dpacman/data/remap/temp.py
-dpacman/data/tfclust/figures
+dpacman/preprocess/remap/*.log
+dpacman/preprocess/remap/temp.py
+dpacman/preprocess/tfclust/figures
 dpacman/softwares/meme*
-dpacman/data/remap/crm_example.csv
-dpacman/data/remap/crm_example_ERG.csv
+dpacman/preprocess/remap/crm_example.csv
+dpacman/preprocess/remap/crm_example_ERG.csv
+dpacman/classifier/old
+dpacman/classifier/__pycache__/
+dpacman/data_tasks/clean/__pycache__/
+dpacman/data_tasks/download/__pycache__/
+dpacman/data_tasks/fimo/__pycache__/
+dpacman/scripts/__pycache__/
+logs/
 tree.txt

configs/data_task/clean/remap.yaml

@@ -0,0 +1,12 @@
+name: remap
+type: clean
+
+nr_raw_path: dpacman/data_files/raw/remap/remap2022_nr_macs2_hg38_v1_0.bed
+nr_processed_dir: dpacman/data_files/processed/remap
+nr_processed_filename: remap2022_nr_macs2_hg38_v1_0_clean.tsv
+
+crm_raw_path: dpacman/data_files/raw/remap/remap2022_crm_macs2_hg38_v1_0.bed
+crm_processed_dir: dpacman/data_files/processed/remap
+crm_processed_filename: remap2022_crm_macs2_hg38_v1_0_clean.tsv
+
+save_example_files: true

configs/data_task/download/genome.yaml

@@ -0,0 +1,5 @@
+name: genome
+type: download
+output_dir: dpacman/classifier/data_files/raw/genomes
+genomes: 
+  - hg38

configs/data_task/download/remap.yaml

@@ -0,0 +1,12 @@
+name: remap
+type: download
+
+nr_url: https://remap.univ-amu.fr/storage/remap2022/hg38/MACS2/remap2022_nr_macs2_hg38_v1_0.bed.gz
+nr_output_dir: dpacman/data_files/raw/remap
+nr_filename: remap2022_nr_macs2_hg38_v1_0.bed.gz
+
+crm_url: https://remap.univ-amu.fr/storage/remap2022/hg38/MACS2/remap2022_crm_macs2_hg38_v1_0.bed.gz
+crm_output_dir: dpacman/data_files/raw/remap
+crm_filename: remap2022_crm_macs2_hg38_v1_0.bed.gz
+
+delete_zip: true

configs/data_task/fimo/post_fimo.yaml

@@ -0,0 +1,6 @@
+name: post_fimo
+type: fimo
+
+input_csv: dpacman/data_files/processed/fimo/remap2022_crm_fimo_output.csv
+output_csv: dpacman/data_files/processed/fimo/remap2022_crm_fimo_output_processed.csv
+json_dir: dpacman/data_files/raw/genomes/hg38

configs/data_task/fimo/pre_fimo.yaml

@@ -0,0 +1,6 @@
+name: pre_fimo
+type: fimo
+
+input_csv: dpacman/data_files/processed/remap/remap2022_crm_macs2_hg38_v1_0_clean.tsv
+output_csv: dpacman/data_files/processed/fimo/remap2022_crm_fimo_input.tsv
+window_total: 500

configs/data_task/fimo/run_fimo.yaml

@@ -0,0 +1,22 @@
+name: post_fimo
+type: fimo
+
+paths:
+  input_csv: dpacman/data_files/processed/fimo/remap2022_crm_fimo_input.tsv
+  output_csv: dpacman/data_files/processed/fimo/remap2022_crm_fimo_output.csv
+  json_dir: dpacman/data_files/raw/genomes/hg38
+
+meme:
+  fimo_bin: dpacman/softwares/meme/bin/fimo
+  fasta_get_markov: dpacman/softwares/meme/libexec/meme-5.5.8/fasta-get-markov
+  jaspar_motif_file: dpacman/softwares/meme-5.5.8/tests/common/JASPAR_CORE_2014_vertebrates.meme
+
+fnames:
+  seq_fasta: to_scan.fa
+  bg_model: bg_model.txt
+  fimo_outdir: fimo_out
+
+fimo:
+  pval_thresh: 1e-4
+  max_stored: 1000000
+  njobs: max

configs/hydra/default.yaml

@@ -0,0 +1,17 @@
+# configs/hydra/default.yaml
+
+defaults:
+  - override hydra_logging: colorlog
+  - override job_logging: colorlog
+
+run:
+  dir: ${paths.log_dir}/${task_name}/runs/${now:%Y-%m-%d}_${now:%H-%M-%S}
+
+sweep:
+  dir: ${paths.log_dir}/${task_name}/multiruns/${now:%Y-%m-%d}_${now:%H-%M-%S}
+  subdir: ${hydra.job.num}
+
+job_logging:
+  handlers:
+    file:
+      filename: ${hydra.runtime.output_dir}/task.log

configs/paths/default.yaml

@@ -0,0 +1,16 @@
+# configs/paths/default.yaml
+
+# project root
+root_dir: ${oc.env:PROJECT_ROOT}
+
+# path to raw and processed data
+data_dir: ${paths.root_dir}/interactome/data_files/
+
+# path to logs
+log_dir: ${paths.root_dir}/logs/
+
+# hydra-managed output dir
+output_dir: ${hydra:runtime.output_dir}
+
+# working dir (original CWD when launched)
+work_dir: ${hydra:runtime.cwd}

configs/preprocess.yaml

@@ -0,0 +1,9 @@
+# configs/download.yaml
+
+defaults:
+  - _self_
+  - paths: default
+  - hydra: default  # ← tells Hydra to use the logging/output config
+  - data_task: download/genome
+
+task_name: preprocess/${data_task.type}

dpacman/data/README.md

@@ -1,41 +0,0 @@
-# Data download directory
-
-## UCSC 
-
-### Raw data download 
-1. `encRegTfbsClusteredWithCells.hg38.bed.gz`
-
-```
-wget https://hgdownload.soe.ucsc.edu/goldenPath/hg38/encRegTfbsClustered/encRegTfbsClusteredWithCells.hg38.bed.gz
-gunzip encRegTfbsClusteredWithCells.hg38.bed.gz
-```
-
-2. `encRegTfbsClusteredWithCells.hg19.bed.gz`
-
-```
-wget https://hgdownload.soe.ucsc.edu/goldenPath/hg19/encRegTfbsClustered/encRegTfbsClusteredWithCells.hg19.bed.gz
-gunzip encRegTfbsClusteredWithCells.hg19.bed.gz
-```
-
-3. ReMap big bed file 
-```
-wget https://hgdownload.soe.ucsc.edu/gbdb/hg38/reMap/reMap2022.bb
-wget http://hgdownload.soe.ucsc.edu/admin/exe/linux.x86_64/bigBedToBed
-chmod +x bigBedToBed
-./bigBedToBed /home/a03-svincoff/DPACMAN/dpacman/data_files/raw/remap/reMap2022.bb /home/a03-svincoff/DPACMAN/dpacman/data_files/raw/remap/reMap2022.bed
-
-```
-
-4. ReMap CRM file from their actual website 
-```
-wget https://remap.univ-amu.fr/storage/remap2022/hg38/MACS2/remap2022_crm_macs2_hg38_v1_0.bed.gz
-gunzip remap2022_crm_macs2_hg38_v1_0.bed.gz
-```
-
-4. Run `download.py` to download:
-- Full sequences of each chromosome for genomes hg38 and hg19
-- encRegTfbsClusteredWithCells, a table of clustered transcription factors by their binding sites, for hg38 and hg19
-- processed databases per genome per chromosome with the following columns: "bin","chrom","chromStart","chromEnd","name","score","scoreCount","sourceIds","sourceScores","seq","seq_flanked","chromStart_flanked","chromEnd_flanked","flank5","flank3"
-
-### Data Processing
-1. Run `combine.py` to combine these individual files into one large DataFrame

dpacman/data/chip_atlas/full_data_loading.py

@@ -1,97 +0,0 @@
-import pandas as pd
-from pathlib import Path
-import subprocess
-
-# Read only cols 0–2, no header
-df = pd.read_csv(
-    "experimentList.tab",
-    sep="\t",
-    header=None,
-    usecols=[0, 1, 2],
-    names=["exp_id", "genome", "assay_group"],
-    engine="python",
-    on_bad_lines="skip",
-    dtype=str,
-)
-
-# Keep only known genome assemblies
-VALID_GENOMES = {
-    "hg19",
-    "hg38",
-    "mm9",
-    "mm10",
-    "rn6",
-    "dm3",
-    "dm6",
-    "ce10",
-    "ce11",
-    "sacCer3",
-}
-df = df[df["genome"].isin(VALID_GENOMES)]
-print("Assemblies in filtered data:", df["genome"].unique())
-
-
-# Classify assay type
-def modality(track):
-    t = track.lower()
-    if "atac" in t:
-        return "ATAC"
-    if "dnase" in t:
-        return "DNase"
-    if "bisulfite" in t or "methyl" in t:
-        return "BS"
-    return "ChIP"
-
-
-df["modality"] = df["assay_group"].apply(modality)
-
-
-# URL templates
-def make_urls(exp, genome, mod):
-    urls = []
-    if mod in ("ChIP", "ATAC", "DNase"):
-        urls.append(f"https://chip-atlas.dbcls.jp/data/{genome}/eachData/bw/{exp}.bw")
-        for thr in ("05", "10", "20"):
-            urls.append(
-                f"https://chip-atlas.dbcls.jp/data/{genome}/eachData/bed{thr}/{exp}.{thr}.bed"
-            )
-            urls.append(
-                f"https://chip-atlas.dbcls.jp/data/{genome}/eachData/bb{thr}/{exp}.{thr}.bb"
-            )
-    else:
-        urls.append(
-            f"https://chip-atlas.dbcls.jp/data/{genome}/eachData/bs/methyl/{exp}.methyl.bw"
-        )
-        urls.append(
-            f"https://chip-atlas.dbcls.jp/data/{genome}/eachData/bs/cover/{exp}.cover.bw"
-        )
-        for sub in ("hmr", "pmd", "hypermr"):
-            urls.append(
-                f"https://chip-atlas.dbcls.jp/data/{genome}/eachData/bs/{sub}/Bed/{exp}.{sub}.bed"
-            )
-            urls.append(
-                f"https://chip-atlas.dbcls.jp/data/{genome}/eachData/bs/{sub}/BigBed/{exp}.{sub}.bb"
-            )
-    return urls
-
-
-# Write URL lists per genome
-urls_dir = Path("urls_by_genome")
-urls_dir.mkdir(exist_ok=True)
-for genome, group in df.groupby("genome"):
-    all_urls = []
-    for _, row in group.iterrows():
-        all_urls += make_urls(row.exp_id, genome, row.modality)
-    uniq = sorted(set(all_urls))
-    (urls_dir / f"urls_{genome}.txt").write_text("\n".join(uniq))
-    print(f"{genome}: {len(uniq)} URLs")
-
-# Download into raw/{genome}/
-for url_file in urls_dir.glob("urls_*.txt"):
-    genome = url_file.stem.split("_", 1)[1]
-    dest = Path("raw") / genome
-    dest.mkdir(parents=True, exist_ok=True)
-    print(f"\nDownloading {genome} → {dest}/…")
-    subprocess.run(["wget", "-nc", "-i", str(url_file), "-P", str(dest)], check=True)
-
-print("Done! Check raw/{genome}/ for your files.")

dpacman/data/chip_atlas/smaller_data_loading.py

@@ -1,160 +0,0 @@
-#!/usr/bin/env python3
-import os, sys, zipfile
-import subprocess
-import random
-from pathlib import Path
-import requests
-import pandas as pd
-from tqdm import tqdm
-
-# ─── PARAMETERS ───────────────────────────────────────────────────────────────
-# total target regions (rough guide; you'll filter post‐download if needed)
-TARGET_REGIONS = 200_000
-
-# Assemblies to include
-ASSEMBLIES = [
-    "hg19",
-    "hg38",
-    "mm9",
-    "mm10",
-    "rn6",
-    "dm3",
-    "dm6",
-    "ce10",
-    "ce11",
-    "sacCer3",
-]
-
-# How many experiments to sample at most per protein (tune up/down)
-MAX_EXPS_PER_PROTEIN = 50
-
-# Number of parallel connections for aria2c
-ARIA2C_CONN = 16
-
-# Working directories
-WORKDIR = Path("chip_atlas_fetch")
-WORKDIR.mkdir(exist_ok=True)
-LIST_DIR = WORKDIR / "lists"
-LIST_DIR.mkdir(exist_ok=True)
-DL_DIR = WORKDIR / "downloads"
-DL_DIR.mkdir(exist_ok=True)
-
-# ─── HELPERS ──────────────────────────────────────────────────────────────────
-
-
-def download_and_extract(url, extract_to: Path):
-    """Fetch a ZIP and unzip it."""
-    local = extract_to / Path(url).name
-    if not local.exists():
-        print(f"→ Downloading {url}")
-        resp = requests.get(url, stream=True)
-        resp.raise_for_status()
-        with open(local, "wb") as f:
-            for chunk in resp.iter_content(1 << 20):
-                f.write(chunk)
-    with zipfile.ZipFile(local, "r") as z:
-        z.extractall(extract_to)
-
-
-# ─── 1) GET MASTER LISTS ────────────────────────────────────────────────────
-
-print("1) Fetching master file & experiment lists…")
-FILELIST_URL = (
-    "https://dbarchive.biosciencedbc.jp/data/chip-atlas/LATEST/chip_atlas_file_list.zip"
-)
-EXPERIMENTLIST_URL = "https://dbarchive.biosciencedbc.jp/data/chip-atlas/LATEST/chip_atlas_experiment_list.zip"
-
-download_and_extract(FILELIST_URL, LIST_DIR)
-download_and_extract(EXPERIMENTLIST_URL, LIST_DIR)
-
-filelist_txt = LIST_DIR / "chip_atlas_file_list.csv"
-experiment_txt = LIST_DIR / "chip_atlas_experiment_list.csv"
-
-# ─── 2) PARSE EXPERIMENT METADATA ────────────────────────────────────────────
-
-print("2) Parsing experiment → protein lookup…")
-exp_df = pd.read_csv(
-    experiment_txt,
-    sep=None,  # let python engine guess (comma vs. tab)
-    engine="python",  # required when sep=None
-    encoding="latin1",  # to avoid UnicodeDecodeErrors
-)
-
-print("Columns in experiment list:", exp_df.columns.tolist())
-
-exp_df = exp_df.loc[:, ["Experimental ID", "Genome assembly", "Antigen"]].rename(
-    columns={
-        "Experimental ID": "exp_id",
-        "Genome assembly": "genome",
-        "Antigen": "assay_group",
-    }
-)
-
-exp_df["protein"] = exp_df["assay_group"].str.replace(r"_ChIP.*", "", regex=True)
-
-# Finally, filter to only the assemblies you care about:
-exp_df = exp_df[exp_df["genome"].isin(ASSEMBLIES)]
-
-# build lookup
-exp_to_genome = exp_df.set_index("exp_id")["genome"].to_dict()
-exp_to_protein = exp_df.set_index("exp_id")["protein"].to_dict()
-
-# ─── 3) BUILD URL LIST DIRECTLY ───────────────────────────────────────────────
-
-print("3) Building URL list for .bw + .10.bed…")
-BASE = "https://dbarchive.biosciencedbc.jp/data/chip-atlas"
-urls_by_exp = {}
-for exp, genome in exp_to_genome.items():
-    urls_by_exp[exp] = [
-        f"{BASE}/data/{genome}/eachData/bw/{exp}.bw",
-        f"{BASE}/data/{genome}/eachData/bed10/{exp}.10.bed",
-    ]
-
-# bucket experiments by protein
-from collections import defaultdict
-
-prot_exps = defaultdict(list)
-for exp, prot in exp_to_protein.items():
-    if exp in urls_by_exp:
-        prot_exps[prot].append(exp)
-
-# sample up to MAX_EXPS_PER_PROTEIN per protein
-sampled_exps = []
-for prot, exps in prot_exps.items():
-    k = min(len(exps), MAX_EXPS_PER_PROTEIN)
-    sampled_exps += random.sample(exps, k)
-
-print(f" → Sampling {len(sampled_exps):,} experiments across {len(prot_exps)} proteins")
-
-# collect URLs for just those experiments
-final_urls = []
-for exp in sampled_exps:
-    final_urls += urls_by_exp[exp]
-random.shuffle(final_urls)
-
-# write out for aria2c
-url_list_file = WORKDIR / "to_download.txt"
-with open(url_list_file, "w") as f:
-    for u in final_urls:
-        f.write(u + "\n")
-print(f" → Wrote {len(final_urls):,} URLs to {url_list_file}")
-
-# ─── 4) PARALLEL DOWNLOAD VIA aria2c ─────────────────────────────────────────
-
-print("4) Downloading with aria2c…")
-subprocess.run(
-    [
-        "aria2c",
-        f"-x{ARIA2C_CONN}",
-        "--dir",
-        str(DL_DIR),
-        "--input-file",
-        str(url_list_file),
-        "--auto-file-renaming=false",
-        "--allow-overwrite=true",
-    ],
-    check=True,
-)
-
-print("✅ Finished downloading all selected files.")
-print(f"Your files are in: {DL_DIR.resolve()}")

dpacman/data/consistency.py

@@ -1,10 +0,0 @@
-"""
-Check for consistency between Remap and Tfclust data
-"""
-
-import logging
-import pandas as pd
-import logging
-import os
-import dask.dataframe as dd
-

dpacman/data/remap/analyze.py

@@ -1,48 +0,0 @@
-import logging
-import pandas as pd
-
-def main(logger=None):
-    if logger is None:
-        logger = logging.getLogger(__name__)
-
-    # Read the BED file 
-    bed_file_path = "../../data_files/raw/remap/reMap2022.bed"
-    df = pd.read_csv(bed_file_path, sep="\t", header=None)
-    df.columns = ["#chrom", "chromStart", "chromEnd", "name", "score", "strand", "thickStart", "thickEnd", "reserved", "TF", "Biotypes"]
-    print(f"{len(df):,}")
-    crm["chromLen"] = crm["chromEnd"] - crm["chromStart"]
-    print(crm["chromLen"].describe())
-    print(df.head(50))
-    
-    crm_bed_file_path = "../../data_files/raw/remap/remap2022_crm_macs2_hg38_v1_0.bed"
-    crm = pd.read_csv(crm_bed_file_path, sep="\t", header=None)
-    crm.columns = ["#chrom", "chromStart", "chromEnd", "name", "score", "strand", "thickStart", "thickEnd", "reserved"]
-    crm["chromLen"] = crm["chromEnd"] - crm["chromStart"]
-    crm["thickLen"] = crm["thickEnd"] - crm["thickStart"]
-    print(f"{len(crm):,}")
-    print(f"thick length statistics:")
-    print(crm["thickLen"].describe())
-    print(f"chrom length statistics:")
-    print(crm["chromLen"].describe())
-    print(crm[["#chrom", "chromStart", "chromEnd", "name", "score", "strand", "thickStart", "thickEnd", "reserved"]].head(50))
-    crm.head(50).to_csv("crm_example.csv",index=False)
-    
-    crm["name"] = crm["name"].apply(lambda x: x.split(","))
-    crm = crm.explode("name").reset_index(drop=True)
-    crm.loc[crm["name"]=="ERG"].reset_index(drop=True).head(50).to_csv("crm_example_ERG.csv",index=False)
-
-if __name__ == "__main__":
-    log_path = "analyze.log" 
-
-    logger = logging.getLogger(__name__)
-    logger.setLevel(logging.DEBUG)
-
-    # Create file handler
-    file_handler = logging.FileHandler(log_path, mode="w", encoding="utf-8")
-    formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
-    file_handler.setFormatter(formatter)
-
-    # Attach handlers
-    logger.addHandler(file_handler)
-
-    main(logger)

dpacman/data/remap/pre_fimo.py

@@ -1,61 +0,0 @@
-#!/usr/bin/env python3
-import pandas as pd
-import numpy as np
-
-# ------------------------------------------------------------------
-# PARAMETERS
-INPUT_CSV  = "/home/a03-akrishna/DPACMAN/dpacman/data/remap/full_crm.csv"
-OUTPUT_CSV = "/home/a03-akrishna/DPACMAN/data_files/processed/clean_pre_fimo.csv"
-WINDOW_TOTAL = 500   # total extra context bp around each peak
-# ------------------------------------------------------------------
-
-def main():
-    # 1) load
-    df = pd.read_csv(INPUT_CSV)
-
-    # 2) normalize chromosomes and exclude non-whole chromosomes
-    df = df.rename(columns={"#chrom": "chrom"})
-    df["chrom"] = df["chrom"].str.replace(r"^chr", "", regex=True)
-
-    valid = [str(i) for i in range(1,23)] + ["X", "Y"]
-    df = df[df["chrom"].isin(valid)].reset_index(drop=True)
-
-    # 3) explode TF names
-    df["TF_list"] = df["name"].str.split(",")
-    df = df.explode("TF_list").rename(columns={"TF_list": "TF"})
-    df["TF"] = df["TF"].str.strip()
-
-    # 4) draw a random left‐flank between 0 and WINDOW_TOTAL,
-    #    then right‐flank is whatever remains to sum to WINDOW_TOTAL
-    n = len(df)
-    df["left_context"]  = np.random.randint(0, WINDOW_TOTAL + 1, size=n)
-    df["right_context"] = WINDOW_TOTAL - df["left_context"]
-
-    # 5) compute contextStart / contextEnd
-    df["contextStart"] = (df["chromStart"] - df["left_context"]).clip(lower=0).astype(int)
-    df["contextEnd"]   = (df["chromEnd"]   + df["right_context"]).astype(int)
-
-    # 6) assemble output
-    out = df[[
-        "chrom",
-        "contextStart",
-        "chromStart",  # original ChIPStart
-        "chromEnd",    # original ChIPEnd
-        "contextEnd",
-        "score",       # original score column
-        "TF"
-    ]].rename(columns={
-        "chrom":      "#chrom",
-        "chromStart": "ChIPStart",
-        "chromEnd":   "ChIPEnd",
-        "score":      "chipscore"
-    })
-
-    # 7) write CSV
-    out.to_csv(OUTPUT_CSV, index=False)
-    print(f"Wrote {len(out)} rows to {OUTPUT_CSV}")
-
-if __name__ == "__main__":
-    main()
-
-

dpacman/data/tfclust/analyze.py

@@ -1,77 +0,0 @@
-import pandas as pd
-import logging
-import os
-import dask.dataframe as dd
-import matplotlib.pyplot as plt
-
-def plot_sequence_lengths_box(lengths, xlog=False, title="Sequence Lengths", out_dir="figures", fname="sequence_lengths_box.png"):
-    """
-    Plot sequence lengths. Can be used with original sequence or flank sequence. 
-    """
-    os.makedirs(out_dir, exist_ok=True)
-    out_path = os.path.join(out_dir, fname)
-    
-    plt.figure(figsize=(10, 4))
-    plt.boxplot(lengths, vert=False)
-    if xlog:
-        plt.xscale('log')
-    plt.xlabel("Sequence Length")
-    plt.title(title)
-    plt.grid(True, axis='y', linestyle='--', alpha=0.6)
-    plt.tight_layout()
-    
-    plt.savefig(out_path, dpi=300)
-
-def plot_sequence_lengths_hist(lengths, xlog=False, title="Sequence Lengths", out_dir="figures", fname="sequence_lengths_hist.png"):
-    """
-    Plot sequence lengths. Can be used with original sequence or flank sequence. 
-    """
-    os.makedirs(out_dir, exist_ok=True)
-    out_path = os.path.join(out_dir, fname)
-    
-    plt.figure(figsize=(10, 4))
-    plt.hist(lengths, bins=100, density=True, alpha=0.75)
-    if xlog:
-        plt.xscale('log')
-    # percentage format
-    plt.gca().yaxis.set_major_formatter(plt.FuncFormatter(lambda y, _: '{:.2%}'.format(100*y)))
-    plt.xlabel("Sequence Length")
-    plt.ylabel("Frequency")
-    plt.title(title)
-    plt.grid(True, axis='y', linestyle='--', alpha=0.6)
-    plt.tight_layout()
-    
-    plt.savefig(out_path, dpi=300)
-
-def main(logger):
-    df_dir = "../../data_files/processed/tfclust/combined"
-    df_savepath = os.path.join(df_dir, "encRegTfbsClustered_hg38_hg19.parquet")
-    logger.info("Starting to load data file from parquet")
-    df = pd.read_parquet(df_savepath, engine="auto")
-    logger.info(df.head())
-    
-    plot_sequence_lengths_hist(df["seq_len"], title="TF Binding Sites",fname="seq_lengths_hist.png")
-    plot_sequence_lengths_hist(df["seq_flanked_len"], title="TF Binding Sites with 1000nt Flanks", fname="seq_lengths_flanked_hist.png")
-    plot_sequence_lengths_box(df["seq_len"], title="TF Binding Sites",fname="seq_lengths_box.png")
-    plot_sequence_lengths_box(df["seq_flanked_len"], title="TF Binding Sites with 1000nt Flanks", fname="seq_lengths_flanked_box.png")
-    
-    plot_sequence_lengths_hist(df["seq_len"], xlog=True, title="TF Binding Sites",fname="seq_lengths_xlog_hist.png")
-    plot_sequence_lengths_hist(df["seq_flanked_len"], xlog=True, title="TF Binding Sites with 1000nt Flanks", fname="seq_lengths_flanked_xlog_hist.png")
-    plot_sequence_lengths_box(df["seq_len"], xlog=True, title="TF Binding Sites",fname="seq_lengths_xlog_box.png")
-    plot_sequence_lengths_box(df["seq_flanked_len"],xlog=True,  title="TF Binding Sites with 1000nt Flanks", fname="seq_lengths_flanked_xlog_box.png")
-    
-if __name__ == "__main__":
-    log_path = "analyze.log" 
-
-    logger = logging.getLogger(__name__)
-    logger.setLevel(logging.DEBUG)
-
-    # Create file handler
-    file_handler = logging.FileHandler(log_path, mode="w", encoding="utf-8")
-    formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
-    file_handler.setFormatter(formatter)
-
-    # Attach handlers
-    logger.addHandler(file_handler)
-
-    main(logger)

dpacman/data/tfclust/api_download.py

@@ -1,448 +0,0 @@
-import requests
-from time import sleep
-import json
-import logging
-import multiprocessing
-from concurrent.futures import ThreadPoolExecutor, as_completed
-import os
-import pandas as pd
-
-def get_all_tfs(genome: str = "hg38"):
-    """
-    Get all the transcription factors from the appropriate encRegTfbsClusteredWithCells.genome.bed file.
-    Available in data_files/raw/tfclust for genomes hg38 and hg19
-    """
-    # Read raw file
-    raw_data = pd.read_csv(
-        "../../data_files/encode3TfbsClusteredWithCells.bed", sep="\t", header=None
-    )
-    raw_data.columns = ["chrom", "start", "end", "tf_name", "score", "cell_line"]
-
-    # Extract all unique TF names
-    all_tfs = encode_raw["tf_name"].unique().tolist()
-    logging.info(f"Found {len(all_tfs)} transcription factors in genome {genome}.")
-
-    return all_tfs
-
-
-def get_all_chroms(genome: str = "hg38", exclude: list=None, include: list=None, logger: logging.Logger=None):
-    """
-    Fetch all chromosome names for a genome.
-    Note: some chromosomes are in unexpected formats (e.g. there is 'chr15', but also 'chr15_ML143371v1_fix')
-    """
-    if logger is None:
-        logger = logging.getLogger(__name__)
-        
-    url = f"https://api.genome.ucsc.edu/list/chromosomes?genome={genome}"
-    try:
-        r = requests.get(url)
-        r.raise_for_status()
-    except:
-        raise ValueError(f"Failed to fetch all chromosomes for genome {genome}")
-
-    if include is not None and exclude is not None:
-        raise ValueError(f"Must pass EITHER exclude or include. Cannot pass both.")
-    
-    all_chroms = [chrom for chrom in r.json()["chromosomes"]]
-    if include:
-        logger.info(f"Including only the following chromosomes: {include}")
-        all_chroms = [chrom for chrom in all_chroms if chrom in include]
-    if exclude:
-        logger.info(f"Excluding the following chromosomes: {exclude}")
-        all_chroms = [chrom for chrom in all_chroms if not(chrom in exclude)]
-        
-    logger.info(f"Found {len(all_chroms)} chromosomes in genome {genome}.")
-
-    return all_chroms
-
-def fetch_tfbs_track(chrom: str, genome: str = "hg38", logger:logging.Logger=None):
-    """
-    Fetch raw data from the track encRegTfbsClustered.
-    Returns json data for the specified chromosome, where key information appears as follows:
-    "encRegTfbsClustered": [
-        {
-            "bin": 585,
-            "chrom": "chr1",
-            "chromStart": 9917,
-            "chromEnd": 10247,
-            "name": "NUFIP1",
-            "score": 680,
-            "sourceCount": 1,
-            "sourceIds": "1063",
-            "sourceScores": "680"
-        },...
-        ]
-
-    """
-    if logger is None:
-        logger = logging.getLogger(__name__)
-        
-    params = {"genome": genome, "track": "encRegTfbsClustered", "chrom": chrom}
-    url = f"https://api.genome.ucsc.edu/getData/track?genome={params['genome']};track={params['track']};chrom={params['chrom']}"
-    try:
-        r = requests.get(url)
-        r.raise_for_status()
-    except:
-        raise ValueError(
-            f"Failed to fetch encRegTfbsClustered for {chrom} in genome {genome}"
-        )
-
-    # Extract the output and save it
-    json_out_dir = f"../../data_files/raw/tfclust/encRegTfbsClustered_data/{genome}"
-    os.makedirs(json_out_dir, exist_ok=True)
-
-    # Save it
-    json_output = r.json()
-    with open(
-        f"{json_out_dir}/{params['genome']}_{params['track']}_{params['chrom']}.json",
-        "w",
-    ) as f:
-        json.dump(json_output, f, indent=4)
-
-    logger.info(
-        f"Saved to {json_out_dir}/{params['genome']}_{params['track']}_{params['chrom']}.json"
-    )
-    return json_output
-
-
-def get_sequence(
-    chrom: str,
-    start: int,
-    end: int,
-    flank5: int = 0,
-    flank3: int = 0,
-    genome: str = "hg38",
-    logger: logging.Logger=None
-):
-    """
-    Given genome, start position, end position, chromosome, and desired flank size, extract the raw DNA sequence
-    """
-    if logger is None:
-        logger = logging.getLogger(__name__)
-        
-    new_start = max(0, start - flank5)
-    new_end = end + flank3
-    region = f"{chrom}:{new_start}-{new_end}"
-    url = f"https://api.genome.ucsc.edu/getData/sequence?genome={genome};chrom={chrom};start={new_start};end={new_end}"
-    
-    try:
-        r = requests.get(url)
-        r.raise_for_status()
-    except:
-        raise ValueError(f"Failed to fetch sequence for {region} in genome {genome}")
-
-    results_dict = {
-        "chromStart": new_start,
-        "chromEnd": new_end,
-        "seq": r.json()["dna"],
-    }
-    return results_dict
-
-
-def extract_tfbs_with_context(
-    genome: str = "hg38",
-    flank5: int = 500,
-    flank3: int = 500,
-    control_run: bool = True,  # if there's a flank, whether to also run without flank
-    out_dir: str = "../../data_files/processed/tfclust",
-    allowed_tfs: list = None,  # e.g., ['CTCF', 'MAX']
-    chroms: list = None,
-    logger: logging.Logger = None 
-):
-    """
-    Loop through raw downloads and extract TF binding sites (bs) with flanks
-    Builds a DataFrame with all the available data for each TF. Columns = ["bin", "chrom", "chromStart", "chromEnd", "name", "score", "scoreCount", "sourceIds", "sourceScores", "seq", "seq_flanked", "chromStart_flanked", "chromEnd_flanked"]
-    """
-    # Prepare logger
-    if logger is None:
-        logger = logging.getLogger(__name__)
-    # Prepare to save output
-    os.makedirs(out_dir, exist_ok=True)
-
-    # Get chromosomes
-    if chroms is None:
-        logger.info(
-            "No chromosomes provided, fetching all chromosomes for the given genome..."
-        )
-        chroms = get_all_chroms(genome, logger = logger)
-    count = 0
-    # Initialize the final DF
-    results_cols = [
-        "bin",
-        "chrom",
-        "chromStart",
-        "chromEnd",
-        "name",
-        "score",
-        "scoreCount",
-        "sourceIds",
-        "sourceScores",
-        "seq",
-        "seq_flanked",
-        "chromStart_flanked",
-        "chromEnd_flanked",
-        "flank5",
-        "flank3",
-    ]
-    results_init = pd.DataFrame(columns=results_cols)
-
-    # Make a list of the types of runs we need
-    queries = [{"flank5": flank5, "flank3": flank3}]
-    if not ((flank5 == 0) and (flank3 == 0) and control_run):
-        queries.append({"type": "control", "flank5": 0, "flank3": 0})
-        queries[0]["type"] = "flank"
-    elif (flank5 == 0) and (flank3 == 0):
-        queries[0]["type"] = "control"
-
-    # For each chromosome, download the encRegTfbsClustered track, extract the features, and fetch the sequences
-    # Loop through chroms
-    for chrom in chroms:
-        chrom_write_count = 0
-        chrom_output_fname = f"{out_dir}/encRegTfbsClustered_{genome}_{chrom}.csv"
-        results_init.to_csv(
-            chrom_output_fname, index=False
-        )
-        logger.info(f"Fetching {chrom}...")
-        # Fetch the data json (has start and end positions in the chrom, but not the sequence)
-        try:
-            data = fetch_tfbs_track(chrom, genome=genome, logger=logger)
-            logger.info(f"  → Fetched {chrom} successfully")
-            features = data.get("encRegTfbsClustered", {})
-            logger.info(f"  → Found {len(features)} features")
-        except Exception as e:
-            logger.info(f"  Failed to fetch {chrom}: {e}")
-            continue
-
-        # Get the sequences of the DNA binding sites
-        for feature_no, feature in enumerate(features):
-            # Initialize new results row
-            new_row = {}
-
-            # Check if tf is valid
-            tf_name = feature.get("name", "UnknownTF")
-            if allowed_tfs and tf_name not in allowed_tfs:
-                logger.warning(f"TF name {tf_name} not in allowed_tfs. Skipping.")
-                continue
-            # Make sure the chromosomes match and we have the right sequence!
-            assert (
-                feature["chrom"] == chrom
-            ), f"Chromosome mismatch: {feature['chrom']} != {chrom}"
-
-            # Add all the cols already in the json, add
-            for c in results_cols:
-                if c in feature:
-                    new_row[c] = feature[c]
-
-            ### Extract sequence
-            start = feature["chromStart"]
-            end = feature["chromEnd"]
-
-            for query in queries:
-                try:
-                    results_dict = get_sequence(
-                        chrom,
-                        start,
-                        end,
-                        flank5=query["flank5"],
-                        flank3=query["flank3"],
-                        genome=genome,
-                        logger = logger
-                    )
-                    # Add the returned info
-                    if query["type"] == "control":
-                        new_row["seq"] = results_dict["seq"]
-                    elif query["type"] == "flank":
-                        new_row["seq_flanked"] = results_dict["seq"]
-                        new_row["chromStart_flanked"] = results_dict["chromStart"]
-                        new_row["chromEnd_flanked"] = results_dict["chromEnd"]
-                        new_row["flank5"] = flank5
-                        new_row["flank3"] = flank3
-                    logger.info(
-                        f" Success on feat. {feature_no} {chrom}:{start}-{end}, type {query['type']}"
-                    )
-                except Exception as e:
-                    logger.info(
-                        f"  Skipped feat. {feature_no} {chrom}:{start}-{end} due to error: {e}"
-                    )
-                    continue
-
-                sleep(0.05)  # Stay within UCSC's 20 req/sec rate limit
-
-            # Fill out any blank columns
-            try:
-                for c in results_cols:
-                    if c not in new_row:
-                        new_row[c] = None
-
-                new_row_df = pd.DataFrame(data=new_row, index=[0])
-                if new_row_df["seq"] is not None:
-                    new_row_df.to_csv(
-                        chrom_output_fname,
-                        mode="a",
-                        index=False,
-                        header=False,
-                    )
-                    logger.info(
-                        f"Wrote new row to {out_dir}/encRegTfbsClustered_{chrom}.csv"
-                    )
-                    chrom_write_count += 1
-                else:
-                    logger.info(f"Did not write new row. {new_row}")
-            except Exception as e:
-                logger.error(F"Failed to write new row to {out_dir}/encRegTfbsClustered_{chrom}.csv: error {e}")
-
-        logger.info(f"Done. Wrote {chrom_write_count} sequences to {out_dir}/encRegTfbsClustered_{chrom}.csv")
-        count += chrom_write_count
-    
-    logger.info(f"Done with all chroms. Wrote {count} sequences to {out_dir}.")
-    
-def setup_chrom_logger(chrom: str, genome: str, out_dir: str) -> logging.Logger:
-    """Set up a dedicated logger for a given chromosome."""
-    logger = logging.getLogger(f"{genome}_{chrom}")
-    logger.setLevel(logging.DEBUG)
-    logger.propagate = False
-
-    # Avoid duplicate handlers if reused
-    if not logger.handlers:
-        os.makedirs(out_dir, exist_ok=True)
-        log_path = os.path.join(out_dir, f"log_{genome}_{chrom}.log")
-        handler = logging.FileHandler(log_path, mode='w', encoding='utf-8')
-        formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
-        handler.setFormatter(formatter)
-        logger.addHandler(handler)
-
-    return logger
-
-# Thread function for one chromosome
-def process_chrom(
-    chrom: str = "chr1",
-    genome: str = "hg38",
-    flank5: int = 500,
-    flank3: int = 500,
-    control_run: bool = True,
-    out_dir: str = "../../data_files/processed/tfclust",
-    allowed_tfs: list = None,
-):
-    """
-    Called within parallel method to strat a thread
-    """
-    chrom_logger = setup_chrom_logger(chrom, genome, f"{out_dir}/logs")
-    chrom_logger.info(f"Starting thread for {chrom}")
-    
-    logging.info(f"Starting thread for {chrom}")
-    try:
-        extract_tfbs_with_context(
-            genome=genome,
-            flank5=flank5,
-            flank3=flank3,
-            control_run=control_run,
-            out_dir=out_dir,
-            allowed_tfs=allowed_tfs,
-            chroms=[chrom],  # important: wrap in list
-            logger=chrom_logger
-        )
-        chrom_logger.info(f"Finished {chrom}")
-    except Exception as e:
-        chrom_logger.error(f"Error processing {chrom}: {e}")
-
-
-import multiprocessing
-from concurrent.futures import ThreadPoolExecutor, as_completed
-
-def parallel_extract_tfbs_for_genome(
-    genome: str,
-    flank5: int,
-    flank3: int,
-    control_run: bool,
-    out_dir: str,
-    allowed_tfs: list,
-    chroms: list,
-    max_workers: int,
-):
-    logger = logging.getLogger(f"{genome}")
-    logger.info(f"Using {max_workers} threads for {genome}...")
-
-    if chroms is None:
-        chroms = get_all_chroms(genome=genome)
-
-    futures = {}
-    with ThreadPoolExecutor(max_workers=max_workers) as executor:
-        for chrom in chroms:
-            future = executor.submit(
-                process_chrom,
-                chrom=chrom,
-                genome=genome,
-                flank5=flank5,
-                flank3=flank3,
-                control_run=control_run,
-                out_dir=f"{out_dir}/{genome}",
-                allowed_tfs=allowed_tfs,
-            )
-            futures[future] = f"{genome}:{chrom}"
-
-        for future in as_completed(futures):
-            label = futures[future]
-            try:
-                future.result()
-            except Exception as e:
-                logger.error(f"{label} raised an exception: {e}")
-
-
-def parallel_extract_tfbs_with_context(
-    genomes=["hg38", "hg19"],
-    flank5=500,
-    flank3=500,
-    control_run=True,
-    out_dir="../../data_files/processed/tfclust",
-    allowed_tfs=None,
-    chroms=None,
-):
-    total_cpus = multiprocessing.cpu_count()
-    cpu_per_genome = total_cpus // len(genomes)
-
-    logging.info(f"Total CPUs: {total_cpus}")
-    logging.info(f"Launching {len(genomes)} genome pipelines with {cpu_per_genome} threads each")
-
-    processes = []
-    for genome in genomes:
-        p = multiprocessing.Process(
-            target=parallel_extract_tfbs_for_genome,
-            args=(
-                genome,
-                flank5,
-                flank3,
-                control_run,
-                out_dir,
-                allowed_tfs,
-                chroms,
-                cpu_per_genome
-            )
-        )
-        p.start()
-        processes.append(p)
-
-    for p in processes:
-        p.join()
-
-def main():
-    genomes = ["hg38", "hg19"]
-
-    parallel_extract_tfbs_with_context(
-        genomes=genomes,
-        flank5=500,
-        flank3=500,
-        control_run=True,  # if there's a flank, whether to also run without flank
-        out_dir=f"../../data_files/processed/tfclust",
-        allowed_tfs=None,  # e.g., ['CTCF', 'MAX']
-        chroms=None,
-    )
-
-if __name__ == "__main__":
-    logger = logging.getLogger(__name__)
-    logging.basicConfig(
-        filename="download.log",
-        encoding="utf-8",
-        level=logging.DEBUG,
-        filemode="w",
-    )
-    main()

dpacman/data/tfclust/combine.py

@@ -1,114 +0,0 @@
-import pandas as pd
-import logging
-import os
-import dask.dataframe as dd
-
-def combine_hg38_hg19(hg38_dir, hg19_dir):
-    # See how many files there are 
-    hg38_files = [os.path.join(hg38_dir,x) for x in os.listdir(hg38_dir) if os.path.isfile(os.path.join(hg38_dir,x))]
-    hg19_files = [os.path.join(hg19_dir,x) for x in os.listdir(hg19_dir) if os.path.isfile(os.path.join(hg19_dir,x))]
-
-    logging.info(f"Total hg38 files: {len(hg38_files)}")
-    logging.info(f"Total hg19 files: {len(hg19_files)}")
-    
-    # See how many datapoints there are 
-    hg38_complete = pd.read_csv(os.path.join(hg38_dir,"logs/completed.txt"), sep="\t")
-    hg19_complete = pd.read_csv(os.path.join(hg19_dir,"logs/completed.txt"), sep="\t")
-    
-    logging.info(f"Total hg38 rows: {sum(hg38_complete['row_count']):,}")
-    logging.info(f"Total hg19 rows: {sum(hg19_complete['row_count']):,}")  
-    logging.info(f"Total: {sum(hg38_complete['row_count']) + sum(hg19_complete['row_count']) :,}")
-    
-    # Now try to combine all the files into one 
-
-    # Read all CSVs in the folder as a single Dask dataframe
-    # Read each genome separately
-    full_df_hg38 = dd.read_csv(hg38_files)
-    full_df_hg38 = full_df_hg38.assign(genome="hg38")  # ✅ Dask-safe assignment
-
-    full_df_hg19 = dd.read_csv(hg19_files)
-    full_df_hg19 = full_df_hg19.assign(genome="hg19")
-
-    # Concatenate both into one Dask DataFrame
-    full_df = dd.concat([full_df_hg38, full_df_hg19])
-
-    logging.info(f"Added all files to ccombined DataFrame. Total rows: {len(full_df)}")
-    
-    full_df["seq_len"] = full_df["seq"].str.len()
-    full_df["seq_flanked_len"] = full_df["seq_flanked"].str.len()
-    logging.info(f"Added sequence length column.")
-    
-    full_df_dir = "../../data_files/processed/tfclust/combined"
-    full_df_savepath = os.path.join(full_df_dir, "encRegTfbsClustered_hg38_hg19.parquet")
-    os.makedirs(full_df_dir, exist_ok=True)
-    full_df.to_parquet(full_df_savepath)  # much faster and more compact
-    logging.info(f"Saved combined DataFrame to {full_df_savepath}.")
-
-    
-# Define the aggregation function
-def collapse_group(group):
-    return pd.Series({
-        "name": ",".join(group["name"]),
-        "score": ",".join(map(str, group["score"])),
-        "bin": ",".join(map(str, group["bin"])),
-        "scoreCount": ",".join(map(str, group["scoreCount"])),
-        "sourceIds": ",".join(map(str, group["sourceIds"])),
-        "sourceScores": ",".join(map(str, group["sourceScores"])),
-    })
-    
-def reorg_like_remap(genome_dir, fname):
-    """
-    Reorganize a chromosome from tfclust processing to be in the format of remap files:
-    #chrom,chromStart,chromEnd,name,score,strand,thickStart,thickEnd,reserved,chromLen,thickLen
-    
-    Original format of my processed tfclust files
-    bin,chrom,chromStart,chromEnd,name,score,scoreCount,sourceIds,sourceScores,seq,seq_flanked,chromStart_flanked,chromEnd_flanked,flank5,flank3
-    """
-    
-    fpath = os.path.join(genome_dir, fname)
-    df = dd.read_csv(fpath)
-    
-    # Show the head 
-    print(df.head())
-    
-    # Keep everything except the sequences 
-    df = df[[
-        "chrom", "chromStart", "chromEnd", "name", "score", # same as other file
-        "bin","scoreCount","sourceIds","sourceScores"
-    ]].rename(columns={"chrom":"#chrom"})
-
-    # Apply groupby with known output types (meta)
-    meta = {
-        "name": str,
-        "score": str,
-        "bin": str,
-        "scoreCount": str,
-        "sourceIds": str,
-        "sourceScores": str
-    }
-
-    grouped = df.groupby(["#chrom", "chromStart", "chromEnd"]).apply(collapse_group, meta=meta)
-
-    # You can now compute it
-    result = grouped.compute()
-    
-    # save the result 
-    savepath = os.path.join(genome_dir, "remap_reorg")
-    os.makedirs(savepath, exist_ok=True)
-    savepath = os.path.join(savepath, fname.replace(".csv", "_reorg.csv"))
-    result.to_csv(os.path.join(genome_dir), index=True)
-    logging.info(f"Saved reorganized file to {savepath}")
-    
-def main():
-    hg38_dir = "../../data_files/processed/tfclust/hg38"
-    hg19_dir = "../../data_files/processed/tfclust/hg19"
-
-    #combine_hg38_hg19(hg38_dir, hg19_dir)
-    
-    for chrom in ["chr1"]:
-        reorg_like_remap(hg38_dir, f"encRegTfbsClustered_hg38_{chrom}.csv")
-      
-if __name__ == "__main__":
-    logging.basicConfig(filename="combine.log", encoding="utf-8", level=logging.DEBUG, filemode="w")
-    
-    main()

dpacman/data/tfclust/download.py

@@ -1,462 +0,0 @@
-import os
-import logging
-import requests
-import pandas as pd
-import json
-import multiprocessing
-from math import ceil
-from datetime import datetime
-
-def get_all_tfs(genome: str = "hg38"):
-    raw_data = pd.read_csv(
-        f"../../data_files/encode3TfbsClusteredWithCells.bed", sep="\t", header=None
-    )
-    raw_data.columns = ["chrom", "start", "end", "tf_name", "score", "cell_line"]
-    all_tfs = raw_data["tf_name"].unique().tolist()
-    logging.info(f"Found {len(all_tfs)} transcription factors in genome {genome}.")
-    return all_tfs
-
-def get_all_chroms(genome: str = "hg38", exclude: list=None, include: list=None, logger: logging.Logger=None):
-    """
-    Fetch all chromosome names for a genome.
-    Note: some chromosomes are in unexpected formats (e.g. there is 'chr15', but also 'chr15_ML143371v1_fix')
-    """
-    if logger is None:
-        logger = logging.getLogger(__name__)
-        
-    url = f"https://api.genome.ucsc.edu/list/chromosomes?genome={genome}"
-    try:
-        r = requests.get(url)
-        r.raise_for_status()
-    except:
-        logger.error(f"Failed to fetch all chromosomes for genome {genome}")
-
-    if include is not None and exclude is not None:
-        raise ValueError(f"Must pass EITHER exclude or include. Cannot pass both.")
-    
-    all_chroms = list(r.json()["chromosomes"].keys())
-    if include is not None:
-        logger.info(f"Including only the following chromosomes: {include}")
-        all_chroms = [chrom for chrom in all_chroms if chrom in include]
-    if exclude is not None:
-        logger.info(f"Excluding the following chromosomes: {exclude}")
-        all_chroms = [chrom for chrom in all_chroms if not(chrom in exclude)]
-        
-    logger.info(f"Found {len(all_chroms)} chromosomes in genome {genome}.")
-
-    return all_chroms
-
-def get_all_chrom_fasta_files(genome: str = "hg38", exclude: list=None, include: list=None, logger: logging.Logger=None, out_dir="../../data_files/raw/genomes"):
-    """
-    Get FASTA files for each chromosome for a current genome
-    """
-    if logger is None:
-        logger = logging.getLogger(__name__)
-        
-    if include is not None and exclude is not None:
-        raise ValueError(f"Must pass EITHER exclude or include. Cannot pass both.")
-        
-    chroms = get_all_chroms(genome=genome, exclude=exclude, include=include, logger=logger)
-    
-    genome_out_dir = os.path.join(out_dir,genome)
-    os.makedirs(genome_out_dir, exist_ok=True)
-    
-    for chrom in chroms:
-        chrom_save_path = os.path.join(genome_out_dir,f"{genome}_{chrom}.json")
-        if not(os.path.exists(chrom_save_path)):
-            url = f"https://api.genome.ucsc.edu/getData/sequence?genome={genome};chrom={chrom}"
-            try:
-                r = requests.get(url)
-                r.raise_for_status()
-                json_output = r.json()
-                
-                with open(chrom_save_path, "w") as f:
-                    json.dump(json_output, f, indent=4)
-                
-                logger.info(f"Downloaded {chrom} in genome {genome}.")
-        
-            except:
-                logger.error(f"Failed to fetch all {chrom} for genome {genome}")
-        else:
-            logger.info(f"Already downloaded {chrom} in genome {genome}. Skipping.")
-            
-    logger.info(f"Downloaded {len(chroms)} chromosomes in genome {genome}.")
-    
-    return chroms
-
-def fetch_tfbs_track(chrom: str, genome: str = "hg38", logger:logging.Logger=None):
-    """
-    Fetch raw data from the track encRegTfbsClustered.
-    Returns json data for the specified chromosome, where key information appears as follows:
-    "encRegTfbsClustered": [
-        {
-            "bin": 585,
-            "chrom": "chr1",
-            "chromStart": 9917,
-            "chromEnd": 10247,
-            "name": "NUFIP1",
-            "score": 680,
-            "sourceCount": 1,
-            "sourceIds": "1063",
-            "sourceScores": "680"
-        },...
-        ]
-
-    """
-    if logger is None:
-        logger = logging.getLogger(__name__)
-        
-    params = {"genome": genome, "track": "encRegTfbsClustered", "chrom": chrom}
-    json_out_dir = os.path.join("../../data_files/raw/tfclust/encRegTfbsClustered_data", genome)
-    json_out_path = os.path.join(json_out_dir, f"{params['genome']}_{params['track']}_{params['chrom']}.json")
-    if not(os.path.exists(json_out_path)):
-        url = f"https://api.genome.ucsc.edu/getData/track?genome={params['genome']};track={params['track']};chrom={params['chrom']}"
-        try:
-            r = requests.get(url)
-            r.raise_for_status()
-            
-            # Extract the output and save it
-            os.makedirs(json_out_dir, exist_ok=True)
-
-            # Save it
-            json_output = r.json()
-            with open(json_out_path, "w") as f:
-                json.dump(json_output, f, indent=4)
-
-            logger.info(
-                f"Saved to {json_out_path}"
-            )
-        except:
-            logger.error(
-                f"Failed to fetch encRegTfbsClustered for {chrom} in genome {genome}"
-            )
-    else:
-        logging.info(f"Already downloaded encRegTfbsClustered for {chrom} in {genome}. Skipping download.")
-        with open(json_out_path, "r") as f:
-            json_output = json.load(f)
-
-    return json_output
-
-def get_sequence(
-    chrom_json: dict,
-    start: int,
-    end: int,
-    flank5: int = 0,
-    flank3: int = 0,
-    genome: str = "hg38",
-    logger: logging.Logger=None
-):
-    """
-    Given genome, start position, end position, chromosome json, and desired flank size, extract the raw DNA sequence
-    
-    chrom_json has keys: "downloadTime", "downloadTimeStamp","genome", "chrom", "start", "end", "dna"
-
-    """
-    if logger is None:
-        logger = logging.getLogger(__name__)
-        
-    chrom_seq = chrom_json["dna"]
-    chrom = chrom_json["chrom"]
-    if chrom_json["start"] != 0:
-        logger.warning(f"Start position of chromosome is not 0. Start position: {chrom_json['start']}")
-    
-    # Calculate new start and end indices
-    new_start = max(0, start - flank5)
-    new_end = end + flank3
-    if new_end > chrom_json["end"]:
-        logger.warning(f"Attempting to query {chrom} from {new_start} to {new_end}, but last index is {chrom_json['end']}. Manually setting last index to {chrom_json['end']}")
-        new_end = chrom_json['end']
-        
-    results_dict = {
-        "chromStart": new_start,
-        "chromEnd": new_end,
-        "seq": chrom_seq[new_start:new_end]# end is NOT inclusive!!
-    }
-    return results_dict
-
-def extract_tfbs_with_context(
-    genome: str,
-    flank5: int=500,
-    flank3: int=500,
-    allowed_tfs: list=None,
-    out_dir: str="../../data_files/processed/tfclust",
-    control_run: bool = True,  # if there's a flank, whether to also run without flank
-    chroms: list = None,
-    logger: logging.Logger=None,
-    redo: bool  = False, # whether to redo even if we've already processed this
-    idx: int=0  # index of worker
-):
-    """
-    Main method for a genome. By calling helpers, gets all chromosomes and their sequences, gets encRegTfbsClustered, and queries the feature indices in encRegTfbsClustered against chrom seqs for binding site sequences.
-    """
-    if logger is None:
-        logger = logging.getLogger(__name__)
-        
-    # Get all chromosomes for the current genome, including downloading thier sequences
-    if chroms is None:
-        all_chroms = get_all_chrom_fasta_files(genome=genome, logger=logger)
-    else:
-        all_chroms = get_all_chrom_fasta_files(
-            genome=genome,
-            exclude=[c for c in get_all_chroms(genome) if c not in chroms],
-            logger=logger
-        )
-    
-    # For each chrom, (1) download full fasta sequence, (2) download encRegTfbsClustered, (3) query features from [2] through [1]
-    # Initialize the final DF
-    results_cols = [
-        "bin",
-        "chrom",
-        "chromStart",
-        "chromEnd",
-        "name",
-        "score",
-        "scoreCount",
-        "sourceIds",
-        "sourceScores",
-        "seq",
-        "seq_flanked",
-        "chromStart_flanked",
-        "chromEnd_flanked",
-        "flank5",
-        "flank3",
-    ]
-    results_init = pd.DataFrame(columns=results_cols)
-    
-    # Make a list of the types of runs we need
-    queries = [{"flank5": flank5, "flank3": flank3}]
-    if not ((flank5 == 0) and (flank3 == 0) and control_run):
-        queries.append({"type": "control", "flank5": 0, "flank3": 0})
-        queries[0]["type"] = "flank"
-    elif (flank5 == 0) and (flank3 == 0):
-        queries[0]["type"] = "control"
-       
-    merged_done_txt_path = os.path.join("../../data_files/processed/tfclust", genome, "logs", f"completed.txt")
-    done_txt_path = os.path.join("../../data_files/processed/tfclust", genome, "logs", f"completed_worker_{idx}.txt")
-    if os.path.exists(merged_done_txt_path):
-        completed_chroms = pd.read_csv(merged_done_txt_path, sep="\t")
-        completed_chroms = list(completed_chroms["chrom"])
-    else:
-        completed_chroms = []
-        
-    with open(done_txt_path, "w") as f:
-        f.write("chrom\trow_count\n")
-
-    logger.info(f"{len(completed_chroms)} already complete: {','.join(completed_chroms)}")
-    
-    count = 0
-    # Iterate through chromosomes (1) download encRegTfbsClustered, (2) query each feature in the chrom sequence
-    for chrom in all_chroms:
-        chrom_write_count = 0
-        chrom_output_fname = os.path.join("../../data_files/processed/tfclust", genome, f"encRegTfbsClustered_{genome}_{chrom}.csv")
-        
-        # If we've already done it, no need
-        if chrom in completed_chroms and not(redo):
-            chrom_write_count = len(pd.read_csv(chrom_output_fname))
-            with open(done_txt_path, "a") as f:
-                f.write(f"{chrom}\t{chrom_write_count}\n")
-            continue
-        
-        #### If we ARE processing this, process it!
-        # Load chromosome sequence
-        with open(os.path.join("../../data_files/raw/genomes",genome,f"{genome}_{chrom}.json"), "r") as f:
-            chrom_json = json.load(f)
-        
-        results_init.to_csv(
-            chrom_output_fname, index=False
-        )
-        logger.info(f"Fetching {chrom}...")
-        
-        # Fetch the data json (has start and end positions in the chrom, but not the sequence)
-        try:
-            data = fetch_tfbs_track(chrom, genome=genome, logger=logger)
-            logger.info(f"  → Fetched {chrom} successfully")
-            features = data.get("encRegTfbsClustered", {})
-            logger.info(f"  → Found {len(features)} features")
-        except Exception as e:
-            logger.info(f"  Failed to fetch {chrom}: {e}")
-            continue
-
-        # Get the sequences of the DNA binding sites
-        for feature_no, feature in enumerate(features):
-            # Initialize new results row
-            new_row = {}
-
-            # Check if tf is valid
-            tf_name = feature.get("name", "UnknownTF")
-            if allowed_tfs and tf_name not in allowed_tfs:
-                logger.warning(f"TF name {tf_name} not in allowed_tfs. Skipping.")
-                continue
-            # Make sure the chromosomes match and we have the right sequence!
-            assert (
-                feature["chrom"] == chrom
-            ), f"Chromosome mismatch: {feature['chrom']} != {chrom}"
-
-            # Add all the cols already in the json, add
-            for c in results_cols:
-                if c in feature:
-                    new_row[c] = feature[c]
-
-            ### Extract sequence
-            start = feature["chromStart"]
-            end = feature["chromEnd"]
-
-            for query in queries:
-                results_dict = get_sequence(
-                    chrom_json,
-                    start,
-                    end,
-                    flank5=query["flank5"],
-                    flank3=query["flank3"],
-                    genome=genome,
-                    logger = logger
-                )
-                # Add the returned info
-                if query["type"] == "control":
-                    new_row["seq"] = results_dict["seq"]    # note: these sequences will have soft-masked repeats!
-                elif query["type"] == "flank":
-                    new_row["seq_flanked"] = results_dict["seq"]
-                    new_row["chromStart_flanked"] = results_dict["chromStart"]
-                    new_row["chromEnd_flanked"] = results_dict["chromEnd"]
-                    new_row["flank5"] = flank5
-                    new_row["flank3"] = flank3
-
-            # Fill out any blank columns
-            try:
-                for c in results_cols:
-                    if c not in new_row:
-                        new_row[c] = None
-
-                new_row_df = pd.DataFrame(data=new_row, index=[0])
-                new_row_df = new_row_df[results_cols]   # assert the right column order 
-                if new_row_df["seq"] is not None:
-                    new_row_df.to_csv(
-                        chrom_output_fname,
-                        mode="a",
-                        index=False,
-                        header=False,
-                    )
-                    logger.info(
-                        f"Wrote new row to {out_dir}/encRegTfbsClustered_{chrom}.csv"
-                    )
-                    chrom_write_count += 1
-                else:
-                    logger.info(f"Did not write new row. {new_row}")
-            except Exception as e:
-                logger.error(F"Failed to write new row to {out_dir}/encRegTfbsClustered_{chrom}.csv: error {e}")
-
-        logger.info(f"Done. Wrote {chrom_write_count} sequences to {out_dir}/encRegTfbsClustered_{chrom}.csv")
-        with open(done_txt_path, "a") as f:
-            f.write(f"{chrom}\t{chrom_write_count}\n")
-        count += chrom_write_count
-    
-    logger.info(f"Done with all chroms. Wrote {count} sequences to {out_dir}.")
-
-def merge_completed_files(genome: str):
-    """
-    Merge all completed_worker_*.txt files into a single completed.txt file
-    """
-    logs_dir = os.path.join("../../data_files/processed/tfclust", genome, "logs")
-    merged_path = os.path.join(logs_dir, "completed.txt")
-
-    with open(merged_path, "w") as outfile:
-        outfile.write("chrom\trow_count\n")  # header
-
-        for fname in os.listdir(logs_dir):
-            if fname.startswith("completed_worker_") and fname.endswith(".txt"):
-                with open(os.path.join(logs_dir, fname), "r") as infile:
-                    for line in infile:
-                        if line.startswith("chrom"):  # skip header lines
-                            continue
-                        outfile.write(line)
-
-    print(f"Merged completed_worker_*.txt into {merged_path}")
-
-def worker(args):
-    """
-    Worker function for parallel processing
-    """
-    # Extract args
-    chrom_group, idx, genome, flank5, flank3, logs_dir = args
-    os.makedirs(logs_dir, exist_ok=True)
-
-    # Define logger
-    logger = logging.getLogger(f"worker_{idx}")
-    logger.setLevel(logging.DEBUG)
-    logger.propagate = False
-
-    log_file = os.path.join(logs_dir, f"worker_{idx}.log")
-    fh = logging.FileHandler(log_file, mode="w", encoding="utf-8")
-    fh.setLevel(logging.DEBUG)
-    formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
-    fh.setFormatter(formatter)
-    logger.addHandler(fh)
-
-    logger.info(f"Starting worker {idx} for chromosomes: {chrom_group}")
-
-    extract_tfbs_with_context(
-        genome=genome,
-        flank5=flank5,
-        flank3=flank3,
-        allowed_tfs=None,
-        out_dir=f"../../data_files/processed/tfclust",
-        control_run=True,
-        chroms=chrom_group,
-        logger=logger,
-        idx=idx
-    )
-
-    logger.info(f"Finished worker {idx}")
-
-
-def parallel_extract(genome: str, flank5: int, flank3: int):
-    """
-    Run extract_tfbs_with_context in parallel for groups of chromosomes in the genome to speed up processing. 
-    """
-    chroms = get_all_chroms(genome)
-    num_cores = multiprocessing.cpu_count()
-
-    # Separate primary vs accessory chromosomes
-    primary_chroms = [c for c in chroms if "_" not in c]
-    accessory_chroms = [c for c in chroms if "_" in c]
-
-    # Distribute primary chromosomes round-robin across workers
-    chunks = [[] for _ in range(num_cores)]
-    for i, chrom in enumerate(primary_chroms):
-        chunks[i % num_cores].append(chrom)
-
-    # Now add accessory chromosomes to the least-loaded chunk (by count)
-    for chrom in accessory_chroms:
-        min_idx = min(range(num_cores), key=lambda i: len(chunks[i]))
-        chunks[min_idx].append(chrom)
-        
-    # Log how we split it up - want to see which chromosomes are in which chunks.
-    logging.info(f"{num_cores} CPU cores available. Primary chromosomes distributed round-robin.")
-    for chunk_no, chunk in enumerate(chunks):
-        logging.info(f"Chunk {chunk_no}. Chromosomes = {','.join(chunk)}")
-    
-    logs_dir = os.path.join("../../data_files/processed/tfclust", genome, "logs")
-    os.makedirs(logs_dir, exist_ok=True)
-
-    args_list = [(chunk, i, genome, flank5, flank3, logs_dir) for i, chunk in enumerate(chunks)]
-
-    with multiprocessing.Pool(processes=num_cores) as pool:
-        pool.map(worker, args_list)
-    
-    merge_completed_files(genome)
-
-def main():
-    genomes = ["hg38", "hg19"]
-    flank5 = 1000
-    flank3 = 1000
-    
-    # Iterate through genomes
-    for genome in genomes:
-        # Extract TF binding sites from bed - 500 flank
-        parallel_extract(genome, flank5, flank3)
-
-if __name__ == "__main__":
-    logging.basicConfig(filename="download.log", encoding="utf-8", level=logging.DEBUG, filemode="w")
-    logger = logging.getLogger(__name__)
-
-    main()

dpacman/data_tasks/clean/remap.py

@@ -0,0 +1,243 @@
+import pandas as pd
+from omegaconf import DictConfig
+from pathlib import Path
+import rootutils
+import logging
+import os
+
+root = rootutils.setup_root(__file__, indicator=".project-root", pythonpath=True)
+logger = logging.getLogger(__name__)
+
+
+def clean_nr(nr_raw_path: Path | str):
+    """
+    Clean the non-redundant peaks BED file.
+    Delete duplicate rows, assign columns, only keep columns we need.
+    """
+    nr = pd.read_csv(nr_raw_path, sep="\t", header=None)
+    nr.columns = [
+        "chrom",
+        "chromStart",
+        "chromEnd",
+        "biotypes",
+        "score",
+        "strand",
+        "thickStart",
+        "thickEnd",
+        "itemRgb",
+    ]
+
+    # make sure we correctly interpret column "biotype" as having one transcription factor separated from all relevant biotypes by ONE colon
+    biotype_colon_counts = (
+        nr["biotypes"]
+        .str.count(":")
+        .value_counts()
+        .reset_index()["biotypes"]
+        .unique()
+        .tolist()
+    )
+    assert biotype_colon_counts == [
+        1
+    ]  # confirm belief that : separates the name of a transcription factor from its biotype - just ONE biotype.
+
+    # then split the column accordingly into tr (transcriptional regulator) and biotypes
+    nr[["tr", "biotypes"]] = nr["biotypes"].str.split(":", expand=True)
+
+    # group and concat the scores
+    logger.info(
+        f"Keeping only the following columns: chrom, chromStart, chromEnd, biotypes, tr, score."
+    )
+    nr = nr[["chrom", "chromStart", "chromEnd", "biotypes", "score", "tr"]]
+
+    # drop duplicate rows - all fields
+    logger.info(f"Size of database before dropping duplicate rows: {len(nr)}")
+    nr = nr.drop_duplicates().reset_index(drop=True)
+    logger.info(f"Size of database after dropping duplicate rows: {len(nr)}")
+
+    # look for duplicate rows where it's clearly the same experiment but somehow different scores - chrom, chromStart, chromEnd, tr, biotypes
+    experiment_dups = len(
+        nr.loc[
+            nr.duplicated(subset=["chrom", "chromStart", "chromEnd", "tr", "biotypes"])
+        ]
+    )
+    logger.info(
+        f"{experiment_dups} total rows with same chrom, chromStart, chromEnd, biotypes, tr but different score."
+    )
+
+    logger.info(
+        f"Grouping by everything except score, comma-concatenating unique scores"
+    )
+    nr = (
+        nr.groupby(["chrom", "chromStart", "chromEnd", "tr", "biotypes"])
+        .agg({"score": lambda x: ",".join(map(str, sorted(set(x))))})
+        .reset_index()
+    )
+
+    logger.info(f"Final database size: {len(nr)}")
+
+    nr["chromLen"] = nr["chromEnd"] - nr["chromStart"]
+
+    return nr
+
+
+def clean_crm(crm_raw_path: Path | str):
+    """
+    Clean the CRM BED file.
+    Delete duplicate rows, assign columns, only keep columns we need.
+    """
+
+    crm = pd.read_csv(crm_raw_path, sep="\t", header=None)
+    crm.columns = [
+        "chrom",
+        "chromStart",
+        "chromEnd",
+        "tr",
+        "score",
+        "strand",
+        "thickStart",
+        "thickEnd",
+        "reserved",
+    ]
+
+    # group and concat the scores
+    logger.info(
+        f"Keeping only the following columns: chrom, chromStart, chromEnd, tr, score."
+    )
+    crm = crm[["chrom", "chromStart", "chromEnd", "tr", "score"]]
+
+    # drop duplicate rows - all fields
+    logger.info(f"Size of database before dropping duplicate rows: {len(crm)}")
+    crm = crm.drop_duplicates().reset_index(drop=True)
+    logger.info(f"Size of database after dropping duplicate rows: {len(crm)}")
+
+    # look for duplicate rows where it's clearly the same experiment but somehow different scores - chrom, chromStart, chromEnd, tr
+    experiment_dups = len(
+        crm.loc[crm.duplicated(subset=["chrom", "chromStart", "chromEnd", "tr"])]
+    )
+    logger.info(
+        f"{experiment_dups} total rows with same chrom, chromStart, chromEnd, tr but different score."
+    )
+
+    logger.info(
+        f"Grouping by everything except score, comma-concatenating unique scores"
+    )
+    crm = (
+        crm.groupby(["chrom", "chromStart", "chromEnd", "tr"])
+        .agg({"score": lambda x: ",".join(map(str, sorted(set(x))))})
+        .reset_index()
+    )
+
+    logger.info(f"Final database size: {len(crm)}")
+
+    crm["chromLen"] = crm["chromEnd"] - crm["chromStart"]
+
+    return crm
+
+
+def main(cfg: DictConfig):
+    # Define the paths
+    nr_raw_path = Path(root) / cfg.data_task.nr_raw_path
+    nr_processed_dir = Path(root) / cfg.data_task.nr_processed_dir
+    nr_processed_filename = cfg.data_task.nr_processed_filename
+    nr_savepath = os.path.join(nr_processed_dir, nr_processed_filename)
+
+    crm_raw_path = Path(root) / cfg.data_task.crm_raw_path
+    crm_processed_dir = Path(root) / cfg.data_task.crm_processed_dir
+    crm_processed_filename = cfg.data_task.crm_processed_filename
+    crm_savepath = os.path.join(crm_processed_dir, crm_processed_filename)
+
+    os.makedirs(nr_processed_dir, exist_ok=True)
+    os.makedirs(crm_processed_dir, exist_ok=True)
+
+    # Clean and save the non redundant peaks file
+    if not (os.path.exists(nr_savepath)):
+        nr_cleaned = clean_nr(nr_raw_path)
+        nr_cleaned.to_csv(nr_savepath, sep="\t", index=False)
+        logger.info(
+            f"Saved cleaned non-redundant peaks (NR) database to: {nr_savepath}"
+        )
+    else:
+        nr_cleaned = None
+        logger.info(f"File already exists at {nr_savepath}. Skipping")
+
+    # Clean and save the CRM file
+    if not (os.path.exists(crm_savepath)):
+        crm_cleaned = clean_crm(crm_raw_path)
+        crm_cleaned.to_csv(crm_savepath, sep="\t", index=False)
+        logger.info(
+            f"Saved cleaned cis-regulatory modules (CRM) database to: {crm_savepath}"
+        )
+    else:
+        crm_cleaned = None
+        logger.info(f"File already exists at {crm_savepath}. Skipping")
+
+    # Save example files
+    if cfg.data_task.save_example_files:
+        example_nr_dir = nr_processed_dir / "examples"
+        os.makedirs(example_nr_dir, exist_ok=True)
+        example_nr_savepath = os.path.join(
+            example_nr_dir, "example500_" + nr_processed_filename
+        )
+
+        if not (os.path.exists(example_nr_savepath)):
+            if nr_cleaned is None:
+                nr_cleaned = pd.read_csv(nr_savepath, sep="\t")
+            nr_cleaned.sample(n=500, random_state=42).reset_index(drop=True).to_csv(
+                example_nr_savepath, sep="\t", index=False
+            )
+            logger.info(
+                f"Saved example NR file with 500 rows to: {example_nr_savepath}"
+            )
+        else:
+            logger.info(
+                f"Example file already exists at {example_nr_savepath}. Skipping"
+            )
+
+        # CRM example
+        example_crm_dir = crm_processed_dir / "examples"
+        os.makedirs(example_crm_dir, exist_ok=True)
+        example_crm_savepath = os.path.join(
+            example_crm_dir, "example500_" + crm_processed_filename
+        )
+        if not (os.path.exists(example_crm_savepath)):
+            if crm_cleaned is None:
+                crm_cleaned = pd.read_csv(crm_savepath, sep="\t")
+            crm_cleaned.sample(n=500, random_state=42).reset_index(drop=True).to_csv(
+                example_crm_savepath, sep="\t", index=False
+            )
+            logger.info(
+                f"Saved example CRM file with 500 rows to: {example_crm_savepath}"
+            )
+        else:
+            logger.info(
+                f"Example file already exists at {example_crm_savepath}. Skipping"
+            )
+
+        # CRM example for one transcription factor
+        example_crm_tf_savepath = os.path.join(
+            example_crm_dir, "example500_ERG_" + crm_processed_filename
+        )
+        if not (os.path.exists(example_crm_tf_savepath)):
+            if crm_cleaned is None:
+                crm_cleaned = pd.read_csv(crm_savepath, sep="\t")
+            crm_example_tf_db = crm_cleaned.copy(deep=True)
+            crm_example_tf_db["tr"] = crm_example_tf_db["tr"].apply(
+                lambda x: x.split(",")
+            )
+            crm_example_tf_db = crm_example_tf_db.explode("tr").reset_index(drop=True)
+            crm_example_tf_db = crm_example_tf_db.loc[crm_example_tf_db["tr"] == "ERG"]
+            crm_example_tf_db = crm_example_tf_db.sample(
+                n=min(500, len(crm_example_tf_db)), random_state=42
+            ).reset_index(drop=True)
+            crm_example_tf_db.to_csv(example_crm_tf_savepath, sep="\t", index=False)
+            logger.info(
+                f"Saved example CRM file for one TF with 500 rows to: {example_crm_tf_savepath}"
+            )
+        else:
+            logger.info(
+                f"Example file already exists at {example_crm_tf_savepath}. Skipping"
+            )
+
+
+if __name__ == "__main__":
+    main()

dpacman/data_tasks/download/README.md

@@ -0,0 +1,26 @@
+This directory holds functions for downloading raw data needed to train the binding site predictor (classifier) on processed ChIP-seq peaks. 
+
+## Human genome
+`genome.py`
+* Download the sequences of all chromosomes from a certain genome (e.g. hg38, used for this project)
+* Configurations associated with this download can be found in `./configs/data_task/download/genome.yaml`
+
+### Running the download 
+To run this download, please change directory to `DPACMAN/dpacman` and run:
+
+```
+python -u -m scripts.preprocess data_task=download/genome
+```
+
+## ReMap 2022
+`remap.py`
+* Download non-redundant peaks: `remap2022_nr_macs2_hg38_v1_0.bed`
+* Download cis-regulatory modules (CRMS): `remap2022_crm_macs2_hg38_v1_0.bed`
+* Configurations associated with this download can be found in `./configs/data_task/download/remap.yaml`
+
+### Running the download 
+To run this download, please change directory to `DPACMAN/dpacman` and run:
+
+```
+python -u -m scripts.preprocess data_task=download/remap
+```

dpacman/data_tasks/download/download_unzip.sh

@@ -0,0 +1,41 @@
+#!/bin/bash
+
+URL="$1"
+DEST_DIR="$2"
+FILENAME="$3"        # e.g., intact.zip or biogrid.tab3.gz
+DELETE_ZIP="$4"      # "true" or "false"
+
+FILE="$DEST_DIR/$FILENAME"
+
+mkdir -p "$DEST_DIR"
+
+echo "Starting download from $URL..."
+wget "$URL" -O "$FILE"
+
+# Handle .zip files
+if [[ "$FILENAME" == *.zip ]]; then
+  echo "File is a .zip archive. Unzipping to $DEST_DIR in 10s..."
+  sleep 10
+  unzip "$FILE" -d "$DEST_DIR"
+
+  if [[ "$DELETE_ZIP" == "true" ]]; then
+    echo "delete_zip=true: removing $FILE"
+    rm -f "$FILE"
+  fi
+
+# Handle .gz files
+elif [[ "$FILENAME" == *.gz ]]; then
+  echo "File is a .gz archive. Extracting in 10s..."
+  sleep 10
+  gunzip -c "$FILE" > "${FILE%.gz}"
+
+  if [[ "$DELETE_ZIP" == "true" ]]; then
+    echo "delete_zip=true: removing $FILE"
+    rm -f "$FILE"
+  fi
+
+else
+  echo "File is not a .zip or .gz archive. Skipping extraction and deletion."
+fi
+
+echo "Done."

dpacman/data_tasks/download/genome.py

@@ -0,0 +1,233 @@
+"""
+Script for downloading the genome, hg38
+"""
+
+import rootutils
+
+root = rootutils.setup_root(__file__, indicator=".project-root", pythonpath=True)
+
+import os
+import logging
+import requests
+import json
+import hydra
+from omegaconf import DictConfig
+from pathlib import Path
+import logging
+import multiprocessing
+from hydra.core.hydra_config import HydraConfig
+
+base_logger = logging.getLogger(__name__)
+
+
+def get_all_chroms(
+    genome: str = "hg38",
+    exclude: list = None,
+    logger: logging.Logger = None,
+    include: list = None,
+):
+    """
+    Fetch all chromosome names for a genome.
+    Note: some chromosomes are in unexpected formats (e.g. there is 'chr15', but also 'chr15_ML143371v1_fix')
+    """
+    if logger is None:
+        logger = logging.getLogger(__name__)
+
+    url = f"https://api.genome.ucsc.edu/list/chromosomes?genome={genome}"
+    try:
+        r = requests.get(url)
+        r.raise_for_status()
+    except:
+        logger.error(f"Failed to fetch all chromosomes for genome {genome}")
+
+    if include is not None and exclude is not None:
+        raise ValueError(f"Must pass EITHER exclude or include. Cannot pass both.")
+
+    all_chroms = list(r.json()["chromosomes"].keys())
+    if include is not None:
+        logger.info(f"Including only the following chromosomes: {include}")
+        all_chroms = [chrom for chrom in all_chroms if chrom in include]
+    if exclude is not None:
+        logger.info(f"Excluding the following chromosomes: {exclude}")
+        all_chroms = [chrom for chrom in all_chroms if not (chrom in exclude)]
+
+    logger.info(f"Found {len(all_chroms)} chromosomes in genome {genome}.")
+
+    return all_chroms
+
+
+def get_all_chrom_fasta_files(
+    genome: str = "hg38",
+    exclude: list = None,
+    include: list = None,
+    logger: logging.Logger = None,
+    output_dir="../../data_files/raw/genomes",
+):
+    """
+    Get FASTA files for each chromosome for a current genome
+    """
+    if logger is None:
+        logger = logging.getLogger(__name__)
+
+    if include is not None and exclude is not None:
+        raise ValueError(f"Must pass EITHER exclude or include. Cannot pass both.")
+
+    chroms = get_all_chroms(
+        genome=genome, exclude=exclude, include=include, logger=logger
+    )
+
+    logger.info(f"Saving downloaded chromosomes to {output_dir}")
+    os.makedirs(output_dir, exist_ok=True)
+
+    for chrom in chroms:
+        chrom_save_path = os.path.join(output_dir, f"{genome}_{chrom}.json")
+        if not (os.path.exists(chrom_save_path)):
+            url = f"https://api.genome.ucsc.edu/getData/sequence?genome={genome};chrom={chrom}"
+            try:
+                r = requests.get(url)
+                r.raise_for_status()
+                json_output = r.json()
+
+                with open(chrom_save_path, "w") as f:
+                    json.dump(json_output, f, indent=4)
+
+                logger.info(
+                    f"Downloaded {chrom} in genome {genome}. Saved to: {chrom_save_path}"
+                )
+
+            except:
+                logger.error(f"Failed to fetch all {chrom} for genome {genome}")
+        else:
+            logger.info(f"Already downloaded {chrom} in genome {genome}. Skipping.")
+
+    logger.info(f"Downloaded {len(chroms)} chromosomes in genome {genome}.")
+
+    return chroms
+
+
+def merge_completed_files(genome: str, logs_dir: Path):
+    """
+    Merge all completed_worker_*.txt files into a single completed.txt file
+    """
+    merged_path = os.path.join(logs_dir, "completed.txt")
+
+    with open(merged_path, "w") as outfile:
+        outfile.write("chrom\trow_count\n")  # header
+
+        for fname in os.listdir(logs_dir):
+            if fname.startswith("completed_worker_") and fname.endswith(".txt"):
+                with open(os.path.join(logs_dir, fname), "r") as infile:
+                    for line in infile:
+                        if line.startswith("chrom"):  # skip header lines
+                            continue
+                        outfile.write(line)
+
+    print(f"Merged completed_worker_*.txt into {merged_path}")
+
+
+def worker(args):
+    """
+    Worker function for parallel processing
+    """
+    # Extract args
+    chrom_group, idx, genome, logs_dir, output_dir = args
+    os.makedirs(logs_dir, exist_ok=True)
+
+    # Define logger
+    wlogger = logging.getLogger(f"worker_{idx}")
+    wlogger.setLevel(logging.DEBUG)
+    wlogger.propagate = False
+
+    log_file = os.path.join(logs_dir, f"worker_{idx}.log")
+    fh = logging.FileHandler(log_file, mode="w", encoding="utf-8")
+    fh.setLevel(logging.DEBUG)
+    formatter = logging.Formatter("%(asctime)s - %(levelname)s - %(message)s")
+    fh.setFormatter(formatter)
+    wlogger.addHandler(fh)
+
+    wlogger.info(f"Starting worker {idx} for chromosomes: {chrom_group}")
+
+    all_chroms = get_all_chrom_fasta_files(
+        genome=genome, include=chrom_group, logger=wlogger, output_dir=output_dir
+    )
+
+    wlogger.info(f"Finished worker {idx}")
+
+
+def parallel_extract(
+    genome: str,
+    include: list = None,
+    exclude: list = None,
+    output_dir: Path = None,
+    logs_dir: Path = None,
+):
+    """
+    Run extract_tfbs_with_context in parallel for groups of chromosomes in the genome to speed up processing.
+    """
+    # Get all chromosomes whose sequences we want to download.
+    chroms = get_all_chroms(genome, exclude=exclude, include=include)
+    num_cores = multiprocessing.cpu_count() - 1
+
+    # Separate primary vs accessory chromosomes
+    primary_chroms = [c for c in chroms if "_" not in c]
+    accessory_chroms = [c for c in chroms if "_" in c]
+
+    base_logger.info(f"Total primary chromosomes: {len(primary_chroms)}")
+    for pc in primary_chroms:
+        base_logger.info(pc)
+    base_logger.info(f"Total accessory chromosomes: {len(accessory_chroms)}")
+    for ac in accessory_chroms:
+        base_logger.info(ac)
+
+    # Distribute primary chromosomes round-robin across workers
+    chunks = [[] for _ in range(num_cores)]
+    for i, chrom in enumerate(primary_chroms):
+        chunks[i % num_cores].append(chrom)
+
+    # Now add accessory chromosomes to the least-loaded chunk (by count)
+    for chrom in accessory_chroms:
+        min_idx = min(range(num_cores), key=lambda i: len(chunks[i]))
+        chunks[min_idx].append(chrom)
+
+    # Log how we split it up - want to see which chromosomes are in which chunks.
+    logging.info(
+        f"{num_cores} CPU cores available (leaving 1 empty). Primary chromosomes distributed round-robin."
+    )
+    for chunk_no, chunk in enumerate(chunks):
+        logging.info(f"Chunk {chunk_no}. Chromosomes = {','.join(chunk)}")
+
+    args_list = [
+        (chunk, i, genome, logs_dir, output_dir) for i, chunk in enumerate(chunks)
+    ]
+
+    with multiprocessing.Pool(processes=num_cores) as pool:
+        pool.map(worker, args_list)
+
+    merge_completed_files(genome, logs_dir)
+
+
+def main(cfg: DictConfig):
+    include = cfg.get("include", None)
+    exclude = cfg.get("exclude", None)
+
+    output_dir = Path(root) / cfg.data_task.output_dir
+    os.makedirs(output_dir, exist_ok=True)
+
+    # Download the sequences of all chromosomes
+    for genome in cfg.data_task.genomes:
+        base_logger.info(f"Downloading all chromsoomes for genome {genome}")
+
+        # Make a subfolder for this specific genome and its logs
+        genome_output_dir = output_dir / genome
+        genome_logs_dir = Path(HydraConfig.get().run.dir) / genome / "logs"
+        parallel_extract(
+            genome,
+            include=include,
+            exclude=exclude,
+            output_dir=genome_output_dir,
+            logs_dir=genome_logs_dir,
+        )
+
+
+if __name__ == "__main__":
+    main()

dpacman/data_tasks/download/remap.py

@@ -0,0 +1,87 @@
+from omegaconf import DictConfig
+from pathlib import Path
+import logging
+import subprocess
+import os
+
+import rootutils
+
+root = rootutils.setup_root(__file__, indicator=".project-root", pythonpath=True)
+
+logger = logging.getLogger(__name__)
+
+
+def recap(result):
+    """
+    Print out info about the success of the download job.
+    """
+    logger.info("STDOUT:\n" + result.stdout)
+    if result.stderr:
+        logger.warning("STDERR:\n" + result.stderr)
+
+    if result.returncode != 0:
+        logger.error(f"Download script exited with code {result.returncode}")
+    else:
+        logger.info("Download completed successfully.")
+
+
+def main(cfg: DictConfig):
+    """
+    Download IntAct file, which is one zip file
+    """
+
+    script_path = root / "dpacman/data_tasks/download/download_unzip.sh"
+    delete_zip = str(cfg.data_task.get("delete_zip", False)).lower()
+    assert delete_zip in ("true", "false")
+
+    nr_url = cfg.data_task.nr_url
+    nr_output_dir = root / cfg.data_task.nr_output_dir
+    nr_filename = cfg.data_task.nr_filename
+
+    logger.info(f"Running {cfg.data_task.type} for {cfg.data_task.name}")
+
+    ##### Non-redundant peaks download #####
+    logger.info(f"Script: {script_path}")
+    logger.info(f"Non-Redundant Peaks - URL: {nr_url}")
+    logger.info(f"Non-Redundant Peaks - Output: {nr_output_dir / nr_filename}")
+
+    os.makedirs(nr_output_dir, exist_ok=True)
+
+    # Run the download.sh script as a subproces
+    result = subprocess.run(
+        ["bash", str(script_path), nr_url, str(nr_output_dir), nr_filename, delete_zip],
+        capture_output=True,
+        text=True,
+    )
+
+    recap(result)
+
+    ##### CRMs download #####
+    crm_url = cfg.data_task.crm_url
+    crm_output_dir = root / cfg.data_task.crm_output_dir
+    crm_filename = cfg.data_task.crm_filename
+
+    logger.info(f"CRMs - URL: {crm_url}")
+    logger.info(f"CRMs - Output: {crm_output_dir / crm_filename}")
+
+    os.makedirs(crm_output_dir, exist_ok=True)
+
+    # Run the download.sh script as a subproces
+    result = subprocess.run(
+        [
+            "bash",
+            str(script_path),
+            crm_url,
+            str(crm_output_dir),
+            crm_filename,
+            delete_zip,
+        ],
+        capture_output=True,
+        text=True,
+    )
+
+    recap(result)
+
+
+if __name__ == "__main__":
+    main()

dpacman/{data → data_tasks/embeddings}/compute_embeddings.py

@@ -14,6 +14,7 @@ Usage example (DNA + protein in one go):
     --out-dir          ../data_files/processed/tfclust/hg38_tf/embeddings \
     --device           cuda
 """
+
 import os
 import re
 import argparse
@@ -28,6 +29,7 @@ import time
 
 # ---- model wrappers ----
 
+
 class CaduceusEmbedder:
     def __init__(self, device, chunk_size=131_072, overlap=0):
         """
@@ -39,12 +41,14 @@ class CaduceusEmbedder:
         self.tokenizer = AutoTokenizer.from_pretrained(
             model_name, trust_remote_code=True
         )
-        self.model = AutoModel.from_pretrained(
-            model_name, trust_remote_code=True
-        ).to(device).eval()
-        self.device     = device
+        self.model = (
+            AutoModel.from_pretrained(model_name, trust_remote_code=True)
+            .to(device)
+            .eval()
+        )
+        self.device = device
         self.chunk_size = chunk_size
-        self.step       = chunk_size - overlap
+        self.step = chunk_size - overlap
 
     def embed(self, seqs):
         """
@@ -78,11 +82,11 @@ class CaduceusEmbedder:
                 return_tensors="pt",
                 padding=False,
                 truncation=True,
-                max_length=self.chunk_size
+                max_length=self.chunk_size,
             ).to(self.device)
             with torch.no_grad():
                 out = self.model(**toks).last_hidden_state  # (1, L, D)
-            outputs.append(out.cpu().numpy()[0])             # (L, D)
+            outputs.append(out.cpu().numpy()[0])  # (L, D)
 
         return np.stack(outputs, axis=0)  # (N, L, D)
 
@@ -106,21 +110,29 @@ class CaduceusEmbedder:
             t1 = time.perf_counter()
             print(f"  length={sz:6,d}  time={(t1-t0)*1000:7.1f} ms")
 
+
 class DNABertEmbedder:
     def __init__(self, device):
-        self.tokenizer = AutoTokenizer.from_pretrained("zhihan1996/DNA_bert_6", trust_remote_code=True)
-        self.model     = AutoModel.from_pretrained("zhihan1996/DNA_bert_6", trust_remote_code=True).to(device)
-        self.device    = device
+        self.tokenizer = AutoTokenizer.from_pretrained(
+            "zhihan1996/DNA_bert_6", trust_remote_code=True
+        )
+        self.model = AutoModel.from_pretrained(
+            "zhihan1996/DNA_bert_6", trust_remote_code=True
+        ).to(device)
+        self.device = device
 
     def embed(self, seqs):
         embs = []
         for s in seqs:
-            tokens = self.tokenizer(s, return_tensors="pt", padding=True)["input_ids"].to(self.device)
+            tokens = self.tokenizer(s, return_tensors="pt", padding=True)[
+                "input_ids"
+            ].to(self.device)
             with torch.no_grad():
                 out = self.model(tokens).last_hidden_state.mean(1)
             embs.append(out.cpu().numpy())
         return np.vstack(embs)
 
+
 class NucleotideTransformerEmbedder:
     def __init__(self, device):
         # HF “feature-extraction” returns a list of (L, D) arrays for each input
@@ -128,7 +140,9 @@ class NucleotideTransformerEmbedder:
         self.pipe = pipeline(
             "feature-extraction",
             model="InstaDeepAI/nucleotide-transformer-500m-1000g",
-            device= -1 if device=="cpu" else 0    # HF uses -1 for CPU, 0 for GPU #:contentReference[oaicite:0]{index=0}
+            device=(
+                -1 if device == "cpu" else 0
+            ),  # HF uses -1 for CPU, 0 for GPU #:contentReference[oaicite:0]{index=0}
         )
 
     def embed(self, seqs):
@@ -138,8 +152,9 @@ class NucleotideTransformerEmbedder:
         """
         all_embeddings = self.pipe(seqs, truncation=True, padding=True)
         # all_embeddings is a List of shape (L, D) arrays
-        pooled = [ np.mean(x, axis=0) for x in all_embeddings ]
-        return np.vstack(pooled)       
+        pooled = [np.mean(x, axis=0) for x in all_embeddings]
+        return np.vstack(pooled)
+
 
 class ESMEmbedder:
     def __init__(self, device):
@@ -157,12 +172,15 @@ class ESMEmbedder:
         reps = results["representations"][33]
         return reps[:, 1:-1].mean(1).cpu().numpy()
 
+
 class ESMDBPEmbedder:
     def __init__(self, device):
-        base_model, alphabet = esm.pretrained.esm1b_t33_650M_UR50S()  
+        base_model, alphabet = esm.pretrained.esm1b_t33_650M_UR50S()
         model_path = (
             Path(__file__).resolve().parent.parent
-            / "pretrained" / "ESM-DBP" / "ESM-DBP.model"
+            / "pretrained"
+            / "ESM-DBP"
+            / "ESM-DBP.model"
         )
         checkpoint = torch.load(model_path, map_location="cpu")
         clean_sd = {}
@@ -189,6 +207,7 @@ class ESMDBPEmbedder:
         # skip start/end tokens
         return reps[:, 1:-1].mean(1).cpu().numpy()
 
+
 class GPNEmbedder:
     def __init__(self, device):
         model_name = "songlab/gpn-msa-sapiens"
@@ -200,16 +219,14 @@ class GPNEmbedder:
 
     def embed(self, seqs):
         inputs = self.tokenizer(
-            seqs,
-            return_tensors="pt",
-            padding=True,
-            truncation=True
+            seqs, return_tensors="pt", padding=True, truncation=True
         ).to(self.device)
 
         with torch.no_grad():
             last_hidden = self.model(**inputs).last_hidden_state
         return last_hidden.mean(dim=1).cpu().numpy()
 
+
 class ProGenEmbedder:
     def __init__(self, device):
         model_name = "jinyuan22/ProGen2-base"
@@ -219,28 +236,34 @@ class ProGenEmbedder:
 
     def embed(self, seqs):
         inputs = self.tokenizer(
-            seqs,
-            return_tensors="pt",
-            padding=True,
-            truncation=True
+            seqs, return_tensors="pt", padding=True, truncation=True
         ).to(self.device)
         with torch.no_grad():
             last_hidden = self.model(**inputs).last_hidden_state
         return last_hidden.mean(dim=1).cpu().numpy()
 
+
 # ---- main pipeline ----
 
+
 def get_embedder(name, device, for_dna=True):
     name = name.lower()
     if for_dna:
-        if name=="caduceus":   return CaduceusEmbedder(device)
-        if name=="dnabert":    return DNABertEmbedder(device)
-        if name=="nucleotide": return NucleotideTransformerEmbedder(device)
-        if name=="gpn":        return GPNEmbedder(device)
+        if name == "caduceus":
+            return CaduceusEmbedder(device)
+        if name == "dnabert":
+            return DNABertEmbedder(device)
+        if name == "nucleotide":
+            return NucleotideTransformerEmbedder(device)
+        if name == "gpn":
+            return GPNEmbedder(device)
     else:
-        if name in ("esm",):    return ESMEmbedder(device)
-        if name in ("esm-dbp","esm_dbp"): return ESMDBPEmbedder(device)
-        if name=="progen":      return ProGenEmbedder(device)
+        if name in ("esm",):
+            return ESMEmbedder(device)
+        if name in ("esm-dbp", "esm_dbp"):
+            return ESMDBPEmbedder(device)
+        if name == "progen":
+            return ProGenEmbedder(device)
     raise ValueError(f"Unknown model {name} (for_dna={for_dna})")
 
 
@@ -250,23 +273,34 @@ def embed_and_save(seqs, ids, embedder, out_path):
     with open(out_path.with_suffix(".ids"), "w") as f:
         f.write("\n".join(ids))
 
-if __name__=="__main__":
+
+if __name__ == "__main__":
 
     p = argparse.ArgumentParser()
-    p.add_argument("--genome-json-dir", default="data_files/raw/genomes/hg38", help="dir of UCSC JSONs")
-    p.add_argument("--skip-dna", action="store_true", help="if set, skip the chromosome embedding step") #if glm embeddings successful but not plm embeddings
-    p.add_argument("--tf-fasta",      required=True, help="input TF FASTA file")
-    p.add_argument("--chrom-model",   default="caduceus")
-    p.add_argument("--tf-model",      default="esm-dbp")
-    p.add_argument("--out-dir",       default="data_files/processed/tfclust/hg38_tf/embeddings")
-    p.add_argument("--device",        default="cpu")
+    p.add_argument(
+        "--genome-json-dir",
+        default="data_files/raw/genomes/hg38",
+        help="dir of UCSC JSONs",
+    )
+    p.add_argument(
+        "--skip-dna",
+        action="store_true",
+        help="if set, skip the chromosome embedding step",
+    )  # if glm embeddings successful but not plm embeddings
+    p.add_argument("--tf-fasta", required=True, help="input TF FASTA file")
+    p.add_argument("--chrom-model", default="caduceus")
+    p.add_argument("--tf-model", default="esm-dbp")
+    p.add_argument(
+        "--out-dir", default="data_files/processed/tfclust/hg38_tf/embeddings"
+    )
+    p.add_argument("--device", default="cpu")
     args = p.parse_args()
 
     os.makedirs(args.out_dir, exist_ok=True)
     device = args.device
 
     if not args.skip_dna:
-        #Load only primary chromosome JSONs (chr1–22, X, Y, M)
+        # Load only primary chromosome JSONs (chr1–22, X, Y, M)
         genome_dir = Path(args.genome_json_dir)
         chrom_seqs, chrom_ids = [], []
         primary_pattern = re.compile(r"^hg38_chr(?:[1-9]|1[0-9]|2[0-2]|X|Y|M)\.json$")
@@ -274,13 +308,17 @@ if __name__=="__main__":
             if not primary_pattern.match(j.name):
                 continue
             data = json.loads(j.read_text())
-            seq  = data.get("dna") or data.get("sequence")
+            seq = data.get("dna") or data.get("sequence")
             chrom = data.get("chrom") or j.stem.split("_")[-1]
             chrom_seqs.append(seq)
             chrom_ids.append(chrom)
         ########################
         cutoff = CaduceusEmbedder(device).chunk_size
-        long_chroms = [(chrom, len(seq)) for chrom, seq in zip(chrom_ids, chrom_seqs) if len(seq) > cutoff]
+        long_chroms = [
+            (chrom, len(seq))
+            for chrom, seq in zip(chrom_ids, chrom_seqs)
+            if len(seq) > cutoff
+        ]
         if long_chroms:
             print("⚠️ Chromosomes exceeding Caduceus max tokens ({}):".format(cutoff))
             for chrom, L in long_chroms:
@@ -290,10 +328,10 @@ if __name__=="__main__":
 
         ####################
         chrom_embedder = get_embedder(args.chrom_model, device, for_dna=True)
-        out_chrom = Path(args.out_dir)/f"chrom_{args.chrom_model}.npy"
+        out_chrom = Path(args.out_dir) / f"chrom_{args.chrom_model}.npy"
         embed_and_save(chrom_seqs, chrom_ids, chrom_embedder, out_chrom)
 
-    #Load TF sequences
+    # Load TF sequences
     tf_seqs, tf_ids = [], []
     for record in SeqIO.parse(args.tf_fasta, "fasta"):
         tf_ids.append(record.id)
@@ -304,4 +342,4 @@ if __name__=="__main__":
     out_tf = Path(args.out_dir) / f"tf_{args.tf_model}.npy"
     embed_and_save(tf_seqs, tf_ids, tf_embedder, out_tf)
 
-    print("Done.")
+    print("Done.")

dpacman/{data/remap → data_tasks/fimo}/post_fimo.py

@@ -7,11 +7,12 @@ import numpy as np
 
 # ─────────────────────────────────────────────────────────────────────────────
 # PATHS — edit these if needed
-INPUT_CSV  = "/home/a03-akrishna/DPACMAN/data_files/processed/post_fimo.csv"
+INPUT_CSV = "/home/a03-akrishna/DPACMAN/data_files/processed/post_fimo.csv"
 OUTPUT_CSV = "/home/a03-akrishna/DPACMAN/data_files/processed/final.csv"
-JSON_DIR   = "/home/a03-svincoff/DPACMAN/dpacman/data_files/raw/genomes/hg38"
+JSON_DIR = "/home/a03-svincoff/DPACMAN/dpacman/data_files/raw/genomes/hg38"
 # ─────────────────────────────────────────────────────────────────────────────
 
+
 def load_chrom_dna(chrom, cache):
     """Load & cache the full chromosome 'dna' string from hg38_chr{chrom}.json."""
     if chrom in cache:
@@ -24,32 +25,34 @@ def load_chrom_dna(chrom, cache):
     cache[chrom] = data["dna"]
     return cache[chrom]
 
+
 def sigmoid_array(arr: np.ndarray) -> np.ndarray:
     """Elementwise logistic sigmoid → values in (0,1)."""
     return 1.0 / (1.0 + np.exp(-arr))
 
+
 def main():
     # 1) load post‐FIMO results
     df = pd.read_csv(INPUT_CSV)
 
     dna_cache = {}
-    records   = []
+    records = []
 
     # 2) for each TF‐peak row, extract sequence & build scores
     for _, row in df.iterrows():
-        tfid      = row["TF_id"]
-        chrom     = str(row["#chrom"])
-        cstart    = int(row["contextStart"])
-        cend      = int(row["contextEnd"])
-        peak_s    = int(row["ChIPStart"])
-        peak_e    = int(row["ChIPEnd"])
+        tfid = row["TF_id"]
+        chrom = str(row["#chrom"])
+        cstart = int(row["contextStart"])
+        cend = int(row["contextEnd"])
+        peak_s = int(row["ChIPStart"])
+        peak_e = int(row["ChIPEnd"])
         chipscore = int(row["chipscore"])
-        jaspar    = str(row["jaspar"])
+        jaspar = str(row["jaspar"])
 
         # pull out the exact context sequence (including any Ns)
         dna = load_chrom_dna(chrom, dna_cache)
         seq = dna[cstart:cend]
-        L   = len(seq)
+        L = len(seq)
 
         # initialize base‐resolution scores
         scores = np.zeros(L, dtype=int)
@@ -68,37 +71,50 @@ def main():
                 scores[hs_i:he_i] = chipscore + 100
 
         # stringify the raw scores
-        score_str    = ",".join(map(str, scores.tolist()))
+        score_str = ",".join(map(str, scores.tolist()))
 
         # sigmoid‐transform
-        sig_vals     = sigmoid_array(scores.astype(float))
-        score_sig    = ",".join(f"{v:.4f}" for v in sig_vals.tolist())
-
-        records.append({
-            "TF_id":        tfid,
-            "dna_sequence": seq,
-            "score_str":    score_str,
-            "score_sig_r2": score_sig
-        })
+        sig_vals = sigmoid_array(scores.astype(float))
+        score_sig = ",".join(f"{v:.4f}" for v in sig_vals.tolist())
+
+        records.append(
+            {
+                "TF_id": tfid,
+                "dna_sequence": seq,
+                "score_str": score_str,
+                "score_sig_r2": score_sig,
+            }
+        )
 
     # 3) assemble into a DataFrame
     final_df = pd.DataFrame.from_records(records)
 
     # 4) drop any exact TF+DNA duplicates
-    final_df = final_df.drop_duplicates(subset=["TF_id","dna_sequence"]).reset_index(drop=True)
+    final_df = final_df.drop_duplicates(subset=["TF_id", "dna_sequence"]).reset_index(
+        drop=True
+    )
 
     # 5) assign random IDs
-    tf_map  = {tf: uuid.uuid4().hex[:8] for tf in final_df["TF_id"].unique()}
+    tf_map = {tf: uuid.uuid4().hex[:8] for tf in final_df["TF_id"].unique()}
     dna_map = {sq: uuid.uuid4().hex[:8] for sq in final_df["dna_sequence"].unique()}
 
-    final_df["tf_seq_id"]  = final_df["TF_id"].map(tf_map)
+    final_df["tf_seq_id"] = final_df["TF_id"].map(tf_map)
     final_df["dna_seq_id"] = final_df["dna_sequence"].map(dna_map)
-    final_df["ID"]         = final_df["tf_seq_id"] + "_" + final_df["dna_seq_id"]
+    final_df["ID"] = final_df["tf_seq_id"] + "_" + final_df["dna_seq_id"]
 
     # 6) reorder and write out
-    cols = ["TF_id","tf_seq_id","dna_sequence","dna_seq_id","score_str","score_sig_r2","ID"]
+    cols = [
+        "TF_id",
+        "tf_seq_id",
+        "dna_sequence",
+        "dna_seq_id",
+        "score_str",
+        "score_sig_r2",
+        "ID",
+    ]
     final_df[cols].to_csv(OUTPUT_CSV, index=False)
     print(f"Wrote {len(final_df)} rows → {OUTPUT_CSV}")
 
+
 if __name__ == "__main__":
     main()

dpacman/data_tasks/fimo/pre_fimo.py

@@ -0,0 +1,72 @@
+#!/usr/bin/env python3
+import pandas as pd
+import numpy as np
+import rootutils
+import logging
+import os
+from omegaconf import DictConfig
+from pathlib import Path
+
+root = rootutils.setup_root(__file__, indicator=".project-root", pythonpath=True)
+logger = logging.getLogger(__name__)
+
+
+def main(cfg: DictConfig):
+    # 1) load
+    input_path = Path(root) / cfg.data_task.input_csv
+    df = pd.read_csv(input_path, sep="\t")
+
+    # 2) normalize chromosomes and exclude non-whole chromosomes
+    df["chrom"] = df["chrom"].str.replace(r"^chr", "", regex=True)
+
+    valid = [str(i) for i in range(1, 23)] + ["X", "Y"]
+    df = df[df["chrom"].isin(valid)].reset_index(drop=True)
+
+    # 3) explode TF names
+    df["tr_list"] = df["tr"].str.split(",")
+    df = df.explode("tr_list").rename(columns={"tr_list": "TR"})
+    df["TR"] = df["TR"].str.strip()
+
+    # 4) draw a random left‐flank between 0 and WINDOW_TOTAL,
+    #    then right‐flank is whatever remains to sum to WINDOW_TOTAL
+    n = len(df)
+    df["left_context"] = np.random.randint(0, cfg.data_task.window_total + 1, size=n)
+    df["right_context"] = cfg.data_task.window_total - df["left_context"]
+
+    # 5) compute contextStart / contextEnd
+    df["contextStart"] = (
+        (df["chromStart"] - df["left_context"]).clip(lower=0).astype(int)
+    )
+    df["contextEnd"] = (df["chromEnd"] + df["right_context"]).astype(int)
+
+    # 6) assemble output
+    out = df[
+        [
+            "chrom",
+            "contextStart",
+            "chromStart",  # original ChIPStart
+            "chromEnd",  # original ChIPEnd
+            "contextEnd",
+            "score",  # original score column
+            "TF",
+        ]
+    ].rename(
+        columns={
+            "chrom": "#chrom",
+            "chromStart": "ChIPStart",
+            "chromEnd": "ChIPEnd",
+            "score": "chipscore",
+        }
+    )
+
+    # 7 make folder for tsv
+    output_path = Path(root) / cfg.data_task.output_csv
+    os.makedirs(output_path.parent, exist_ok=True)
+
+    # 8) write tsv
+    out.to_csv(output_path, sep="\t", index=False)
+    print(f"Wrote {len(out)} rows to {output_path}")
+
+
+if __name__ == "__main__":
+    main()

dpacman/{data/remap → data_tasks/fimo}/run_fimo.py

@@ -5,63 +5,50 @@ import subprocess
 import pandas as pd
 from multiprocessing import Pool, cpu_count
 from tqdm import tqdm
+import rootutils
+import logging
+from omegaconf import DictConfig
+from pathlib import Path
 
-# ─────────────────────────────────────────────────────────────────────────────
-# CONFIG — edit these paths if needed
-INPUT_CSV         = "/home/a03-akrishna/DPACMAN/data_files/processed/clean_pre_fimo.csv"
-OUTPUT_CSV        = "/home/a03-akrishna/DPACMAN/data_files/processed/post_fimo.csv"
-JSON_DIR          = "/home/a03-svincoff/DPACMAN/dpacman/data_files/raw/genomes/hg38"
+root = rootutils.setup_root(__file__, indicator=".project-root", pythonpath=True)
+logger = logging.getLogger(__name__)
 
-# Full paths to MEME‐suite binaries
-FIMO_BIN          = "/home/a03-svincoff/meme/bin/fimo"
-FASTA_GET_MARKOV  = "/home/a03-svincoff/meme/libexec/meme-5.5.8/fasta-get-markov"
 
-# JASPAR MEME file
-MOTIF_FILE        = "/home/a03-svincoff/DPACMAN/dpacman/softwares/meme-5.5.8/tests/common/JASPAR_CORE_2014_vertebrates.meme"
-
-# Working filenames
-SEQ_FASTA         = "to_scan.fa"
-BG_MODEL          = "bg_model.txt"
-FIMO_OUTDIR       = "fimo_out"
-
-# FIMO parameters
-PVAL_THRESH       = 1e-4
-MAX_STORED        = 1000000
-
-# How many parallel FIMO jobs (defaults to all cores)
-N_JOBS            = cpu_count()
-# ─────────────────────────────────────────────────────────────────────────────
-
-def load_chrom_dna(chrom, cache):
+def load_chrom_dna(chrom, cache, json_dir):
     if chrom in cache:
         return cache[chrom]
-    fname = os.path.join(JSON_DIR, f"hg38_chr{chrom}.json")
+    fname = os.path.join(json_dir, f"hg38_chr{chrom}.json")
     if not os.path.isfile(fname):
         raise FileNotFoundError(f"Chrom JSON not found: {fname}")
     with open(fname) as f:
         cache[chrom] = json.load(f)["dna"]
     return cache[chrom]
 
-def extract_sequences(df):
+
+def extract_sequences(df, seq_fasta, json_dir):
     dna_cache = {}
-    with open(SEQ_FASTA, "w") as fa:
+    with open(seq_fasta, "w") as fa:
         for idx, row in df.iterrows():
             chrom = str(row["#chrom"])
-            dna   = load_chrom_dna(chrom, dna_cache)
+            dna = load_chrom_dna(chrom, dna_cache, json_dir)
             start = int(row["contextStart"])
-            end   = int(row["contextEnd"])
-            seq   = dna[start:end]
+            end = int(row["contextEnd"])
+            seq = dna[start:end]
             fa.write(f">{idx}\n{seq}\n")
 
-def run_markov():
-    subprocess.check_call([FASTA_GET_MARKOV, SEQ_FASTA, BG_MODEL],
-                          stdout=subprocess.DEVNULL,
-                          stderr=subprocess.DEVNULL)
 
-def split_fasta(n_chunks):
+def run_markov(fasta_get_markov, seq_fasta, bg_model):
+    subprocess.check_call(
+        [fasta_get_markov, seq_fasta, bg_model],
+        stdout=subprocess.DEVNULL,
+        stderr=subprocess.DEVNULL,
+    )
+
+
+def split_fasta(n_chunks, seq_fasta):
     """Round-robin split SEQ_FASTA into chunked FASTA files."""
-    out_handles = [open(f"to_scan_{i}.fa","w") for i in range(n_chunks)]
-    with open(SEQ_FASTA) as inf:
+    out_handles = [open(f"to_scan_{i}.fa", "w") for i in range(n_chunks)]
+    with open(seq_fasta) as inf:
         header = None
         seq_lines = []
         for line in inf:
@@ -83,78 +70,139 @@ def split_fasta(n_chunks):
         o.close()
     return [f"to_scan_{i}.fa" for i in range(n_chunks)]
 
-def run_fimo_chunk(args):
-    """Run FIMO on one FASTA chunk."""
-    chunk_id, fasta_path = args
-    outdir = f"{FIMO_OUTDIR}_{chunk_id}"
+
+def run_fimo_chunk(cfg):
+    """
+    Run FIMO for a chunk.
+    Args:
+        cfg: dict with keys:
+            - chunk_id
+            - fasta_path
+            - fimo_outdir
+            - fimo_bin
+            - bg_model
+            - max_stored
+            - motif_file
+            - pval_thresh
+    """
+    outdir = f"{cfg['fimo_outdir']}_{cfg['chunk_id']}"
     os.makedirs(outdir, exist_ok=True)
-    print(f"▶ Chunk {chunk_id} starting FIMO", flush=True)
-    subprocess.check_call([
-        FIMO_BIN,
-        "--oc",                outdir,
-        "--bgfile",            BG_MODEL,
-        "--max-stored-scores", str(MAX_STORED),
-        "--thresh",            str(PVAL_THRESH),
-        MOTIF_FILE,
-        fasta_path
-    ])
-    print(f"▶ Chunk {chunk_id} finished", flush=True)
+    logger.info(f"Chunk {cfg['chunk_id']} starting FIMO")
+    subprocess.check_call(
+        [
+            cfg["fimo_bin"],
+            "--oc",
+            outdir,
+            "--bgfile",
+            cfg["bg_model"],
+            "--max-stored-scores",
+            str(cfg["max_stored"]),
+            "--thresh",
+            str(cfg["pval_thresh"]),
+            cfg["motif_file"],
+            cfg["fasta_path"],
+        ]
+    )
+    logger.info(f"Chunk {cfg['chunk_id']} finished")
     return os.path.join(outdir, "fimo.tsv")
 
+
 def annotate_with_fimo(df, fimo_tsv):
     fdf = pd.read_csv(fimo_tsv, sep="\t", comment="#")
     fdf["idx"] = fdf["sequence_name"].astype(int)
-    fdf = fdf.merge(df[["idx","contextStart"]], on="idx", how="left")
+    fdf = fdf.merge(df[["idx", "contextStart"]], on="idx", how="left")
     fdf["genomic_start"] = fdf["contextStart"] + fdf["start"] - 1
-    fdf["genomic_end"]   = fdf["contextStart"] + fdf["stop"]
+    fdf["genomic_end"] = fdf["contextStart"] + fdf["stop"]
     fdf["coord"] = (
-        fdf["genomic_start"].astype(str)
-        + "-" +
-        fdf["genomic_end"].astype(str)
+        fdf["genomic_start"].astype(str) + "-" + fdf["genomic_end"].astype(str)
     )
     agg = fdf.groupby("idx")["coord"].agg(lambda hits: ",".join(hits))
     df["jaspar"] = df["idx"].map(agg).fillna("")
     return df
 
-def main():
+
+def main(cfg: DictConfig):
+    """
+    Main method for running FIMO analysis, searching JASPAR motifs against ChIP-seq peaks
+    """
+    # 0) configs
+    paths = cfg.data_task.paths
+    fimo = cfg.data_task.fimo
+    fnames = cfg.data_task.fnames
+    meme = cfg.data_task.meme
+
+    # set njobs to max or whatever # is specified by user
+    njobs = fimo.njobs
+    if njobs == "max":
+        njobs = cpu_count() - 1
+    else:
+        njobs = min(cpu_count() - 1, int(njobs))
+
     # 1) load & explode
-    df = pd.read_csv(INPUT_CSV, low_memory=False)
-    df = df.reset_index().rename(columns={"index":"idx"})
+    input_csv_path = Path(root) / paths.input_csv
+    df = pd.read_csv(input_csv_path, low_memory=False)
+    df = df.reset_index().rename(columns={"index": "idx"})
     df["TF_occurrence"] = df.groupby("TF").cumcount() + 1
-    df["TF_id"]         = df["TF"] + "_seq" + df["TF_occurrence"].astype(str)
+    df["TF_id"] = df["TF"] + "_seq" + df["TF_occurrence"].astype(str)
 
     # 2) extract sequences & build BG model
-    extract_sequences(df)
-    print("▶ Building background model…", flush=True)
-    run_markov()
+    extract_sequences(df, fnames.seq_fasta, paths.json_dir)
+    logger.info("Building background model…")
+    run_markov(meme.fasta_get_markov, fnames.seq_fasta, fnames.bg_model)
 
     # 3) chunk FASTA and run FIMO in parallel
-    chunks = split_fasta(N_JOBS)
-    print(f"▶ Running FIMO in parallel ({N_JOBS} jobs)…", flush=True)
-    with Pool(N_JOBS) as pool:
-        tsv_paths = list(tqdm(
-            pool.imap(run_fimo_chunk, enumerate(chunks)),
-            total=len(chunks),
-            desc="FIMO chunks",
-            leave=True
-        ))
+    chunks = split_fasta(njobs)
+    chunk_cfgs = [
+        dict(
+            chunk_id=i,
+            fasta_path=chunk,
+            fimo_outdir=fnames.fimo_outdir,
+            fimo_bin=paths.fimo_bin,
+            bg_model=fnames.bg_model,
+            max_stored=fimo.max_stored,
+            motif_file=meme.jaspar_motif_file,
+            pval_thresh=fimo.pval_thresh,
+        )
+        for i, chunk in enumerate(chunks)
+    ]
+    logger.info(f"Running FIMO in parallel ({njobs} jobs)…")
+    with Pool(njobs) as pool:
+        tsv_paths = list(
+            tqdm(
+                pool.imap(run_fimo_chunk, chunk_cfgs),
+                total=len(chunks),
+                desc="FIMO chunks",
+                leave=True,
+            )
+        )
 
     # 4) merge chunked TSVs
-    combined = pd.concat([
-        pd.read_csv(tsv, sep="\t", comment="#")
-        for tsv in tsv_paths
-    ], ignore_index=True)
+    combined = pd.concat(
+        [pd.read_csv(tsv, sep="\t", comment="#") for tsv in tsv_paths],
+        ignore_index=True,
+    )
     merged_tsv = "fimo_combined.tsv"
     combined.to_csv(merged_tsv, sep="\t", index=False)
 
     # 5) annotate & write final CSV
     df = annotate_with_fimo(df, merged_tsv)
-    final = df[[
-        "#chrom","contextStart","ChIPStart","ChIPEnd",
-        "contextEnd","chipscore","TF","TF_id","jaspar"
-    ]]
-    final.to_csv(OUTPUT_CSV, index=False)
-    print(f"▶ Wrote {len(final)} rows → {OUTPUT_CSV}")
+    final = df[
+        [
+            "#chrom",
+            "contextStart",
+            "ChIPStart",
+            "ChIPEnd",
+            "contextEnd",
+            "chipscore",
+            "TF",
+            "TF_id",
+            "jaspar",
+        ]
+    ]
+    output_csv_path = Path(root) / paths.output_csv
+    final.to_csv(output_csv_path, index=False)
+    logger.info(f"Wrote {len(final)} rows → {output_csv_path}")
+
 
 if __name__ == "__main__":
     main()

dpacman/{data → data_tasks/visualize}/visualizations.py

@@ -5,6 +5,7 @@ import glob
 import re
 from pathlib import Path
 
+
 def trim_sequence(seq: str, seq_flanked: str, total_len: int):
     """
     Return a substring of seq_flanked of length total_len that contains seq
@@ -34,47 +35,48 @@ def process_and_plot(input_csv: str, total_len: int, output_csv: Path, fig_dir:
     ups, downs, abs_pos, rel_pos = [], [], [], []
     trimmed_seqs = []
     for _, row in df.iterrows():
-        trimmed, u, d = trim_sequence(row['seq'], row['seq_flanked'], total_len)
+        trimmed, u, d = trim_sequence(row["seq"], row["seq_flanked"], total_len)
         trimmed_seqs.append(trimmed)
         ups.append(u)
         downs.append(d)
         abs_pos.append(u)
-        rel_pos.append(u / (total_len - len(row['seq'])))
+        rel_pos.append(u / (total_len - len(row["seq"])))
     df_out = df.copy()
-    df_out['seq_trimmed'] = trimmed_seqs
-    df_out['motif_abs_start'] = abs_pos
-    df_out['motif_rel_pos'] = rel_pos
+    df_out["seq_trimmed"] = trimmed_seqs
+    df_out["motif_abs_start"] = abs_pos
+    df_out["motif_rel_pos"] = rel_pos
     df_out.to_csv(output_csv, index=False)
 
     basename = input_csv.stem
     # Absolute position histogram
-    plt.figure(figsize=(6,4))
-    plt.hist(df_out['motif_abs_start'], bins=50, edgecolor='k')
-    plt.title(f'{basename}: Absolute Motif Start')
-    plt.xlabel('Start Index (nt)')
-    plt.ylabel('Count')
+    plt.figure(figsize=(6, 4))
+    plt.hist(df_out["motif_abs_start"], bins=50, edgecolor="k")
+    plt.title(f"{basename}: Absolute Motif Start")
+    plt.xlabel("Start Index (nt)")
+    plt.ylabel("Count")
     plt.tight_layout()
     plt.savefig(fig_dir / f"{basename}_abs.png")
     plt.close()
     # Relative position histogram
-    plt.figure(figsize=(6,4))
-    plt.hist(df_out['motif_rel_pos'], bins=50, edgecolor='k')
-    plt.title(f'{basename}: Relative Motif Position')
-    plt.xlabel('Relative Position')
-    plt.ylabel('Count')
+    plt.figure(figsize=(6, 4))
+    plt.hist(df_out["motif_rel_pos"], bins=50, edgecolor="k")
+    plt.title(f"{basename}: Relative Motif Position")
+    plt.xlabel("Relative Position")
+    plt.ylabel("Count")
     plt.tight_layout()
     plt.savefig(fig_dir / f"{basename}_rel.png")
     plt.close()
 
-if __name__ == '__main__':
+
+if __name__ == "__main__":
     # === USER SETTINGS ===
-    PATTERN      = '/home/a03-svincoff/DPACMAN/dpacman/data_files/processed/tfclust/hg38/encRegTfbsClustered_hg38_chr*.csv'
-    CHR_FILTER   = re.compile(
-        r'encRegTfbsClustered_hg38_chr([1-9]|1[0-9]|2[0-2]|X|Y)\.csv$'
+    PATTERN = "/home/a03-svincoff/DPACMAN/dpacman/data_files/processed/tfclust/hg38/encRegTfbsClustered_hg38_chr*.csv"
+    CHR_FILTER = re.compile(
+        r"encRegTfbsClustered_hg38_chr([1-9]|1[0-9]|2[0-2]|X|Y)\.csv$"
     )
-    DESIRED_LEN  = 1000
-    OUTPUT_DIR   = Path('trimmed_csvs')
-    FIG_DIR      = Path('figures')
+    DESIRED_LEN = 1000
+    OUTPUT_DIR = Path("trimmed_csvs")
+    FIG_DIR = Path("figures")
     # =====================
 
     OUTPUT_DIR.mkdir(exist_ok=True)

dpacman/scripts/preprocess.py

@@ -0,0 +1,58 @@
+import rootutils
+
+root = rootutils.setup_root(__file__, indicator=".project-root", pythonpath=True)
+
+import hydra
+from omegaconf import DictConfig
+import logging
+
+logger = logging.getLogger(__name__)
+
+# import your processing entry points here
+from dpacman.data_tasks.download.genome import main as download_genome_main
+from dpacman.data_tasks.download.remap import main as download_remap_main
+from dpacman.data_tasks.clean.remap import main as clean_remap_main
+from dpacman.data_tasks.fimo.pre_fimo import main as pre_fimo_main
+from dpacman.data_tasks.fimo.run_fimo import main as run_fimo_main
+from dpacman.data_tasks.fimo.post_fimo import main as post_fimo_main
+
+
+@hydra.main(
+    config_path=str(root / "configs"), config_name="preprocess", version_base="1.3"
+)
+def main(cfg: DictConfig):
+    task_type = cfg.data_task.type
+    task_name = cfg.data_task.name.lower()
+
+    logger.info(f"Running {task_type} task: {task_name}")
+
+    if task_type == "download":
+        if task_name == "genome":
+            download_genome_main(cfg)
+        elif task_name == "remap":
+            download_remap_main(cfg)
+        else:
+            raise ValueError(f"No download pipeline defined for: {task_name}")
+
+    elif task_type == "clean":
+        if task_name == "remap":
+            clean_remap_main(cfg)
+        else:
+            raise ValueError(f"No clean pipeline defined for: {task_name}")
+
+    elif task_type == "fimo":
+        if task_name == "pre_fimo":
+            pre_fimo_main(cfg)
+        elif task_name == "run_fimo":
+            run_fimo_main(cfg)
+        elif task_name == "post_fimo":
+            post_fimo_main(cfg)
+        else:
+            raise ValueError(f"No clean pipeline defined for: {task_name}")
+
+    else:
+        raise ValueError(f"Unknown task type: {task_type}")
+
+
+if __name__ == "__main__":
+    main()

dpacman/scripts/run_download.sh

@@ -0,0 +1,16 @@
+#!/bin/bash
+
+# Manually specify values used in the config
+main_task="preprocess"
+data_task_type="download"
+timestamp=$(date "+%Y-%m-%d_%H-%M-%S")
+
+run_dir="$HOME/DPACMAN/logs/${main_task}/${data_task_type}/runs/${timestamp}"
+mkdir -p "$run_dir"
+
+CUDA_VISIBLE_DEVICES=0 nohup python -u -m scripts.preprocess \
+    hydra.run.dir="${run_dir}" \
+    data_task=${data_task_type}/remap \
+    > "${run_dir}/run.log" 2>&1 &
+
+echo $! > "${run_dir}/pid.txt"

dpacman/scripts/run_fimo.sh

@@ -0,0 +1,16 @@
+#!/bin/bash
+
+# Manually specify values used in the config
+main_task="preprocess"
+data_task_type="fimo"
+timestamp=$(date "+%Y-%m-%d_%H-%M-%S")
+
+run_dir="$HOME/DPACMAN/logs/${main_task}/${data_task_type}/runs/${timestamp}"
+mkdir -p "$run_dir"
+
+CUDA_VISIBLE_DEVICES=0 nohup python -u -m scripts.preprocess \
+    hydra.run.dir="${run_dir}" \
+    data_task=${data_task_type}/pre_fimo \
+    > "${run_dir}/run.log" 2>&1 &
+
+echo $! > "${run_dir}/pid.txt"

environment.yaml

@@ -26,9 +26,13 @@ dependencies:
   - pip>=23
   - pip:
     - rootutils
+    - hydra-core==1.3.2         # Hydra for config management
+    - hydra-colorlog==1.2.0     # Allow colorful logging in Hydra
+    - omegaconf==2.3.0          # Required by hydra-core
     - pandas==2.2.3
     - lxml==5.3.0
     - pymex==0.9.31
     - gitpython==3.1.44
-    - matplotlib
+    - black==25.1.0   # code formatter
+    - matplotlib==3.10.3
     - -e .