Skip to content

Core: Data

All dataset, preprocessing, representation, and discovery machinery.

Current layout

  • datasets/ - raw source adapters and dataset/source builders.
  • discovery/ - signal profiles, canonical entities, and provisional hypotheses for cross-vehicle alignment.
  • preprocessing/ - explicit representations, views, segments, materialization, PyG packing, temporal streams, scaler config, vocab config, and graph transforms.
  • datamodule/ - training-time loaders and batching policy.
  • state.py - process-local dataset state for reuse within one Python process.

graphids.core.data

data

Data-layer public API.

The runtime datamodules are imported lazily so preprocessing and discovery can be used without importing optional training dependencies.

CANBusDataset 

CANBusDataset(root: str | Path, raw_dir: str | Path, *, val_fraction: float, split: str = 'train', source_dirs: list[str] | None = None, seed: int = 42, shared_vocab: dict | None = None, shared_vocab_digest: str | None = None, vocab_scope: Literal['train', 'all'] = 'train', scaler_cfg: ScalerCfg = _DEFAULT_SCALER_CFG, representation_cfg: GraphRepresentationCfg = _DEFAULT_REPRESENTATION_CFG, transform=None, pre_transform=None)

Bases: BaseGraphDataset

One graph is one sliding window of CAN messages.

Source code in graphids/core/data/datasets/_base.py 
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
def __init__(
    self,
    root: str | Path,
    raw_dir: str | Path,
    *,
    val_fraction: float,
    split: str = "train",
    source_dirs: list[str] | None = None,
    seed: int = 42,
    shared_vocab: dict | None = None,
    shared_vocab_digest: str | None = None,
    vocab_scope: Literal["train", "all"] = "train",
    scaler_cfg: ScalerCfg = _DEFAULT_SCALER_CFG,
    representation_cfg: GraphRepresentationCfg = _DEFAULT_REPRESENTATION_CFG,
    transform=None,
    pre_transform=None,
):
    self.raw_data_dir = Path(raw_dir)
    self.split = split
    self.val_fraction = val_fraction
    self.source_dirs = source_dirs
    self.seed = seed
    self._shared_vocab = shared_vocab
    self._shared_vocab_digest = shared_vocab_digest
    self.vocab_scope = vocab_scope
    self.scaler_cfg = scaler_cfg
    self.scaler_strategy = scaler_kind(scaler_cfg)
    self.representation_cfg = representation_cfg
    self.representation_kind = representation_kind(representation_cfg)
    self._split_plan = None
    self.window_size, self.stride = representation_window_defaults(representation_cfg)
    super().__init__(str(root), transform, pre_transform)
    self.load(self.processed_paths[0])
    self.num_ids = int(getattr(self._data, "num_ids", len(shared_vocab or {}) + 1))
    if self.split in ("train", "val"):
        self._split_plan = build_blocked_split_plan(
            self._data,
            self.slices,
            self.representation_cfg,
            val_fraction=self.val_fraction,
            seed=self.seed,
        )
        idx = self._split_plan.val_idx if self.split == "val" else self._split_plan.train_idx
        self._indices = idx.tolist()

CANBusSource dataclass 

CANBusSource(name: str, lake_root: str | None = None, val_fraction: float = 0.2, seed: int = 42, scaler_cfg: ScalerCfg = ZBenignScalerCfg(), representation_cfg: GraphRepresentationCfg = SnapshotRepresentationCfg(), vocab_scope: Literal['train', 'all'] = 'train')

Bases: BaseGraphSource

Catalog to train/val/test CANBusDataset builder.

DatasetState dataclass 

DatasetState(train: Any, val: Any, test: dict[str, Any])

Ready-to-serve train/val/test splits.

clear_cache 

clear_cache() -> None

Drop all cached states. Intended for test teardown.

Source code in graphids/core/data/state.py 
37
38
39
def clear_cache() -> None:
    """Drop all cached states. Intended for test teardown."""
    _REGISTRY.clear()

get_or_build 

get_or_build(dataset: _CacheableDataset) -> DatasetState

Return cached DatasetState for dataset.

Source code in graphids/core/data/state.py 
27
28
29
30
31
32
33
34
def get_or_build(dataset: _CacheableDataset) -> DatasetState:
    """Return cached ``DatasetState`` for ``dataset``."""
    key = dataset.cache_key
    state = _REGISTRY.get(key)
    if state is None:
        state = dataset.build()
        _REGISTRY[key] = state
    return state

datamodule

DataModule primitives for graph and fusion datasets.

GraphDataModule 

GraphDataModule(dataset, batch_size: int = 32, num_workers: int | None = None, prefetch_factor: int = 2, dynamic_batching: bool = True, label_filter: str | None = None, difficulty: Callable[..., Tensor] | None = None, scope_label: int = 0, min_steps_per_epoch: int = 1, require_cache: bool = False)

Bases: LightningDataModule

Source code in graphids/core/data/datamodule/graph.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
def __init__(
    self,
    dataset,
    batch_size: int = 32,
    num_workers: int | None = None,
    prefetch_factor: int = 2,
    dynamic_batching: bool = True,
    label_filter: str | None = None,
    difficulty: Callable[..., torch.Tensor] | None = None,
    scope_label: int = 0,
    min_steps_per_epoch: int = 1,
    require_cache: bool = False,
):
    super().__init__()
    self.source = dataset
    self.batch_size = batch_size
    self.num_workers = num_workers
    self.prefetch_factor = prefetch_factor
    self.dynamic_batching = dynamic_batching
    self.label_filter = label_filter
    self.difficulty = difficulty
    self.scope_label = scope_label
    self.min_steps_per_epoch = min_steps_per_epoch
    self.require_cache = require_cache
    self._train: InMemoryDataset | None = None
    self._val: InMemoryDataset | None = None
    self._tests: dict[str, InMemoryDataset] = {}
    self._train_graphs: list | None = None
    self._train_plans: list[list[int]] | None = None
    self._budget = None
train_eval_dataloader 
train_eval_dataloader()

Eval-style train loader for calibration and centroid stats.

Source code in graphids/core/data/datamodule/graph.py 
228
229
230
def train_eval_dataloader(self):
    """Eval-style train loader for calibration and centroid stats."""
    return self._fixed_loader(self._train_view(), shuffle=False)

fusion

Fusion data module for pre-extracted TensorDict caches.

graph

Lightning data module for graph datasets.

GraphDataModule 
GraphDataModule(dataset, batch_size: int = 32, num_workers: int | None = None, prefetch_factor: int = 2, dynamic_batching: bool = True, label_filter: str | None = None, difficulty: Callable[..., Tensor] | None = None, scope_label: int = 0, min_steps_per_epoch: int = 1, require_cache: bool = False)

Bases: LightningDataModule

