Enhance preprocessing scripts and checksum comparison functionality

code/preprocessing/00_preprocess.sh

@@ -3,11 +3,11 @@
 #
 # Usage:
 #   bash 00_preprocess.sh                          (runs all 6 steps)
-#   bash 00_preprocess.sh --start-step 3            (steps 03-06 only)
-#   bash 00_preprocess.sh --stop-step 4             (steps 01-04 only)
+#   bash 00_preprocess.sh --start_step 3            (steps 03-06 only)
+#   bash 00_preprocess.sh --stop_step 4             (steps 01-04 only)
 #   bash 00_preprocess.sh --steps 1,3,5             (only listed steps)
-#   bash 00_preprocess.sh --scan-dir /path/raw      (override scan path)
-#   bash 00_preprocess.sh --save-dir /path/output   (override save path)
+#   bash 00_preprocess.sh --scan_dir /path/raw      (override scan path)
+#   bash 00_preprocess.sh --save_dir /path/output   (override save path)
 # =============================================================================
 
 set -euo pipefail
@@ -20,10 +20,10 @@ STEPS_FILTER=""
 
 while [ $# -gt 0 ]; do
   case "$1" in
-    --scan-dir)   SCAN_DIR="$2"; shift 2 ;;
-    --save-dir)   SAVE_DIR="$2"; shift 2 ;;
-    --start-step) START_STEP="$2"; shift 2 ;;
-    --stop-step)  STOP_STEP="$2"; shift 2 ;;
+    --scan_dir)   SCAN_DIR="$2"; shift 2 ;;
+    --save_dir)   SAVE_DIR="$2"; shift 2 ;;
+    --start_step) START_STEP="$2"; shift 2 ;;
+    --stop_step)  STOP_STEP="$2"; shift 2 ;;
     --steps)      STEPS_FILTER="$2"; shift 2 ;;
     *)            shift ;;
   esac
@@ -42,8 +42,8 @@ should_run() {
 # Step 01
 if should_run 1; then
   STEP01_ARGS=()
-  [ -n "$SCAN_DIR" ] && STEP01_ARGS+=("--scan-dir" "$SCAN_DIR")
-  [ -n "$SAVE_DIR" ] && STEP01_ARGS+=("--save-dir" "$SAVE_DIR")
+  [ -n "$SCAN_DIR" ] && STEP01_ARGS+=("--scan_dir" "$SCAN_DIR")
+  [ -n "$SAVE_DIR" ] && STEP01_ARGS+=("--save_dir" "$SAVE_DIR")
   python /FL_system/code/preprocessing/01_scanDicom.py "${STEP01_ARGS[@]}"
 else
   echo "Skipping step 01"

code/preprocessing/01_scanDicom.py

@@ -24,10 +24,10 @@
     --profile: Enable profiling with yappi.
     --dir_idx (int): Index of the directory to process (for HPC array jobs).
     --dir_list (str): Path to the list of directories (for HPC array jobs).
-    --sample-pct (float): Percentage of files to sample per directory (0 = full scan).
-    --sample-seed (int): Random seed for sampling.
-    --checkpoint-dir (str): Directory for storing checkpoints.
-    --profile-dir (str): Directory for storing profiling output.
+    --sample_pct (float): Percentage of files to sample per directory (0 = full scan).
+    --sample_seed (int): Random seed for sampling.
+    --checkpoint_dir (str): Directory for storing checkpoints.
+    --profile_dir (str): Directory for storing profiling output.
     --resume: Resume from available checkpoints if present.
 
 Dependencies:
@@ -96,13 +96,13 @@ def build_config() -> ScanConfig:
                         help='Index of the folder to process from dirs_to_process.pkl (for HPC array jobs)')
     parser.add_argument('--dir_list', type=str, default='dirs_to_process.pkl',
                         help='Path to the directory list file (for HPC array jobs)')
