CardiovascularImagingVHIR
/

PrePostEVAR_CTA_Annotation

@@ -1,350 +0,0 @@
-from pathlib import Path
-import json
-import shutil
-import torch
-import re
-SOURCE_ROOT = Path(r"\\NAS3_Z\all\Juan\Documents\MATLAB\AutoAGM\AutoAGM_files\pythonfiles\folders\nnunet_results")
-DEST_ROOT = Path(r"\\NAS3_Z\all\Juan\EVARAI\HuggingFace_models")
-DATASETS = {
-    "Dataset370_PreInterventionV3": "Dataset100_SegmentationPreEVAR",
-    "Dataset380_LandmarksPreInterventionV2": "Dataset101_LandmarksPreEVAR",
-    "Dataset371_PostInterventionV3": "Dataset200_SegmentationPostEVAR",
-    "Dataset381_LandmarksPostInterventionV2": "Dataset201_LandmarksPostEVAR",
-}
-OVERWRITE = True
-def get_dataset_id(name):
-    match = re.match(r"Dataset(\d{3})_", name)
-    if not match:
-        raise ValueError(f"Nombre de dataset no válido: {name}")
-    return int(match.group(1))
-def normalize_key(key):
-    return str(key).lower().replace("-", "_").replace(" ", "_")
-def is_dataset_id_key(key):
-    key = normalize_key(key)
-    return key in {"dataset_id", "datasetid", "dataset_name_or_id"} or key.endswith("_dataset_id")
-def is_dataset_name_key(key):
-    key = normalize_key(key)
-    return key == "dataset_name" or key.endswith("_dataset_name")
-def load_checkpoint(path):
-    try:
-        return torch.load(path, map_location="cpu", weights_only=False)
-    except TypeError:
-        return torch.load(path, map_location="cpu")
-def find_dataset_references(obj, old_name, old_id, path="root", parent_key=None):
-    """Busca referencias al nombre/ID antiguo sin modificar nada."""
-    found = []
-    if isinstance(obj, dict):
-        for key, value in obj.items():
-            current_path = f"{path}.{key}"
-            if isinstance(key, str) and (old_name in key or f"Dataset{old_id:03d}" in key):
-                found.append((current_path + " [KEY]", key))
-            # No tiene sentido recorrer millones de parámetros
-            if key == "network_weights":
-                continue
-            found.extend(find_dataset_references(value, old_name, old_id, current_path, key))
-    elif isinstance(obj, (list, tuple)):
-        for i, value in enumerate(obj):
-            found.extend(find_dataset_references(value, old_name, old_id, f"{path}[{i}]", parent_key))
-    elif isinstance(obj, str):
-        if old_name in obj or f"Dataset{old_id:03d}" in obj:
-            found.append((path, obj))
-        elif parent_key is not None and is_dataset_id_key(parent_key) and obj == str(old_id):
-            found.append((path, obj))
-    elif isinstance(obj, int) and not isinstance(obj, bool):
-        if parent_key is not None and is_dataset_id_key(parent_key) and obj == old_id:
-            found.append((path, obj))
-    return found
-def rewrite_metadata(obj, old_name, new_name, old_id, new_id, parent_key=None):
-    """
-    Cambia únicamente referencias identificables del dataset.
-    No hace un reemplazo global old_id -> new_id.
-    """
-    if isinstance(obj, dict):
-        result = {}
-        for key, value in obj.items():
-            new_key = key
-            if isinstance(key, str):
-                new_key = key.replace(old_name, new_name)
-            if new_key == "network_weights":
-                result[new_key] = value
-                continue
-            result[new_key] = rewrite_metadata(value, old_name, new_name, old_id, new_id, new_key)
-        return result
-    if isinstance(obj, list):
-        return [rewrite_metadata(v, old_name, new_name, old_id, new_id, parent_key) for v in obj]
-    if isinstance(obj, tuple):
-        return tuple(rewrite_metadata(v, old_name, new_name, old_id, new_id, parent_key) for v in obj)
-    if isinstance(obj, str):
-        value = obj.replace(old_name, new_name)
-        if parent_key is not None and is_dataset_name_key(parent_key):
-            if value.startswith(f"Dataset{old_id:03d}_") or value == f"Dataset{old_id:03d}":
-                value = new_name
-        if parent_key is not None and is_dataset_id_key(parent_key) and value == str(old_id):
-            value = str(new_id)
-        return value
-    if isinstance(obj, int) and not isinstance(obj, bool):
-        if parent_key is not None and is_dataset_id_key(parent_key) and obj == old_id:
-            return new_id
-    return obj
-def verify_weights_identical(original, modified):
-    """Comprueba que network_weights no ha cambiado."""
-    if "network_weights" not in original or "network_weights" not in modified:
-        raise RuntimeError("El checkpoint no contiene 'network_weights'.")
-    weights_a = original["network_weights"]
-    weights_b = modified["network_weights"]
-    if weights_a.keys() != weights_b.keys():
-        raise RuntimeError("Las claves de network_weights han cambiado.")
-    for key in weights_a:
-        a, b = weights_a[key], weights_b[key]
-        if torch.is_tensor(a) and torch.is_tensor(b):
-            if not torch.equal(a, b):
-                raise RuntimeError(f"Los pesos han cambiado en: {key}")
-        elif a != b:
-            raise RuntimeError(f"Ha cambiado un elemento de network_weights: {key}")
-def audit_json(path, old_name, old_id):
-    with open(path, "r", encoding="utf-8") as f:
-        data = json.load(f)
-    return find_dataset_references(data, old_name, old_id, path.name)
-def audit_checkpoint(path, old_name, old_id):
-    checkpoint = load_checkpoint(path)
-    return find_dataset_references(checkpoint, old_name, old_id, path.name)
-def save_modified_json(src, dst, old_name, new_name, old_id, new_id):
-    with open(src, "r", encoding="utf-8") as f:
-        data = json.load(f)
-    data = rewrite_metadata(data, old_name, new_name, old_id, new_id)
-    dst.parent.mkdir(parents=True, exist_ok=True)
-    with open(dst, "w", encoding="utf-8") as f:
-        json.dump(data, f, indent=4, ensure_ascii=False)
-def save_modified_checkpoint(src, dst, old_name, new_name, old_id, new_id):
-    checkpoint = load_checkpoint(src)
-    # Guardamos una referencia a los pesos originales.
-    original_weights = checkpoint["network_weights"]
-    # Modificamos todos los metadatos excepto network_weights.
-    modified = rewrite_metadata(checkpoint, old_name, new_name, old_id, new_id)
-    # Comprobación antes de guardar.
-    verify_weights_identical({"network_weights": original_weights}, modified)
-    dst.parent.mkdir(parents=True, exist_ok=True)
-    torch.save(modified, dst)
-    # Volvemos a cargar el archivo generado y comprobamos de nuevo.
-    reloaded = load_checkpoint(dst)
-    verify_weights_identical({"network_weights": original_weights}, reloaded)
-def find_trainer_dirs(dataset_dir):
-    trainers = []
-    for plans_file in dataset_dir.rglob("plans.json"):
-        trainer_dir = plans_file.parent
-        if (trainer_dir / "dataset.json").exists() and any(trainer_dir.glob("fold_*")):
-            trainers.append(trainer_dir)
-    return sorted(set(trainers))
-def print_references(refs, indent="      "):
-    if not refs:
-        print(f"{indent}No se encontraron referencias.")
-        return
-    for path, value in refs:
-        value = str(value)
-        if len(value) > 150:
-            value = value[:147] + "..."
-        print(f"{indent}{path}: {value}")
-def export_dataset(old_name, new_name):
-    old_id = get_dataset_id(old_name)
-    new_id = get_dataset_id(new_name)
-    src_dataset = SOURCE_ROOT / old_name
-    dst_dataset = DEST_ROOT / new_name
-    print("\n" + "=" * 80)
-    print(f"{old_name}  ->  {new_name}")
-    print(f"ID: {old_id} -> {new_id}")
-    if not src_dataset.exists():
-        raise FileNotFoundError(f"No existe: {src_dataset}")
-    if dst_dataset.exists() and not OVERWRITE:
-        raise FileExistsError(
-            f"Ya existe {dst_dataset}. Bórralo o cambia OVERWRITE = True."
-        )
-    if dst_dataset.exists() and OVERWRITE:
-        shutil.rmtree(dst_dataset)
-    trainer_dirs = find_trainer_dirs(src_dataset)
-    if not trainer_dirs:
-        raise RuntimeError(f"No se encontraron trainers dentro de {src_dataset}")
-    for trainer_dir in trainer_dirs:
-        relative = trainer_dir.relative_to(src_dataset)
-        dst_trainer = dst_dataset / relative
-        print(f"\n  Trainer: {relative}")
-        # ------------------------------------------------------------------
-        # AUDITORÍA ORIGINAL
-        # ------------------------------------------------------------------
-        for filename in ("dataset.json", "plans.json"):
-            src = trainer_dir / filename
-            print(f"\n    Auditoría {filename}:")
-            refs = audit_json(src, old_name, old_id)
-            print_references(refs)
-            save_modified_json(
-                src, dst_trainer / filename,
-                old_name, new_name, old_id, new_id
-            )
-        # ------------------------------------------------------------------
-        # FOLDS
-        # ------------------------------------------------------------------
-        fold_dirs = sorted(
-            [p for p in trainer_dir.glob("fold_*") if p.is_dir()],
-            key=lambda p: p.name
-        )
-        for fold_dir in fold_dirs:
-            checkpoint = fold_dir / "checkpoint_best.pth"
-            if not checkpoint.exists():
-                print(f"\n    [WARNING] {fold_dir.name}: no existe checkpoint_best.pth")
-                continue
-            print(f"\n    Auditoría {fold_dir.name}/checkpoint_best.pth:")
-            refs = audit_checkpoint(checkpoint, old_name, old_id)
-            print_references(refs)
-            dst_checkpoint = dst_trainer / fold_dir.name / "checkpoint_best.pth"
-            save_modified_checkpoint(
-                checkpoint, dst_checkpoint,
-                old_name, new_name, old_id, new_id
-            )
-            print("      ✓ checkpoint copiado")
-            print("      ✓ metadatos actualizados")
-            print("      ✓ network_weights idénticos")
-        # ------------------------------------------------------------------
-        # AUDITORÍA FINAL
-        # ------------------------------------------------------------------
-        print("\n    Comprobación final...")
-        residual_refs = []
-        for filename in ("dataset.json", "plans.json"):
-            residual_refs.extend(
-                audit_json(dst_trainer / filename, old_name, old_id)
-            )
-        for checkpoint in dst_trainer.glob("fold_*/checkpoint_best.pth"):
-            residual_refs.extend(
-                audit_checkpoint(checkpoint, old_name, old_id)
-            )
-        if residual_refs:
-            print("\n    [ERROR] Quedan referencias al dataset antiguo:")
-            print_references(residual_refs)
-            raise RuntimeError(
-                f"La exportación de {old_name} contiene referencias antiguas."
-            )
-        print("    ✓ No quedan referencias al nombre/ID antiguo.")
-def main():
-    if not SOURCE_ROOT.exists():
-        raise FileNotFoundError(f"No existe SOURCE_ROOT: {SOURCE_ROOT}")
-    DEST_ROOT.mkdir(parents=True, exist_ok=True)
-    print("SOURCE:", SOURCE_ROOT)
-    print("DEST:  ", DEST_ROOT)
-    for old_name, new_name in DATASETS.items():
-        export_dataset(old_name, new_name)
-    print("\n" + "=" * 80)
-    print("EXPORTACIÓN COMPLETADA")
-    print("=" * 80)
-    for new_name in DATASETS.values():
-        print(f"  {DEST_ROOT / new_name}")
-if __name__ == "__main__":
-    main()