Source code in graphids/core/data/datamodule/graph.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
def __init__(
    self,
    dataset,
    batch_size: int = 32,
    num_workers: int | None = None,
    prefetch_factor: int = 2,
    dynamic_batching: bool = True,
    label_filter: str | None = None,
    difficulty: Callable[..., torch.Tensor] | None = None,
    scope_label: int = 0,
    min_steps_per_epoch: int = 1,
    require_cache: bool = False,
):
    super().__init__()
    self.source = dataset
    self.batch_size = batch_size
    self.num_workers = num_workers
    self.prefetch_factor = prefetch_factor
    self.dynamic_batching = dynamic_batching
    self.label_filter = label_filter
    self.difficulty = difficulty
    self.scope_label = scope_label
    self.min_steps_per_epoch = min_steps_per_epoch
    self.require_cache = require_cache
    self._train: InMemoryDataset | None = None
    self._val: InMemoryDataset | None = None
    self._tests: dict[str, InMemoryDataset] = {}
    self._train_graphs: list | None = None
    self._train_plans: list[list[int]] | None = None
    self._budget = None
train_eval_dataloader 
train_eval_dataloader()

Eval-style train loader for calibration and centroid stats.

Source code in graphids/core/data/datamodule/graph.py 
228
229
230
def train_eval_dataloader(self):
    """Eval-style train loader for calibration and centroid stats."""
    return self._fixed_loader(self._train_view(), shuffle=False)

sampler

Offline next-fit decreasing packer for variable-size graphs.

pack_offline 
pack_offline(sizes: Tensor, max_num: int, *, edge_sizes: Tensor | None = None, max_edges: int | None = None) -> list[list[int]]

Pack graph indices under node and edge budgets.

The sorted next-fit strategy is intentionally linear after sorting. Exact first-fit gives slightly tighter bins, but it is quadratic on large cached graph datasets and can spend minutes on CPU before the first GPU step.

Source code in graphids/core/data/datamodule/sampler.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
def pack_offline(
    sizes: torch.Tensor,
    max_num: int,
    *,
    edge_sizes: torch.Tensor | None = None,
    max_edges: int | None = None,
) -> list[list[int]]:
    """Pack graph indices under node and edge budgets.

    The sorted next-fit strategy is intentionally linear after sorting. Exact
    first-fit gives slightly tighter bins, but it is quadratic on large cached
    graph datasets and can spend minutes on CPU before the first GPU step.
    """
    if max_num <= 0:
        raise ValueError(f"max_num must be positive, got {max_num}")
    if edge_sizes is not None:
        if len(edge_sizes) != len(sizes):
            raise ValueError(
                f"edge_sizes length ({len(edge_sizes)}) != sizes length ({len(sizes)})"
            )
        if max_edges is None or max_edges <= 0:
            raise ValueError("max_edges must be a positive int when edge_sizes is given")

    sizes = sizes.to(torch.long)
    es = edge_sizes.to(torch.long) if edge_sizes is not None else None
    order = torch.argsort(sizes, descending=True).tolist()

    bins: list[_Bin] = []
    current = _Bin()
    skipped = 0
    for i in order:
        n_i = int(sizes[i])
        e_i = int(es[i]) if es is not None else 0
        if n_i > max_num or (max_edges is not None and e_i > max_edges):
            skipped += 1
            continue
        fits_current = current.n_sum + n_i <= max_num and (
            max_edges is None or current.e_sum + e_i <= max_edges
        )
        if current.indices and not fits_current:
            bins.append(current)
            current = _Bin()
        current.indices.append(i)
        current.n_sum += n_i
        current.e_sum += e_i

    if current.indices:
        bins.append(current)

    if skipped:
        log.warning(
            "sampler_skipped_oversize",
            n_skipped=skipped,
            n_total=len(sizes),
            max_nodes=max_num,
            max_edges=max_edges,
        )
    return [b.indices for b in bins]

temporal

Lightning data module for temporal PyG event streams.

TemporalDataModule 
TemporalDataModule(dataset, batch_size: int = 256)

Bases: LightningDataModule

Serve temporal event streams with PyG's TemporalDataLoader.

Source code in graphids/core/data/datamodule/temporal.py 
13
14
15
16
17
18
19
def __init__(self, dataset, batch_size: int = 256):
    super().__init__()
    self.source = dataset
    self.batch_size = batch_size
    self._train = None
    self._val = None
    self._tests: dict[str, object] = {}

datasets

CANBusDataset 

CANBusDataset(root: str | Path, raw_dir: str | Path, *, val_fraction: float, split: str = 'train', source_dirs: list[str] | None = None, seed: int = 42, shared_vocab: dict | None = None, shared_vocab_digest: str | None = None, vocab_scope: Literal['train', 'all'] = 'train', scaler_cfg: ScalerCfg = _DEFAULT_SCALER_CFG, representation_cfg: GraphRepresentationCfg = _DEFAULT_REPRESENTATION_CFG, transform=None, pre_transform=None)

Bases: BaseGraphDataset

One graph is one sliding window of CAN messages.

Source code in graphids/core/data/datasets/_base.py 
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
def __init__(
    self,
    root: str | Path,
    raw_dir: str | Path,
    *,
    val_fraction: float,
    split: str = "train",
    source_dirs: list[str] | None = None,
    seed: int = 42,
    shared_vocab: dict | None = None,
    shared_vocab_digest: str | None = None,
    vocab_scope: Literal["train", "all"] = "train",
    scaler_cfg: ScalerCfg = _DEFAULT_SCALER_CFG,
    representation_cfg: GraphRepresentationCfg = _DEFAULT_REPRESENTATION_CFG,
    transform=None,
    pre_transform=None,
):
    self.raw_data_dir = Path(raw_dir)
    self.split = split
    self.val_fraction = val_fraction
    self.source_dirs = source_dirs
    self.seed = seed
    self._shared_vocab = shared_vocab
    self._shared_vocab_digest = shared_vocab_digest
    self.vocab_scope = vocab_scope
    self.scaler_cfg = scaler_cfg
    self.scaler_strategy = scaler_kind(scaler_cfg)
    self.representation_cfg = representation_cfg
    self.representation_kind = representation_kind(representation_cfg)
    self._split_plan = None
    self.window_size, self.stride = representation_window_defaults(representation_cfg)
    super().__init__(str(root), transform, pre_transform)
    self.load(self.processed_paths[0])
    self.num_ids = int(getattr(self._data, "num_ids", len(shared_vocab or {}) + 1))
    if self.split in ("train", "val"):
        self._split_plan = build_blocked_split_plan(
            self._data,
            self.slices,
            self.representation_cfg,
            val_fraction=self.val_fraction,
            seed=self.seed,
        )
        idx = self._split_plan.val_idx if self.split == "val" else self._split_plan.train_idx
        self._indices = idx.tolist()

CANBusSource dataclass 

CANBusSource(name: str, lake_root: str | None = None, val_fraction: float = 0.2, seed: int = 42, scaler_cfg: ScalerCfg = ZBenignScalerCfg(), representation_cfg: GraphRepresentationCfg = SnapshotRepresentationCfg(), vocab_scope: Literal['train', 'all'] = 'train')