-    parser.add_argument('--sample-pct', type=float, default=0.0,
+    parser.add_argument('--sample_pct', type=float, default=0.0,
                         help='Percent of .dcm files to sample per directory (0 = full scan)')
-    parser.add_argument('--sample-seed', type=int, default=None,
+    parser.add_argument('--sample_seed', type=int, default=None,
                         help='Optional random seed for sampling reproducibility')
-    parser.add_argument('--checkpoint-dir', type=str, default=None,
+    parser.add_argument('--checkpoint_dir', type=str, default=None,
                         help='Directory to store checkpoint files (default: <SAVE_DIR>/checkpoints/)')
-    parser.add_argument('--profile-dir', type=str, default=None,
+    parser.add_argument('--profile_dir', type=str, default=None,
                         help='Directory to store profiling output (default: <SAVE_DIR>/profiles/)')
     parser.add_argument('--resume', action='store_true',
                         help='Resume from available checkpoints if present')

code/preprocessing/03_saveNifti.py

@@ -179,15 +179,18 @@ def run_cmd(command, commands):
                 LOGGER.debug(f'Created directory for {SessionID}')
         except FileExistsError:
             LOGGER.warning(f'Directory for {SessionID} already exists')
+    LOGGER.info(f'Executing: dcm2niix -o {command[2]} -f {command[4]} {command[-1]}')
     try:
         if DEBUG == 0:
             result = subprocess.run(command, check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
         else:
             result = subprocess.run(command, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
             print(result.stdout.decode())
+        LOGGER.info(f'Completed: {command[4]} from {command[-1]}')
         with disk_space_lock:
             commands.remove(command)
     except subprocess.CalledProcessError as e:
+        LOGGER.error(f'Failed: {command[4]} from {command[-1]}')
         error_message = e.stderr.decode() if e.stderr else 'No error message available'
         LOGGER.error(f'Error converting {command[-1]}: {error_message}')
     #progress_queue.put((None, f'Converting'))

tools/data_checksum_analysis/compare_checksum.py

@@ -21,34 +21,52 @@
     scan2_data = json.load(f)
     print(f'Loaded secondary scan: {scans[scan2_index]} with {len(scan2_data["results"])} directories')
 
-# Compare files at the individual level across both scans
-# Files in primary that also exist in secondary with matching checksums -> marked for deletion from primary
-# Files in primary that are missing in secondary or have different checksums -> marked for transfer/replacement
+# Compare at the session level: any file mismatch flags the entire session for transfer.
+# Only sessions where every file matches go to ready_for_deletion.
+# Sessions only in secondary are flagged as missing from primary.
 report = {
     'ready_for_deletion': [],
     'need_transfer': [],
+    'missing_from_primary': [],
 }
 secondary_file_index = {}
+secondary_session_set = set()
 for dir_name, dir_data in scan2_data['results'].items():
+    secondary_session_set.add(dir_name)
     for f in dir_data['files']:
         key = os.path.join(dir_name, f['file_name'])
         secondary_file_index[key] = f['md5']
 
+primary_session_set = set()
 for dir_name, dir_data in scan1_data['results'].items():
+    primary_session_set.add(dir_name)
+    session_needs_transfer = False
+
     for f in dir_data['files']:
         key = os.path.join(dir_name, f['file_name'])
         secondary_md5 = secondary_file_index.get(key)
-        if secondary_md5 is not None and secondary_md5 == f['md5']:
+        if secondary_md5 is None or secondary_md5 != f['md5']:
+            session_needs_transfer = True
+            break
+
+    if session_needs_transfer:
+        report['need_transfer'].append({
+            'session': dir_name,
+            'file_count': len(dir_data['files']),
+        })
+    else:
+        for f in dir_data['files']:
             report['ready_for_deletion'].append({
-                'path': key,
+                'path': os.path.join(dir_name, f['file_name']),
                 'md5': f['md5'],
             })
-        else:
-            report['need_transfer'].append({
-                'path': key,
-                'primary_md5': f['md5'],
-                'secondary_md5': secondary_md5 if secondary_md5 else None,
-            })
+
+for dir_name in (secondary_session_set - primary_session_set):
+    dir_data = scan2_data['results'][dir_name]
+    report['missing_from_primary'].append({
+        'session': dir_name,
+        'file_count': len(dir_data['files']),
+    })
 
 stop_time = datetime.now(timezone.utc) # Record the stop time of the comparison in UTC timezone
 header = {
@@ -72,5 +90,6 @@
 print('-='*20)
 print('SUMMARY')
 print('-='*20)
-print(f'Need Transfer: {len(output['report']['need_transfer'])}')
-print(f'Deletion Ready: {len(output['report']['ready_for_deletion'])}')
+print(f'Need Transfer: {len(output["report"]["need_transfer"])}')
+print(f'Deletion Ready: {len(output["report"]["ready_for_deletion"])}')
+print(f'Missing from Primary: {len(output["report"]["missing_from_primary"])}')

tools/data_checksum_analysis/digest_comparison.py

@@ -0,0 +1,53 @@
+import json
+import os
+import sys
+from argparse import ArgumentParser
+
+parser = ArgumentParser(description="Digest a comparison report JSON into plain-text session ID lists.")
+parser.add_argument("report", help="Path to the comparison report JSON file (output from compare_checksum.py).")
+parser.add_argument("-o", "--outdir", help="Directory to write output files. Defaults to comparison_findings/.")
+args = parser.parse_args()
+
+if not os.path.exists(args.report):
+    print(f"Error: {args.report} not found.", file=sys.stderr)
+    sys.exit(1)
+
+with open(args.report, 'r') as f:
+    data = json.load(f)
+
+report = data["report"]
+outdir = args.outdir if args.outdir else "comparison_findings"
+
+# need_transfer: sessions flagged because at least one file differed
+transfer_sessions = sorted(item["session"] for item in report["need_transfer"])
+
+# ready_for_deletion: extract unique sessions from individual file entries
+deletion_sessions = sorted(item["path"].split("/", 1)[0] for item in report["ready_for_deletion"])
+deletion_sessions = sorted(set(deletion_sessions))
+
+# missing_from_primary: sessions in secondary with no primary counterpart
+missing_sessions = sorted(item["session"] for item in report["missing_from_primary"])
+
+files_written = []
+
+with open(os.path.join(outdir, "sessions_need_transfer.txt"), 'w') as f:
+    f.write("\n".join(transfer_sessions))
+    if transfer_sessions:
+        f.write("\n")
+    files_written.append(("sessions_need_transfer.txt", len(transfer_sessions)))
+
+with open(os.path.join(outdir, "sessions_ready_for_deletion.txt"), 'w') as f:
+    f.write("\n".join(deletion_sessions))
+    if deletion_sessions:
+        f.write("\n")
+    files_written.append(("sessions_ready_for_deletion.txt", len(deletion_sessions)))
+
+with open(os.path.join(outdir, "sessions_missing_from_primary.txt"), 'w') as f:
+    f.write("\n".join(missing_sessions))
+    if missing_sessions:
+        f.write("\n")
+    files_written.append(("sessions_missing_from_primary.txt", len(missing_sessions)))
+
+print(f"Digested {args.report}")
+for fname, count in files_written:
+    print(f"  {fname}: {count} sessions")

tools/data_checksum_analysis/merge_checksums.py

@@ -0,0 +1,91 @@
+import json
+import os
+from argparse import ArgumentParser
+from datetime import datetime, timezone
+
+parser = ArgumentParser(description="Merge two checksum scan result JSON files into a single comparison file.")
+parser.add_argument("scan1", help="Path to the first scan result JSON file (primary/source).")
+parser.add_argument("scan2", help="Path to the second scan result JSON file (secondary/destination).")
+parser.add_argument("-o", "--output", help="Output file path. Defaults to merged_comparison.json in the current directory.")
+args = parser.parse_args()
+
+start_time = datetime.now(timezone.utc)
+
+with open(args.scan1, 'r') as f:
+    scan1 = json.load(f)
+with open(args.scan2, 'r') as f:
+    scan2 = json.load(f)
+
+print(f"Loaded scan1: {args.scan1}  ({len(scan1['results'])} sessions, {sum(len(d['files']) for d in scan1['results'].values())} files)")
+print(f"Loaded scan2: {args.scan2}  ({len(scan2['results'])} sessions, {sum(len(d['files']) for d in scan2['results'].values())} files)")
+
+index1 = {}
+for session, data in scan1['results'].items():
+    for f in data['files']:
+        key = os.path.join(session, f['file_name'])
+        index1[key] = f['md5']
+
+index2 = {}
+for session, data in scan2['results'].items():
+    for f in data['files']:
+        key = os.path.join(session, f['file_name'])
+        index2[key] = f['md5']
+
+all_paths = sorted(set(index1.keys()) | set(index2.keys()))
+
+merged_results = {}
+stats = {"identical": 0, "modified": 0, "primary_only": 0, "secondary_only": 0}
+
+for path in all_paths:
+    md5_1 = index1.get(path)
+    md5_2 = index2.get(path)
+
+    session, file_name = os.path.split(path)
+    file_entry = {"file_name": file_name}
+
+    if md5_1 is not None and md5_2 is not None:
+        if md5_1 == md5_2:
+            file_entry["md5"] = md5_1
+            file_entry["source"] = "both"
+            stats["identical"] += 1
+        else:
+            file_entry["md5"] = md5_1
+            file_entry["md5_secondary"] = md5_2
+            file_entry["source"] = "both_modified"
+            stats["modified"] += 1
+    elif md5_1 is not None:
+        file_entry["md5"] = md5_1
+        file_entry["source"] = "primary"
+        stats["primary_only"] += 1
+    else:
+        file_entry["md5"] = md5_2
+        file_entry["source"] = "secondary"
+        stats["secondary_only"] += 1
+
+    if session not in merged_results:
+        merged_results[session] = {"files": []}
+    merged_results[session]["files"].append(file_entry)
+
+stop_time = datetime.now(timezone.utc)
+
+output = {
+    "header": {
+        "primary_scan": scan1["header"],
+        "secondary_scan": scan2["header"],
+        "merged_at": stop_time.isoformat(),
+        "summary": stats,
+    },
+    "results": merged_results,
+}
+
+output_path = args.output if args.output else "merged_comparison.json"
+with open(output_path, 'w') as f:
+    json.dump(output, f, indent=2)
+
+total_files = sum(len(d["files"]) for d in merged_results.values())
+print(f"\nMerged {total_files} files ({len(merged_results)} sessions) -> {output_path}")
+print(f"  Identical:      {stats['identical']}")
+print(f"  Modified:       {stats['modified']}")
+print(f"  Primary only:   {stats['primary_only']}")
+print(f"  Secondary only: {stats['secondary_only']}")
+print(f"  Elapsed: {(stop_time - start_time).total_seconds():.1f}s")

tools/data_checksum_analysis/move_sessions.py

@@ -0,0 +1,61 @@
+import os
+import sys
+import shutil
+from argparse import ArgumentParser
+
+parser = ArgumentParser(description="Move session directories listed in a plain-text file from one base path to another.")
+parser.add_argument("session_file", help="Path to the text file containing session IDs (one per line).")
+parser.add_argument("source", help="Source base directory (parent of session directories).")
+parser.add_argument("destination", help="Destination base directory to move sessions into.")
+parser.add_argument("--dry-run", action="store_true", help="Print what would be moved without actually moving.")
+args = parser.parse_args()
+
+if not os.path.isfile(args.session_file):
+    print(f"Error: {args.session_file} not found.", file=sys.stderr)
+    sys.exit(1)
+
+if not os.path.isdir(args.source):
+    print(f"Error: source directory {args.source} not found.", file=sys.stderr)
+    sys.exit(1)
+
+with open(args.session_file, 'r') as f:
+    sessions = [line.strip() for line in f if line.strip()]
+
+if not sessions:
+    print(f"No session IDs found in {args.session_file}.")
+    sys.exit(0)
+
+total = len(sessions)
+moved = 0
+skipped = 0
+errors = 0
+
+mode = "Would move" if args.dry_run else "Moving"
+print(f"{mode} {total} sessions from {args.source} -> {args.destination}")
+print(f"{'='*60}")
+
+for i, session_id in enumerate(sessions, 1):
+    src_path = os.path.join(args.source, session_id)
+    dst_path = os.path.join(args.destination, session_id)
+
+    if not os.path.exists(src_path):
+        print(f"[{i}/{total}] SKIP (not found): {session_id}")
+        skipped += 1
+        continue
+
+    if args.dry_run:
+        print(f"[{i}/{total}] DRY-RUN: {session_id}")
+        moved += 1
+        continue
+
+    try:
+        os.makedirs(args.destination, exist_ok=True)
+        shutil.move(src_path, dst_path)
+        print(f"[{i}/{total}] OK: {session_id}")
+        moved += 1
+    except Exception as e:
+        print(f"[{i}/{total}] ERROR: {session_id} -> {e}", file=sys.stderr)
+        errors += 1
+
+print(f"{'='*60}")
+print(f"Done: {moved} moved, {skipped} skipped, {errors} errors out of {total} sessions.")