Add parquet_to_npy utility for converting local HuggingFace parquet files to .npy; update pipeline configuration and README

README.md

@@ -58,7 +58,8 @@ FOXES
 │   ├── download_sdo.py          # Download SDO/AIA EUV images from JSOC
 │   ├── sxr_downloader.py        # Download GOES SXR flux data
 │   ├── hugging_face_data_download.py  # Download pre-processed data from HuggingFace Hub
-│   └── hf_download_config.yaml  # Config for HuggingFace downloader
+│   ├── parquet_to_npy.py        # Convert locally-downloaded HF parquet files to .npy
+│   └── hf_download_config.yaml  # Config for HuggingFace downloader and parquet_to_npy
 ├── forecasting                  # Model training and inference
 │   ├── data_loaders
 │   │   ├── SDOAIA_dataloader.py # PyTorch Lightning DataModule for AIA+SXR
@@ -115,6 +116,7 @@ FOXES uses a single orchestrator script (`run_pipeline.py`) and a top-level conf
 | # | Step | Description                                                                    |
 |---|------|--------------------------------------------------------------------------------|
 | 0 | `hf_download` | Download pre-processed, pre-split data from HuggingFace *(replaces steps 1–6)* |
+| 0b | `parquet_to_npy` | Convert already-downloaded HF parquet files to `.npy` *(skips network download)* |
 | 1 | `download_aia` | Download SDO/AIA EUV images from JSOC                                          |
 | 2 | `download_sxr` | Download GOES SXR flux data                                                    |
 | 3 | `combine_sxr` | Combine raw GOES `.nc` files into per-satellite CSVs                           |
@@ -138,6 +140,9 @@ python run_pipeline.py --config pipeline_config.yaml --steps all
 # Quick-start: download pre-processed data from HuggingFace, then train
 python run_pipeline.py --config pipeline_config.yaml --steps hf_download,train,inference,evaluate
 
+# Already have parquet files locally? Convert them to .npy, then train
+python run_pipeline.py --config pipeline_config.yaml --steps parquet_to_npy,train,inference,evaluate
+
 # Run specific steps
 python run_pipeline.py --config pipeline_config.yaml --steps train,inference,evaluate
 