Bases: BaseGraphSource

Catalog to train/val/test CANBusDataset builder.

can_bus

CAN bus dataset adapter and schema.

CANBusDataset 
CANBusDataset(root: str | Path, raw_dir: str | Path, *, val_fraction: float, split: str = 'train', source_dirs: list[str] | None = None, seed: int = 42, shared_vocab: dict | None = None, shared_vocab_digest: str | None = None, vocab_scope: Literal['train', 'all'] = 'train', scaler_cfg: ScalerCfg = _DEFAULT_SCALER_CFG, representation_cfg: GraphRepresentationCfg = _DEFAULT_REPRESENTATION_CFG, transform=None, pre_transform=None)

Bases: BaseGraphDataset

One graph is one sliding window of CAN messages.

Source code in graphids/core/data/datasets/_base.py 
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
def __init__(
    self,
    root: str | Path,
    raw_dir: str | Path,
    *,
    val_fraction: float,
    split: str = "train",
    source_dirs: list[str] | None = None,
    seed: int = 42,
    shared_vocab: dict | None = None,
    shared_vocab_digest: str | None = None,
    vocab_scope: Literal["train", "all"] = "train",
    scaler_cfg: ScalerCfg = _DEFAULT_SCALER_CFG,
    representation_cfg: GraphRepresentationCfg = _DEFAULT_REPRESENTATION_CFG,
    transform=None,
    pre_transform=None,
):
    self.raw_data_dir = Path(raw_dir)
    self.split = split
    self.val_fraction = val_fraction
    self.source_dirs = source_dirs
    self.seed = seed
    self._shared_vocab = shared_vocab
    self._shared_vocab_digest = shared_vocab_digest
    self.vocab_scope = vocab_scope
    self.scaler_cfg = scaler_cfg
    self.scaler_strategy = scaler_kind(scaler_cfg)
    self.representation_cfg = representation_cfg
    self.representation_kind = representation_kind(representation_cfg)
    self._split_plan = None
    self.window_size, self.stride = representation_window_defaults(representation_cfg)
    super().__init__(str(root), transform, pre_transform)
    self.load(self.processed_paths[0])
    self.num_ids = int(getattr(self._data, "num_ids", len(shared_vocab or {}) + 1))
    if self.split in ("train", "val"):
        self._split_plan = build_blocked_split_plan(
            self._data,
            self.slices,
            self.representation_cfg,
            val_fraction=self.val_fraction,
            seed=self.seed,
        )
        idx = self._split_plan.val_idx if self.split == "val" else self._split_plan.train_idx
        self._indices = idx.tolist()
CANBusSource dataclass 
CANBusSource(name: str, lake_root: str | None = None, val_fraction: float = 0.2, seed: int = 42, scaler_cfg: ScalerCfg = ZBenignScalerCfg(), representation_cfg: GraphRepresentationCfg = SnapshotRepresentationCfg(), vocab_scope: Literal['train', 'all'] = 'train')

Bases: BaseGraphSource

Catalog to train/val/test CANBusDataset builder.

infer_attack_type 
infer_attack_type(csv: Path) -> int

Infer the attack code from filename/path substrings.

Source code in graphids/core/data/datasets/can_bus.py 
148
149
150
151
152
153
154
def infer_attack_type(csv: Path) -> int:
    """Infer the attack code from filename/path substrings."""
    s = csv.stem.lower() + " " + csv.parent.name.lower()
    for kw, code in ATTACK_TYPE_CODES.items():
        if kw in s:
            return code
    return 0
load_can_rows 
load_can_rows(raw_dir: Path, source_dirs: list[str]) -> pl.DataFrame

Load, normalize, and parse raw CAN CSVs from source dirs.

Source code in graphids/core/data/datasets/can_bus.py 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
def load_can_rows(raw_dir: Path, source_dirs: list[str]) -> pl.DataFrame:
    """Load, normalize, and parse raw CAN CSVs from source dirs."""
    if not source_dirs:
        raise ValueError("source_dirs is empty; cannot load CAN rows")
    frames: list[pl.LazyFrame] = []
    for sub in source_dirs:
        sub_path = raw_dir / sub
        if not sub_path.is_dir():
            raise FileNotFoundError(f"declared source_dir {sub!r} missing under {raw_dir}")
        for csv_path in sorted(sub_path.rglob("*.csv")):
            at = infer_attack_type(csv_path)
            frames.append(
                pl.scan_csv(csv_path).with_columns(
                    pl.lit(at).alias("attack_type"),
                    pl.lit(sub).alias("vehicle_id"),
                    pl.lit(sub).alias("source_dir"),
                    pl.lit(str(csv_path.relative_to(raw_dir))).alias("source_file"),
                )
            )
    if not frames:
        raise ValueError(f"no CSVs under any of {source_dirs!r} in {raw_dir}")

    combined = pl.concat(frames).sort("timestamp")
    cols = combined.collect_schema().names()
    renames = {
        old: new
        for old, new in (("arbitration_id", "arb_id"), ("data_field", "payload"))
        if old in cols
    }
    if renames:
        combined = combined.rename(renames)
    return parse_payload(combined).collect()
parse_payload 
parse_payload(lf: LazyFrame) -> pl.LazyFrame

Hex payload to byte_0..7 plus Shannon entropy.

Source code in graphids/core/data/datasets/can_bus.py 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
def parse_payload(lf: pl.LazyFrame) -> pl.LazyFrame:
    """Hex ``payload`` to ``byte_0..7`` plus Shannon entropy."""
    if "byte_0" in lf.collect_schema().names():
        return lf
    byte_exprs = [
        pl.col("payload").str.slice(i * 2, 2).str.to_integer(base=16, strict=False)
        .fill_null(0).cast(pl.Float32).alias(f"byte_{i}")
        for i in range(N_BYTES)
    ]
    lf = lf.with_columns(byte_exprs)
    bcols = [pl.col(c) for c in BYTE_COLS]
    row_sum = pl.sum_horizontal(bcols).clip(1e-12, None)
    entropy = pl.sum_horizontal(
        [pl.when(c > 0).then(-(c / row_sum) * (c / row_sum).log()).otherwise(0.0) for c in bcols]
    ).alias("entropy")
    return lf.with_columns(entropy)

discovery

Signal profile artifacts for CAN cache builds.

build_signal_profiles 

build_signal_profiles(df: DataFrame) -> pl.DataFrame

Aggregate raw CAN rows into one profile per vehicle/arbitration ID.

