Cross-Pipeline Caveats

1import numpy as np
2
3
4def detect_pipeline(state_dict_keys: list[str],
5                     dropout_per_dataset: dict[str, float] | None = None) -> str:
6    """Heuristic to identify which paper pipeline produced a checkpoint.
7
8    V7 (legacy AMNL):
9        - state-dict keys start with "backbone.", "rul_branch.", "health_branch."
10        - per-dataset dropout dict (FD001:0.3, FD002:0.2, ...)
11        - separate model class DualTaskEnhancedModel
12
13    GRACE refactor:
14        - state-dict keys start with "backbone.", "rul_head.", "health_head."
15        - uniform dropout (constant 0.3 across all datasets)
16        - unified DualTaskModel + ModelConfig dataclass
17    """
18    has_v7_branches = any("rul_branch." in k for k in state_dict_keys)
19    has_grace_heads = any("rul_head." in k for k in state_dict_keys)
20
21    has_per_dataset_dropout = (
22        dropout_per_dataset is not None
23        and len(set(dropout_per_dataset.values())) > 1
24    )
25
26    if has_v7_branches and has_per_dataset_dropout:
27        return "V7"
28    if has_grace_heads and not has_per_dataset_dropout:
29        return "GRACE"
30    return "UNKNOWN"
31
32
33def comparison_verdict(method_a_pipeline: str,
34                        method_b_pipeline: str) -> str:
35    """Decide if two methods can be compared head-to-head."""
36    if method_a_pipeline == method_b_pipeline:
37        return "DIRECT"
38    if "UNKNOWN" in (method_a_pipeline, method_b_pipeline):
39        return "INSUFFICIENT-INFO"
40    return "CROSS-PIPELINE (annotate with dagger)"
41
42
43# ---------- Worked example: classify three real checkpoints ----------
44v7_keys = [
45    "backbone.cnn.0.weight",   "backbone.cnn.2.weight",
46    "backbone.lstm.weight_ih_l0",
47    "rul_branch.0.weight",     "rul_branch.2.weight",     # ← V7 marker
48    "health_branch.0.weight",
49]
50grace_keys = [
51    "backbone.cnn.0.weight",   "backbone.cnn.2.weight",
52    "backbone.lstm.weight_ih_l0",
53    "rul_head.0.weight",       "rul_head.2.weight",       # ← GRACE marker
54    "health_head.0.weight",
55]
56
57v7_dropout    = {"FD001": 0.3, "FD002": 0.2, "FD003": 0.3, "FD004": 0.2}
58grace_dropout = {"FD001": 0.3, "FD002": 0.3, "FD003": 0.3, "FD004": 0.3}
59
60print(f"AMNL row pipeline    : {detect_pipeline(v7_keys, v7_dropout)}")
61print(f"GABA row pipeline    : {detect_pipeline(grace_keys, grace_dropout)}")
62print(f"GRACE row pipeline   : {detect_pipeline(grace_keys, grace_dropout)}")
63print()
64print(f"AMNL vs GABA  : {comparison_verdict('V7', 'GRACE')}")
65print(f"GABA vs GRACE : {comparison_verdict('GRACE', 'GRACE')}")
66print(f"AMNL vs unknown: {comparison_verdict('V7', 'UNKNOWN')}")

1import torch
2import torch.nn as nn
3
4
5def pipeline_signature(model: nn.Module) -> dict:
6    """Read state-dict keys + dropout config to fingerprint a model.
7
8    Returns:
9        Dict with 'model_class', 'has_v7_branches', 'has_grace_heads',
10        'dropout_value' if extractable.
11    """
12    keys = list(model.state_dict().keys())
13    sig = {
14        "model_class":     type(model).__name__,
15        "has_v7_branches": any("rul_branch."  in k for k in keys),
16        "has_grace_heads": any("rul_head."    in k for k in keys),
17    }
18    # Try to extract a representative dropout value
19    for module in model.modules():
20        if isinstance(module, nn.Dropout):
21            sig["dropout_value"] = module.p
22            break
23    else:
24        sig["dropout_value"] = None
25    return sig
26
27
28# ---------- Stub model classes for the smoke test ----------
29class DualTaskEnhancedModel(nn.Module):       # V7 legacy
30    def __init__(self, dropout: float = 0.3):
31        super().__init__()
32        self.backbone     = nn.Linear(14, 256)
33        self.rul_branch    = nn.Sequential(nn.Linear(256, 1), nn.Dropout(dropout))
34        self.health_branch = nn.Sequential(nn.Linear(256, 3), nn.Dropout(dropout))
35
36
37class DualTaskModel(nn.Module):               # GRACE refactor
38    def __init__(self, dropout: float = 0.3):
39        super().__init__()
40        self.backbone    = nn.Linear(14, 256)
41        self.rul_head     = nn.Sequential(nn.Linear(256, 1), nn.Dropout(dropout))
42        self.health_head  = nn.Sequential(nn.Linear(256, 3), nn.Dropout(dropout))
43
44
45# ---------- Smoke test ----------
46torch.manual_seed(0)
47v7_model    = DualTaskEnhancedModel(dropout=0.2)        # FD002 V7 dropout
48grace_model = DualTaskModel(dropout=0.3)                 # GRACE uniform
49
50print(f"{'model':<26s} | class                       | v7_branches | grace_heads | dropout")
51print(f"{'-' * 110}")
52for name, m in [("AMNL FD002 (V7 legacy)", v7_model),
53                ("GRACE FD002 (refactor)", grace_model)]:
54    sig = pipeline_signature(m)
55    print(f"{name:<26s} | {sig['model_class']:<27s} | "
56          f"{str(sig['has_v7_branches']):<11s} | {str(sig['has_grace_heads']):<11s} | "
57          f"{sig['dropout_value']}")