@@ -187,6 +192,32 @@ Run the downloader standalone:
 python download/hugging_face_data_download.py --config download/hf_download_config.yaml
 ```
 
+### Converting Local Parquet Files to .npy
+
+If you've already downloaded the HuggingFace parquet files (e.g., via `huggingface-cli` or the HF web UI), use `parquet_to_npy.py` to convert them directly — no network connection needed. The output is identical to what `hf_download` produces.
+
+```bash
+# All splits at once — parquet_root should contain train/, validation/, test/ subdirs
+python download/parquet_to_npy.py \
+    --parquet_root /path/to/parquet \
+    --config download/hf_download_config.yaml
+
+# Single split
+python download/parquet_to_npy.py \
+    --parquet_dir /path/to/parquet/train \
+    --split train \
+    --aia_dir /Volumes/T9/AIA_hg_processed \
+    --sxr_dir /Volumes/T9/SXR_hg_processed
+```
+
+Configure it via `pipeline_config.yaml` to use it as a pipeline step:
+
+```yaml
+parquet_to_npy:
+  config: "download/hf_download_config.yaml"  # provides aia_dir, sxr_dir, num_workers
+  parquet_root: "/path/to/your/parquet"        # dir with train/, validation/, test/ subdirs
+```
+
 ### Configuration
 
 Edit `pipeline_config.yaml` to set data paths, date ranges, and hyperparameters. Each step has its own section, and an `overrides` block lets you override values from the step's base config without editing it directly.

download/parquet_to_npy.py

@@ -0,0 +1,182 @@
+"""
+Convert Local HuggingFace Parquet Files to .npy
+================================================
+Same output layout as hugging_face_data_download.py, but reads from
+parquet files you've already downloaded instead of streaming from HF.
+
+Expected parquet columns: filename, aia_stack, sxr_value
+
+Usage:
+    # Convert one split at a time (parquet files flat in a directory)
+    python download/parquet_to_npy.py \\
+        --parquet_dir /path/to/parquet/train \\
+        --split train \\
+        --config download/hf_download_config.yaml
+
+    # Or specify output dirs directly
+    python download/parquet_to_npy.py \\
+        --parquet_dir /path/to/parquet/validation \\
+        --split validation \\
+        --aia_dir /Volumes/T9/AIA_hg_processed \\
+        --sxr_dir /Volumes/T9/SXR_hg_processed
+
+    # Auto-discover split subdirs (train/, validation/, test/) under a root
+    python download/parquet_to_npy.py \\
+        --parquet_root /path/to/parquet \\
+        --config download/hf_download_config.yaml
+"""
+
+import argparse
+import os
+import sys
+import time
+from concurrent.futures import ThreadPoolExecutor, as_completed
+from pathlib import Path
+
+import numpy as np
+import pyarrow.parquet as pq
+import yaml
+
+
+HF_TO_LOCAL = {"validation": "val"}
+
+
+def load_config(path: str) -> dict:
+    with open(path) as f:
+        return yaml.safe_load(f)
+
+
+def _write_arrays(filename: str, aia_arr: np.ndarray, sxr_arr: np.ndarray,
+                  aia_split_dir: str, sxr_split_dir: str) -> bool:
+    """Save arrays to disk. Returns True if written, False if already exists."""
+    aia_path = os.path.join(aia_split_dir, filename)
+    sxr_path = os.path.join(sxr_split_dir, filename)
+
+    if os.path.exists(aia_path) and os.path.exists(sxr_path):
+        return False
+
+    np.save(aia_path, aia_arr)
+    np.save(sxr_path, sxr_arr)
+    return True
+
+
+def convert_split(parquet_dir: str, hf_split: str, aia_base: str, sxr_base: str,
+                  num_workers: int = 8, print_every: int = 500):
+    local_split = HF_TO_LOCAL.get(hf_split, hf_split)
+    aia_split_dir = os.path.join(aia_base, local_split)
+    sxr_split_dir = os.path.join(sxr_base, local_split)
+    os.makedirs(aia_split_dir, exist_ok=True)
+    os.makedirs(sxr_split_dir, exist_ok=True)
+
+    parquet_files = sorted(Path(parquet_dir).glob("*.parquet"))
+    if not parquet_files:
+        print(f"No parquet files found in {parquet_dir}", file=sys.stderr)
+        return
+
+    print(f"\n{'='*50}")
+    print(f"Converting split: {hf_split} -> local dir: {local_split}")
+    print(f"{'='*50}")
+    print(f"  Parquet dir: {parquet_dir} ({len(parquet_files)} files)")
+    print(f"  AIA -> {aia_split_dir}")
+    print(f"  SXR -> {sxr_split_dir}")
+
+    saved = skipped = submitted = 0
+    start = time.time()
+
+    with ThreadPoolExecutor(max_workers=num_workers) as pool:
+        futures = {}
+
+        for pq_file in parquet_files:
+            table = pq.read_table(pq_file, columns=["filename", "aia_stack", "sxr_value"])
+
+            for i in range(len(table)):
+                row = table.slice(i, 1)
+                filename = row["filename"][0].as_py()
+                aia_arr = np.array(row["aia_stack"][0].as_py(), dtype=np.float32)
+                sxr_arr = np.array(row["sxr_value"][0].as_py(), dtype=np.float32)
+
+                fut = pool.submit(_write_arrays, filename, aia_arr, sxr_arr,
+                                  aia_split_dir, sxr_split_dir)
+                futures[fut] = submitted
+                submitted += 1
+
+                if submitted % print_every == 0:
+                    done = [f for f in futures if f.done()]
+                    for f in done:
+                        if f.result():
+                            saved += 1
+                        else:
+                            skipped += 1
+                        del futures[f]
+
+                    elapsed = time.time() - start
+                    rate = submitted / elapsed if elapsed > 0 else 0
+                    print(
+                        f"[{hf_split}] submitted={submitted} | saved={saved} skipped={skipped} | "
+                        f"{rate:.1f} rows/sec",
+                        flush=True,
+                    )
+
+        for fut in as_completed(futures):
+            if fut.result():
+                saved += 1
+            else:
+                skipped += 1
+
+    elapsed = time.time() - start
+    print(f"[{hf_split}] Done — {saved} saved, {skipped} skipped | {elapsed/60:.1f} min")
+
+
+def main():
+    parser = argparse.ArgumentParser(
+        description="Convert locally-downloaded HF parquet files to .npy arrays"
+    )
+    parser.add_argument("--config", type=str, default=None,
+                        help="Path to hf_download_config.yaml (provides aia_dir, sxr_dir, num_workers)")
+    parser.add_argument("--aia_dir", type=str, default=None,
+                        help="Output base dir for AIA .npy files (overrides config)")
+    parser.add_argument("--sxr_dir", type=str, default=None,
+                        help="Output base dir for SXR .npy files (overrides config)")
+    parser.add_argument("--parquet_dir", type=str, default=None,
+                        help="Directory containing parquet files for a single split")
+    parser.add_argument("--split", type=str, default=None,
+                        help="Split name for --parquet_dir (train, validation, test)")
+    parser.add_argument("--parquet_root", type=str, default=None,
+                        help="Root dir with split subdirs (train/, validation/, test/)")
+    parser.add_argument("--splits", type=str, default="train,validation,test",
+                        help="Comma-separated splits to process when using --parquet_root")
+    parser.add_argument("--num_workers", type=int, default=None,
+                        help="Parallel write threads (default: from config or 8)")
+    parser.add_argument("--print_every", type=int, default=500,
+                        help="Log progress every N rows")
+    args = parser.parse_args()
+
+    cfg = load_config(args.config) if args.config else {}
+
+    aia_dir = args.aia_dir or cfg.get("aia_dir")
+    sxr_dir = args.sxr_dir or cfg.get("sxr_dir")
+    num_workers = args.num_workers or cfg.get("num_workers", 8)
+
+    if not aia_dir or not sxr_dir:
+        parser.error("Provide --aia_dir and --sxr_dir, or --config with those keys set.")
+
+    if args.parquet_root:
+        splits = [s.strip() for s in args.splits.split(",")]
+        for split in splits:
+            split_dir = os.path.join(args.parquet_root, split)
+            if not os.path.isdir(split_dir):
+                print(f"[warn] Split dir not found, skipping: {split_dir}")
+                continue
+            convert_split(split_dir, split, aia_dir, sxr_dir, num_workers, args.print_every)
+    elif args.parquet_dir:
+        if not args.split:
+            parser.error("--split is required when using --parquet_dir")
+        convert_split(args.parquet_dir, args.split, aia_dir, sxr_dir, num_workers, args.print_every)
+    else:
+        parser.error("Provide either --parquet_dir + --split, or --parquet_root")
+
+    print("\nDone.")
+
+
+if __name__ == "__main__":
+    main()

pipeline_config.yaml

@@ -24,6 +24,24 @@ checkpoint: "/Users/griffingoodwin/Downloads/FOXES_Model_Checkpoint.ckpt"
 hf_download:
   config: "download/hf_download_config.yaml"
 
+# -----------------------------------------------------------------------------
+# Local parquet → .npy  (step: parquet_to_npy)
+# Use this if you've already downloaded HF parquet files and want to skip the
+# network step. Point parquet_root at a directory with split subdirs
+# (train/, validation/, test/) or use parquet_dir + split for a single split.
+# aia_dir / sxr_dir default to the values in hf_download_config.yaml if
+# --config is also provided; override here to use different paths.
+# -----------------------------------------------------------------------------
+parquet_to_npy:
+  config: "download/hf_download_config.yaml"   # provides aia_dir, sxr_dir, num_workers
+  parquet_root: ""   # root dir containing train/, validation/, test/ subdirs
+  # parquet_dir: ""  # alternative: single split dir (also set split: below)
+  # split: "train"
+  # aia_dir: "${base_dir}/AIA_hg_processed"    # override config if needed
+  # sxr_dir: "${base_dir}/SXR_hg_processed"
+  # splits: "train,validation,test"            # which subdirs to process
+  # num_workers: 8
+
 # -----------------------------------------------------------------------------
 # Shared date range (used by download_aia and download_sxr)
 # -----------------------------------------------------------------------------

requirements.txt

@@ -36,4 +36,6 @@ imageio-ffmpeg
 # Utilities
 tqdm
 wandb
-PyYAML
+PyYAML
+huggingface_hub
+datasets

run_pipeline.py

@@ -106,6 +106,7 @@ def write_merged_config(base_path: str, overrides: dict, out_name: str) -> Path:
 
 STEP_ORDER = [
     "hf_download",
+    "parquet_to_npy",
     "download_aia",
     "download_sxr",
     "combine_sxr",
@@ -125,6 +126,10 @@ STEP_INFO = {
         "description": "Download processed+split AIA/SXR data from HuggingFace Hub (replaces download→preprocess→split)",
         "script": ROOT / "download" / "hugging_face_data_download.py",
     },
+    "parquet_to_npy": {
+        "description": "Convert locally-downloaded HF parquet files to .npy (skips network download)",
+        "script": ROOT / "download" / "parquet_to_npy.py",
+    },
     "download_aia": {
         "description": "Download SDO/AIA EUV images from JSOC",
         "script": ROOT / "download" / "download_sdo.py",
@@ -200,6 +205,31 @@ def build_commands(step: str, cfg: dict, force: bool) -> list[list[str]] | None:
         config_path = hf.get("config", "download/hf_download_config.yaml")
         return [[sys.executable, str(STEP_INFO[step]["script"]), "--config", config_path]]
 
+    if step == "parquet_to_npy":
+        p2n = cfg.get("parquet_to_npy", {})
+        cmd = [sys.executable, str(STEP_INFO[step]["script"])]
+        if p2n.get("config"):
+            cmd += ["--config", p2n["config"]]
+        if p2n.get("parquet_root"):
+            cmd += ["--parquet_root", p2n["parquet_root"]]
+        elif p2n.get("parquet_dir"):
+            if not p2n.get("split"):
+                log.error("pipeline_config.yaml parquet_to_npy.split is required when parquet_dir is set")
+                return None
+            cmd += ["--parquet_dir", p2n["parquet_dir"], "--split", p2n["split"]]
+        else:
+            log.error("pipeline_config.yaml parquet_to_npy requires parquet_root or parquet_dir")
+            return None
+        if p2n.get("aia_dir"):
+            cmd += ["--aia_dir", p2n["aia_dir"]]
+        if p2n.get("sxr_dir"):
+            cmd += ["--sxr_dir", p2n["sxr_dir"]]
+        if p2n.get("splits"):
+            cmd += ["--splits", p2n["splits"]]
+        if p2n.get("num_workers"):
+            cmd += ["--num_workers", str(p2n["num_workers"])]
+        return [cmd]
+
     if step == "download_aia":
         if not require(["download_dir", "email"], "aia") or not require(["start_date"]):
             return None