Source code in graphids/core/data/discovery/hypotheses.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
def build_signal_profiles(df: pl.DataFrame) -> pl.DataFrame:
    """Aggregate raw CAN rows into one profile per vehicle/arbitration ID."""

    missing = [c for c in ("vehicle_id", "arb_id") if c not in df.columns]
    if missing:
        raise ValueError(f"build_signal_profiles missing columns: {missing}")

    sort_cols = [c for c in ("vehicle_id", "arb_id", "timestamp") if c in df.columns]
    if sort_cols:
        df = df.sort(sort_cols)

    aggs: list[pl.Expr] = [pl.len().cast(pl.Int64).alias("msg_count")]
    if "timestamp" in df.columns:
        aggs.extend(
            [
                pl.col("timestamp").min().cast(pl.Float64).alias("timestamp_min"),
                pl.col("timestamp").max().cast(pl.Float64).alias("timestamp_max"),
                (pl.col("timestamp").max() - pl.col("timestamp").min()).cast(pl.Float64).alias("duration"),
                pl.col("timestamp").diff().mean().cast(pl.Float64).alias("iat_mean"),
                pl.col("timestamp").diff().std().fill_nan(0).cast(pl.Float64).alias("iat_std"),
            ]
        )
    if "entropy" in df.columns:
        aggs.extend(
            [
                pl.col("entropy").mean().cast(pl.Float64).alias("entropy_mean"),
                pl.col("entropy").std().fill_nan(0).cast(pl.Float64).alias("entropy_std"),
            ]
        )

    byte_cols = _byte_cols(df)
    aggs.extend(pl.col(c).mean().cast(pl.Float64).alias(f"{c}_mean") for c in byte_cols)
    aggs.extend(pl.col(c).std().fill_nan(0).cast(pl.Float64).alias(f"{c}_std") for c in byte_cols)
    aggs.extend((pl.col(c).max() - pl.col(c).min()).cast(pl.Float64).alias(f"{c}_range") for c in byte_cols)
    if byte_cols:
        aggs.append(
            pl.mean_horizontal(
                *[(pl.col(c).diff().abs().drop_nulls() > 0).mean().fill_null(0) for c in byte_cols]
            ).cast(pl.Float64).alias("change_rate")
        )
    if "attack" in df.columns:
        aggs.extend(
            [
                pl.col("attack").max().cast(pl.Int64).alias("attack_max"),
                pl.col("attack").mean().cast(pl.Float64).alias("attack_rate"),
            ]
        )

    return df.group_by("vehicle_id", "arb_id").agg(*aggs).with_columns(
        pl.concat_str([pl.col("vehicle_id").cast(pl.Utf8), pl.col("arb_id").cast(pl.Utf8)], separator="::").alias("signal_key")
    )

initialize_hypotheses 

initialize_hypotheses(profiles: DataFrame) -> pl.DataFrame

Create empty editable mapping rows for profile review.

Source code in graphids/core/data/discovery/hypotheses.py 
69
70
71
72
73
74
75
76
77
78
79
80
81
def initialize_hypotheses(profiles: pl.DataFrame) -> pl.DataFrame:
    """Create empty editable mapping rows for profile review."""

    required = ["vehicle_id", "arb_id", "signal_key"]
    missing = [c for c in required if c not in profiles.columns]
    if missing:
        raise ValueError(f"initialize_hypotheses missing columns: {missing}")
    return profiles.select(*required).with_columns(
        pl.lit(None, dtype=pl.Utf8).alias("candidate_canonical_id"),
        pl.lit(0.0).cast(pl.Float64).alias("confidence"),
        pl.lit("unreviewed").alias("status"),
        pl.lit("").cast(pl.Utf8).alias("evidence"),
    )

hypotheses

Signal profile artifacts written beside graph caches.

build_signal_profiles 
build_signal_profiles(df: DataFrame) -> pl.DataFrame

Aggregate raw CAN rows into one profile per vehicle/arbitration ID.

Source code in graphids/core/data/discovery/hypotheses.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
def build_signal_profiles(df: pl.DataFrame) -> pl.DataFrame:
    """Aggregate raw CAN rows into one profile per vehicle/arbitration ID."""

    missing = [c for c in ("vehicle_id", "arb_id") if c not in df.columns]
    if missing:
        raise ValueError(f"build_signal_profiles missing columns: {missing}")

    sort_cols = [c for c in ("vehicle_id", "arb_id", "timestamp") if c in df.columns]
    if sort_cols:
        df = df.sort(sort_cols)

    aggs: list[pl.Expr] = [pl.len().cast(pl.Int64).alias("msg_count")]
    if "timestamp" in df.columns:
        aggs.extend(
            [
                pl.col("timestamp").min().cast(pl.Float64).alias("timestamp_min"),
                pl.col("timestamp").max().cast(pl.Float64).alias("timestamp_max"),
                (pl.col("timestamp").max() - pl.col("timestamp").min()).cast(pl.Float64).alias("duration"),
                pl.col("timestamp").diff().mean().cast(pl.Float64).alias("iat_mean"),
                pl.col("timestamp").diff().std().fill_nan(0).cast(pl.Float64).alias("iat_std"),
            ]
        )
    if "entropy" in df.columns:
        aggs.extend(
            [
                pl.col("entropy").mean().cast(pl.Float64).alias("entropy_mean"),
                pl.col("entropy").std().fill_nan(0).cast(pl.Float64).alias("entropy_std"),
            ]
        )

    byte_cols = _byte_cols(df)
    aggs.extend(pl.col(c).mean().cast(pl.Float64).alias(f"{c}_mean") for c in byte_cols)
    aggs.extend(pl.col(c).std().fill_nan(0).cast(pl.Float64).alias(f"{c}_std") for c in byte_cols)
    aggs.extend((pl.col(c).max() - pl.col(c).min()).cast(pl.Float64).alias(f"{c}_range") for c in byte_cols)
    if byte_cols:
        aggs.append(
            pl.mean_horizontal(
                *[(pl.col(c).diff().abs().drop_nulls() > 0).mean().fill_null(0) for c in byte_cols]
            ).cast(pl.Float64).alias("change_rate")
        )
    if "attack" in df.columns:
        aggs.extend(
            [
                pl.col("attack").max().cast(pl.Int64).alias("attack_max"),
                pl.col("attack").mean().cast(pl.Float64).alias("attack_rate"),
            ]
        )

    return df.group_by("vehicle_id", "arb_id").agg(*aggs).with_columns(
        pl.concat_str([pl.col("vehicle_id").cast(pl.Utf8), pl.col("arb_id").cast(pl.Utf8)], separator="::").alias("signal_key")
    )
initialize_hypotheses 
initialize_hypotheses(profiles: DataFrame) -> pl.DataFrame

Create empty editable mapping rows for profile review.

Source code in graphids/core/data/discovery/hypotheses.py 
69
70
71
72
73
74
75
76
77
78
79
80
81
def initialize_hypotheses(profiles: pl.DataFrame) -> pl.DataFrame:
    """Create empty editable mapping rows for profile review."""

    required = ["vehicle_id", "arb_id", "signal_key"]
    missing = [c for c in required if c not in profiles.columns]
    if missing:
        raise ValueError(f"initialize_hypotheses missing columns: {missing}")
    return profiles.select(*required).with_columns(
        pl.lit(None, dtype=pl.Utf8).alias("candidate_canonical_id"),
        pl.lit(0.0).cast(pl.Float64).alias("confidence"),
        pl.lit("unreviewed").alias("status"),
        pl.lit("").cast(pl.Utf8).alias("evidence"),
    )

extract

Extract and cache fusion features as a TensorDict.

Each upstream model implements extract_features(batch, device) -> dict[str, Tensor] returning per-graph named feature tensors. This module collects those dicts under the model name (vgae, gat, ...), stacks across batches, and saves the resulting nested TensorDict to disk. No flat state vector, no offsets — the fusion side reads keys directly.

Invoked by the experiment extraction pipeline. Idempotent on output_dir.

extract_states 

extract_states(*, checkpoints: dict[str, str], dataset: str, output_dir: str, max_samples: int = 150000, max_val_samples: int = 30000, batch_size: int = 256, seed: int = 42, val_fraction: float = 0.2, representation_cfg: GraphRepresentationCfg) -> None

Load model checkpoints, extract and cache fusion features.

Idempotent per-file: each split's cache is checked independently so re-running after adding test splits only extracts the missing files.

Source code in graphids/core/data/extract.py 
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
def extract_states(
    *,
    checkpoints: dict[str, str],
    dataset: str,
    output_dir: str,
    max_samples: int = 150_000,
    max_val_samples: int = 30_000,
    batch_size: int = 256,
    seed: int = 42,
    val_fraction: float = 0.2,
    representation_cfg: GraphRepresentationCfg,
) -> None:
    """Load model checkpoints, extract and cache fusion features.

    Idempotent per-file: each split's cache is checked independently so
    re-running after adding test splits only extracts the missing files.
    """
    from graphids.core.data.datamodule.graph import GraphDataModule
    from graphids.core.data.datasets.can_bus import CANBusSource
    from graphids.core.models.base import safe_load_checkpoint

    # Build DM first so test split names are known before the idempotency check.
    source = CANBusSource(
        name=dataset,
        seed=seed,
        val_fraction=val_fraction,
        representation_cfg=representation_cfg,
    )
    dm = GraphDataModule(dataset=source, dynamic_batching=False)
    dm.setup(None)

    out = Path(output_dir)
    train_path = out / TRAIN_FILENAME
    val_path = out / VAL_FILENAME
    test_paths = {name: out / f"{name}_states.pt" for name in dm.test_datasets.keys()}

    def _version_ok(p: Path) -> bool:
        if not p.exists():
            return False
        try:
            return (
                torch.load(p, map_location="cpu", weights_only=False).get("version")
                == CACHE_VERSION
            )
        except Exception:
            return False

    if all(_version_ok(p) for p in [train_path, val_path, *test_paths.values()]):
        log.info("cache_hit", output_dir=str(out), version=CACHE_VERSION)
        return

    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
    models = {}
    for model_type, ckpt_path in checkpoints.items():
        log.info("loading_model", model_type=model_type, ckpt=ckpt_path)
        module = safe_load_checkpoint(model_type, Path(ckpt_path), map_location=device)
        module.to(device).eval()
        models[model_type] = module

    train_ds, val_ds = dm.train_dataset, dm.val_dataset

    # Stash schema's attack code → name map so the fusion test path can emit
    # ``auroc_per_attack/{name}`` keys (looked up in FusionDataModule).
    schema = getattr(type(train_ds), "SCHEMA", None)
    names_map = getattr(schema, "attack_type_names", None) if schema is not None else None
    blob_extras = {"version": CACHE_VERSION, "attack_type_names": dict(names_map or {0: "benign"})}

    out.mkdir(parents=True, exist_ok=True)

    if not _version_ok(train_path):
        log.info("extracting_train", n_graphs=len(train_ds), max_samples=max_samples)
        train_td = _extract_states(models, list(train_ds), device, max_samples, batch_size).cpu()
        torch.save({"td": train_td.to_dict(), **blob_extras}, train_path)

    if not _version_ok(val_path):
        log.info("extracting_val", n_graphs=len(val_ds), max_samples=max_val_samples)
        val_td = _extract_states(models, list(val_ds), device, max_val_samples, batch_size).cpu()
        torch.save({"td": val_td.to_dict(), **blob_extras}, val_path)

    for name, test_ds in dm.test_datasets.items():
        p = test_paths[name]
        if not _version_ok(p):
            n = len(test_ds)
            log.info("extracting_test", split=name, n_graphs=n)
            test_td = _extract_states(models, list(test_ds), device, n, batch_size).cpu()
            torch.save({"td": test_td.to_dict(), **blob_extras}, p)

    log.info("states_saved", output_dir=str(out), version=CACHE_VERSION)

preprocessing

Core graph preprocessing.

graph_tables_to_pyg 

graph_tables_to_pyg(tables: GraphTables, *, node_col_order: list[str], edge_col_order: tuple[str, ...], label_exprs: list[Expr]) -> tuple[Data, dict, int, int]

Compose staged tables into pre-collated PyG tensors.

Source code in graphids/core/data/preprocessing/pyg.py 
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
def graph_tables_to_pyg(
    tables: GraphTables,
    *,
    node_col_order: list[str],
    edge_col_order: tuple[str, ...],
    label_exprs: list[pl.Expr],
) -> tuple[Data, dict, int, int]:
    """Compose staged tables into pre-collated PyG tensors."""
    label_names = [e.meta.output_name() for e in label_exprs]
    x = tables.node_stats.select(node_col_order).fill_null(0).fill_nan(0).to_torch(dtype=pl.Float32)
    node_id = tables.node_stats.select("node_id").to_torch(dtype=pl.Int64).squeeze(-1)
    edge_index = tables.edge_df.select("src_local", "dst_local").to_torch(dtype=pl.Int64).t().contiguous()
    edge_attr = tables.edge_df.select(list(edge_col_order)).fill_null(0).fill_nan(0).to_torch(dtype=pl.Float32)
    kept_wids = tables.node_stats.group_by("_wid", maintain_order=True).first().select("_wid")
    num_graphs = len(kept_wids)
    node_counts = tables.node_stats.group_by("_wid", maintain_order=True).len()["len"]
    edge_counts = tables.edge_df.group_by("_wid", maintain_order=True).len()["len"]
    node_slice = _slices_from_counts(node_counts)
    edge_slice = _slices_from_counts(edge_counts)
    graph_idx = torch.arange(num_graphs + 1, dtype=torch.long)
    labels_aligned = kept_wids.join(tables.labels, on="_wid", how="left").fill_null(0)
    label_tensors = {
        n: labels_aligned.select(n).to_torch(dtype=pl.Int64).squeeze(-1)
        for n in label_names
    }
    extra_tensors: dict[str, torch.Tensor] = {}
    extra_slices: dict[str, torch.Tensor] = {}
    node_optional_cols = {
        "sequence_id": "node_sequence_id",
        "sequence_step": "node_sequence_step",
        "sequence_length": "node_sequence_length",
        "sequence_stride": "node_sequence_stride",
        "snapshot_wid": "node_snapshot_wid",
        "window_start_row": "node_window_start_row",
        "window_end_row": "node_window_end_row",
        "window_ordinal": "node_window_ordinal",
    }
    for col, attr in node_optional_cols.items():
        if col in tables.node_stats.columns:
            extra_tensors[attr] = _optional_tensor(tables.node_stats, col, dtype=pl.Int64)
            extra_slices[attr] = node_slice

    edge_optional_cols = {
        "sequence_id": "edge_sequence_id",
        "sequence_step": "edge_sequence_step",
        "sequence_length": "edge_sequence_length",
        "sequence_stride": "edge_sequence_stride",
        "snapshot_wid": "edge_snapshot_wid",
        "window_start_row": "edge_window_start_row",
        "window_end_row": "edge_window_end_row",
        "window_ordinal": "edge_window_ordinal",
    }
    for col, attr in edge_optional_cols.items():
        if col in tables.edge_df.columns:
            extra_tensors[attr] = _optional_tensor(tables.edge_df, col, dtype=pl.Int64)
            extra_slices[attr] = edge_slice

    extra_tensors["graph_wid"] = _optional_tensor(kept_wids, "_wid", dtype=pl.Int64)
    extra_slices["graph_wid"] = graph_idx

    graph_optional_cols = (
        "sequence_id",
        "sequence_length",
        "sequence_stride",
        "target_snapshot_wid",
        "window_start_row",
        "window_end_row",
        "window_ordinal",
        "source_dir_n_unique",
        "source_file_n_unique",
    )
    for col in graph_optional_cols:
        if col in labels_aligned.columns:
            extra_tensors[col] = _optional_tensor(labels_aligned, col, dtype=pl.Int64)
            extra_slices[col] = graph_idx

    data = Data(
        x=x,
        edge_index=edge_index,
        edge_attr=edge_attr,
        node_id=node_id,
        **label_tensors,
        **extra_tensors,
    )
    slices = {
        "x": node_slice,
        "edge_index": edge_slice,
        "edge_attr": edge_slice,
        "node_id": node_slice,
        **{n: graph_idx for n in label_names},
        **extra_slices,
    }
    return data, slices, num_graphs, tables.n_rows

curriculum

Curriculum difficulty scorers used by the graph datamodule.

score_random 
score_random(graphs: list, seed: int = 0) -> torch.Tensor

Uniform random per-graph difficulty for curriculum control runs.

Source code in graphids/core/data/preprocessing/curriculum.py 
47
48
49
50
def score_random(graphs: list, seed: int = 0) -> torch.Tensor:
    """Uniform random per-graph difficulty for curriculum control runs."""
    g = torch.Generator().manual_seed(int(seed))
    return torch.rand(len(graphs), generator=g)
score_vgae 
score_vgae(graphs: list, ckpt_path: str) -> torch.Tensor

Per-graph reconstruction MSE from a trained VGAE checkpoint.

Higher = harder. Loads the VGAE on CPU, computes per-graph mean MSE via torch_geometric.utils.scatter, releases the model.

Source code in graphids/core/data/preprocessing/curriculum.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
@torch.no_grad()
def score_vgae(graphs: list, ckpt_path: str) -> torch.Tensor:
    """Per-graph reconstruction MSE from a trained VGAE checkpoint.

    Higher = harder. Loads the VGAE on CPU, computes per-graph mean MSE
    via ``torch_geometric.utils.scatter``, releases the model.
    """
    if not ckpt_path:
        raise ValueError("score_vgae requires a non-empty ckpt_path")

    from torch_geometric.loader import DataLoader as PyGDataLoader
    from torch_geometric.utils import scatter

    from graphids.core.models.base import safe_load_checkpoint

    vgae = safe_load_checkpoint("vgae", Path(ckpt_path), map_location="cpu")
    try:
        device = next(vgae.parameters()).device
        was_training = vgae.training
        vgae.eval()
        try:
            scores: list[float] = []
            for batch in PyGDataLoader(graphs, batch_size=500):
                batch = batch.clone().to(device, non_blocking=True)
                cont, _canid, _nbr, _z, _kl, _edge = vgae(batch)
                node_mse = (cont - batch.x).pow(2).mean(dim=1)
                graph_mse = scatter(node_mse, batch.batch, reduce="mean")
                scores.extend(graph_mse.tolist())
        finally:
            vgae.train(was_training)
    finally:
        del vgae
        gc.collect()
    return torch.tensor(scores, dtype=torch.float)

edge_policy

Declarative edge construction policies for graph preprocessing.

EdgePolicy dataclass 
EdgePolicy(name: str, src_col: str = 'node_id', dst_col: str = 'node_id', dst_shift: int = -1, src_alias: str = 'src', dst_alias: str = 'dst')

How to derive directed edges from windowed rows.

temporal_edge_policy 
temporal_edge_policy(*, src_col: str = 'node_id', dst_col: str = 'node_id', dst_shift: int = -1) -> EdgePolicy

Temporal adjacency policy: edge from row t to row t + dst_shift.

Source code in graphids/core/data/preprocessing/edge_policy.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
def temporal_edge_policy(
    *,
    src_col: str = "node_id",
    dst_col: str = "node_id",
    dst_shift: int = -1,
) -> EdgePolicy:
    """Temporal adjacency policy: edge from row ``t`` to row ``t + dst_shift``."""
    return EdgePolicy(
        name="temporal_shift",
        src_col=src_col,
        dst_col=dst_col,
        dst_shift=dst_shift,
    )

graph_ops

Composable graph transforms over node/edge preprocessing tables.

GraphTransform dataclass 
GraphTransform(name: str, requires: tuple[str, ...], produces: tuple[str, ...], fn: Callable[[DataFrame, DataFrame], tuple[DataFrame, DataFrame]])

A declarative graph transform with explicit input/output columns.

default_graph_transforms 
default_graph_transforms() -> list[GraphTransform]

Default graph transforms used in cache builds.

Source code in graphids/core/data/preprocessing/graph_ops.py 
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
def default_graph_transforms() -> list[GraphTransform]:
    """Default graph transforms used in cache builds."""
    return [
        GraphTransform(
            name="edge_frequency",
            requires=("_wid", "src", "dst"),
            produces=("edge_freq",),
            fn=_add_edge_frequency,
        ),
        GraphTransform(
            name="bidir",
            requires=("_wid", "src", "dst"),
            produces=("bidir",),
            fn=_add_bidir,
        ),
        GraphTransform(
            name="topology",
            requires=("_wid", "node_id", "src", "dst"),
            produces=("clustering_coeff", "in_degree", "out_degree"),
            fn=_add_graph_topology,
        ),
    ]
secondary_graph_transforms 
secondary_graph_transforms() -> list[GraphTransform]

Additional exploratory graph transforms used in feature tests.

Source code in graphids/core/data/preprocessing/graph_ops.py 
151
152
153
154
155
156
157
158
159
160
def secondary_graph_transforms() -> list[GraphTransform]:
    """Additional exploratory graph transforms used in feature tests."""
    return [
        GraphTransform(
            name="secondary_node_stats",
            requires=("in_degree", "out_degree"),
            produces=("in_out_ratio", "neighbor_entropy"),
            fn=_add_secondary_node_stats,
        )
    ]

materialization

Raw CAN rows to graph tables.

metadata

Cache metadata contract for dataset builds.

load_metadata 
load_metadata(cache_dir: Path) -> dict[str, Any]

Read and version-gate cache_metadata.json.

Source code in graphids/core/data/preprocessing/metadata.py 
35
36
37
38
39
40
41
42
43
44
45
46
def load_metadata(cache_dir: Path) -> dict[str, Any]:
    """Read and version-gate ``cache_metadata.json``."""
    path = cache_dir / "cache_metadata.json"
    if not path.exists():
        raise FileNotFoundError(f"cache_metadata.json missing at {path}; run rebuild-caches")
    meta = json.loads(path.read_text())
    ver = meta.get("metadata_schema_version")
    if ver != METADATA_SCHEMA_VERSION:
        raise ValueError(
            f"{path} schema {ver!r} != expected {METADATA_SCHEMA_VERSION}; rebuild caches"
        )
    return meta
merge_split_into_metadata 
merge_split_into_metadata(cache_dir: Path, split_name: str, split_entry: dict[str, Any], *, invariants: dict[str, Any], dataset_name: str, num_arb_ids: int) -> dict[str, Any]

Merge one split's entry into cache_metadata.json under FileLock.

First writer seeds top-level fields; later writers must match invariants + dataset name or raise.

Source code in graphids/core/data/preprocessing/metadata.py 
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
def merge_split_into_metadata(
    cache_dir: Path,
    split_name: str,
    split_entry: dict[str, Any],
    *,
    invariants: dict[str, Any],
    dataset_name: str,
    num_arb_ids: int,
) -> dict[str, Any]:
    """Merge one split's entry into ``cache_metadata.json`` under FileLock.

    First writer seeds top-level fields; later writers must match
    invariants + dataset name or raise.
    """
    cache_dir.mkdir(parents=True, exist_ok=True)
    meta_path = cache_dir / "cache_metadata.json"

    missing = [k for k in INVARIANT_KEYS if k not in invariants]
    if missing:
        raise ValueError(f"invariants missing required keys: {missing}")

    with FileLock(str(cache_dir / ".metadata_lock")):
        existing: dict[str, Any] = {}
        if meta_path.exists():
            existing = json.loads(meta_path.read_text())
            ver = existing.get("metadata_schema_version")
            if ver not in (None, METADATA_SCHEMA_VERSION):
                raise ValueError(
                    f"{meta_path} schema {ver!r} != {METADATA_SCHEMA_VERSION}; "
                    "delete or rebuild --delete-existing"
                )
            for k in INVARIANT_KEYS:
                if k in existing and existing[k] != invariants[k]:
                    raise ValueError(
                        f"{meta_path} invariant mismatch: {k}={existing[k]!r} "
                        f"!= writer {invariants[k]!r}; rebuild caches"
                    )
            if existing.get("dataset") not in (None, dataset_name):
                raise ValueError(
                    f"{meta_path} dataset={existing.get('dataset')!r} != writer {dataset_name!r}"
                )

        meta: dict[str, Any] = {
            "metadata_schema_version": METADATA_SCHEMA_VERSION,
            "dataset": dataset_name,
            "built_at": existing.get("built_at") or datetime.now(UTC).isoformat(),
            "num_arb_ids": num_arb_ids,
            **{k: invariants[k] for k in INVARIANT_KEYS},
            "splits": dict(existing.get("splits") or {}),
        }
        meta["splits"][split_name] = split_entry
        meta["aggregate"] = _aggregate(meta["splits"])
        atomic_write_text(meta_path, json.dumps(meta, indent=2, sort_keys=True))
        return meta

pyg

PyG tensor packing primitives for staged graph tables.

graph_tables_to_pyg 
graph_tables_to_pyg(tables: GraphTables, *, node_col_order: list[str], edge_col_order: tuple[str, ...], label_exprs: list[Expr]) -> tuple[Data, dict, int, int]

Compose staged tables into pre-collated PyG tensors.

Source code in graphids/core/data/preprocessing/pyg.py 
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
def graph_tables_to_pyg(
    tables: GraphTables,
    *,
    node_col_order: list[str],
    edge_col_order: tuple[str, ...],
    label_exprs: list[pl.Expr],
) -> tuple[Data, dict, int, int]:
    """Compose staged tables into pre-collated PyG tensors."""
    label_names = [e.meta.output_name() for e in label_exprs]
    x = tables.node_stats.select(node_col_order).fill_null(0).fill_nan(0).to_torch(dtype=pl.Float32)
    node_id = tables.node_stats.select("node_id").to_torch(dtype=pl.Int64).squeeze(-1)
    edge_index = tables.edge_df.select("src_local", "dst_local").to_torch(dtype=pl.Int64).t().contiguous()
    edge_attr = tables.edge_df.select(list(edge_col_order)).fill_null(0).fill_nan(0).to_torch(dtype=pl.Float32)
    kept_wids = tables.node_stats.group_by("_wid", maintain_order=True).first().select("_wid")
    num_graphs = len(kept_wids)
    node_counts = tables.node_stats.group_by("_wid", maintain_order=True).len()["len"]
    edge_counts = tables.edge_df.group_by("_wid", maintain_order=True).len()["len"]
    node_slice = _slices_from_counts(node_counts)
    edge_slice = _slices_from_counts(edge_counts)
    graph_idx = torch.arange(num_graphs + 1, dtype=torch.long)
    labels_aligned = kept_wids.join(tables.labels, on="_wid", how="left").fill_null(0)
    label_tensors = {
        n: labels_aligned.select(n).to_torch(dtype=pl.Int64).squeeze(-1)
        for n in label_names
    }
    extra_tensors: dict[str, torch.Tensor] = {}
    extra_slices: dict[str, torch.Tensor] = {}
    node_optional_cols = {
        "sequence_id": "node_sequence_id",
        "sequence_step": "node_sequence_step",
        "sequence_length": "node_sequence_length",
        "sequence_stride": "node_sequence_stride",
        "snapshot_wid": "node_snapshot_wid",
        "window_start_row": "node_window_start_row",
        "window_end_row": "node_window_end_row",
        "window_ordinal": "node_window_ordinal",
    }
    for col, attr in node_optional_cols.items():
        if col in tables.node_stats.columns:
            extra_tensors[attr] = _optional_tensor(tables.node_stats, col, dtype=pl.Int64)
            extra_slices[attr] = node_slice

    edge_optional_cols = {
        "sequence_id": "edge_sequence_id",
        "sequence_step": "edge_sequence_step",
        "sequence_length": "edge_sequence_length",
        "sequence_stride": "edge_sequence_stride",
        "snapshot_wid": "edge_snapshot_wid",
        "window_start_row": "edge_window_start_row",
        "window_end_row": "edge_window_end_row",
        "window_ordinal": "edge_window_ordinal",
    }
    for col, attr in edge_optional_cols.items():
        if col in tables.edge_df.columns:
            extra_tensors[attr] = _optional_tensor(tables.edge_df, col, dtype=pl.Int64)
            extra_slices[attr] = edge_slice

    extra_tensors["graph_wid"] = _optional_tensor(kept_wids, "_wid", dtype=pl.Int64)
    extra_slices["graph_wid"] = graph_idx

    graph_optional_cols = (
        "sequence_id",
        "sequence_length",
        "sequence_stride",
        "target_snapshot_wid",
        "window_start_row",
        "window_end_row",
        "window_ordinal",
        "source_dir_n_unique",
        "source_file_n_unique",
    )
    for col in graph_optional_cols:
        if col in labels_aligned.columns:
            extra_tensors[col] = _optional_tensor(labels_aligned, col, dtype=pl.Int64)
            extra_slices[col] = graph_idx

    data = Data(
        x=x,
        edge_index=edge_index,
        edge_attr=edge_attr,
        node_id=node_id,
        **label_tensors,
        **extra_tensors,
    )
    slices = {
        "x": node_slice,
        "edge_index": edge_slice,
        "edge_attr": edge_slice,
        "node_id": node_slice,
        **{n: graph_idx for n in label_names},
        **extra_slices,
    }
    return data, slices, num_graphs, tables.n_rows

representations

Graph representation configs used by preprocessing.

scaler

Per-column feature scalers for tensor-based graph preprocessing.

splits

Leakage-safe train/validation graph indices.

graph_touched_base_units 
graph_touched_base_units(data: Data, slices: dict[str, Tensor]) -> list[tuple[int, ...]]

Return base snapshot windows touched by each graph.

Source code in graphids/core/data/preprocessing/splits.py 
82
83
84
85
86
87
88
89
90
def graph_touched_base_units(data: Data, slices: dict[str, Tensor]) -> list[tuple[int, ...]]:
    """Return base snapshot windows touched by each graph."""

    n_graphs = _num_graphs(data, slices)
    if hasattr(data, "node_snapshot_wid") and "node_snapshot_wid" in slices:
        return [_sliced_unique(data, slices, "node_snapshot_wid", idx) for idx in range(n_graphs)]
    if hasattr(data, "graph_wid"):
        return [(int(v),) for v in data.graph_wid[:n_graphs].tolist()]
    return [(idx,) for idx in range(n_graphs)]

vocab

Vocabulary scan, digest, persist, and load primitives.

load_vocab 
load_vocab(path: Path) -> tuple[dict[str, int], str]

Return (entries, digest) from a persisted vocab file.

Source code in graphids/core/data/preprocessing/vocab.py 
61
62
63
64
def load_vocab(path: Path) -> tuple[dict[str, int], str]:
    """Return ``(entries, digest)`` from a persisted vocab file."""
    payload = json.loads(path.read_text())
    return payload["entries"], payload["digest"]
persist_vocab 
persist_vocab(vocab: dict[Any, int], path: Path) -> str

Atomic write and return the digest.

Source code in graphids/core/data/preprocessing/vocab.py 
49
50
51
52
53
54
55
56
57
58
def persist_vocab(vocab: dict[Any, int], path: Path) -> str:
    """Atomic write and return the digest."""
    digest = vocab_digest(vocab)
    payload = {
        "digest": digest,
        "unk_index": UNK_INDEX,
        "entries": {str(k): v for k, v in vocab.items()},
    }
    atomic_write_text(path, json.dumps(payload, indent=2, sort_keys=True))
    return digest
scan_arb_ids 
scan_arb_ids(raw_dir: Path, source_dirs: list[str]) -> list[Any]

Sorted unique arb_id across every CSV under source_dirs.

Source code in graphids/core/data/preprocessing/vocab.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
def scan_arb_ids(raw_dir: Path, source_dirs: list[str]) -> list[Any]:
    """Sorted unique ``arb_id`` across every CSV under ``source_dirs``."""
    if not source_dirs:
        raise ValueError("source_dirs is empty; cannot scan for arb_ids")
    frames: list[pl.LazyFrame] = []
    for sub in source_dirs:
        sub_path = raw_dir / sub
        if not sub_path.is_dir():
            raise FileNotFoundError(f"Source dir missing: {sub_path}")
        for csv_path in sorted(sub_path.rglob("*.csv")):
            lf = pl.scan_csv(csv_path)
            cols = lf.collect_schema().names()
            col = "arbitration_id" if "arbitration_id" in cols else "arb_id"
            if col not in cols:
                raise ValueError(
                    f"{csv_path} has neither arbitration_id nor arb_id; got {cols!r}"
                )
            frames.append(lf.select(pl.col(col).alias("arb_id")))
    if not frames:
        raise ValueError(f"No CSVs under {source_dirs!r} in {raw_dir}")
    return pl.concat(frames).collect()["arb_id"].unique().sort().to_list()
vocab_digest 
vocab_digest(vocab: dict[Any, int]) -> str

SHA256 over (id, index) pairs sorted by index.

Source code in graphids/core/data/preprocessing/vocab.py 
40
41
42
43
44
45
46
def vocab_digest(vocab: dict[Any, int]) -> str:
    """SHA256 over ``(id, index)`` pairs sorted by index."""
    canon = json.dumps(
        sorted(((str(k), v) for k, v in vocab.items()), key=lambda kv: kv[1]),
        sort_keys=True,
    )
    return hashlib.sha256(canon.encode()).hexdigest()

state

Process-level dataset cache.

DatasetState dataclass 

DatasetState(train: Any, val: Any, test: dict[str, Any])

Ready-to-serve train/val/test splits.

clear_cache 

clear_cache() -> None

Drop all cached states. Intended for test teardown.

Source code in graphids/core/data/state.py 
37
38
39
def clear_cache() -> None:
    """Drop all cached states. Intended for test teardown."""
    _REGISTRY.clear()

get_or_build 

get_or_build(dataset: _CacheableDataset) -> DatasetState

Return cached DatasetState for dataset.

Source code in graphids/core/data/state.py 
27
28
29
30
31
32
33
34
def get_or_build(dataset: _CacheableDataset) -> DatasetState:
    """Return cached ``DatasetState`` for ``dataset``."""
    key = dataset.cache_key
    state = _REGISTRY.get(key)
    if state is None:
        state = dataset.build()
        _REGISTRY[key] = state
    return state