torchcell.adapters.CellAdapter

class CellAdapter(config: DictConfig, dataset: Dataset, process_workers: int, io_workers: int, chunk_size: int = 10000, loader_batch_size: int = 1000, inprocess_max_records: int = 0)

Bases: object

Convert a cell experiment dataset into BioCypher nodes and edges for import.

log_method_table() → None

Log a wandb table summarizing the configured node and edge methods.

get_data_by_type(chunk_processing_func: Callable[[...], Any], method_name: str, is_edge: bool = False) → Iterator[Any]

Yield processed data by chunking the dataset and mapping over processes.

Submission is windowed: at most process_workers + 2 chunks are in flight, and a new one is submitted only as an earlier result is consumed. Results are still yielded in submission order, so callers see no behavioral change.

The window is what keeps peak memory a function of chunk size rather than dataset size. Submitting every chunk up front lets workers run arbitrarily far ahead of the consumer, and because results are yielded in order, each finished chunk is held in this process until its turn – for a 20.7M-record dataset at chunk_size 1e5 that is 207 results of ~1 GB each. Job 1543 died that way, with a worker’s fork for its loader failing ENOMEM against the container’s limit.

The window bounds what THIS process holds; CHUNKS_PER_WORKER bounds what a worker holds, by rebuilding the pool every group of chunks. Job 1545 needed both: windowed but long-lived workers still ratcheted to 38-50 GB of heap apiece.

data_chunker() → Callable[[...], list[Any]]

Wrap a chunk handler so it loads, transforms, and collects each item.

get_memory_reduction_factor(method_name: str, is_edge: bool = False) → float

Return the configured memory reduction factor for a method (default 1.0).

The single-pass method carries every chunked method’s output per record, so its factor is the smallest configured factor divided by the number of methods folded in: a chunk then holds about as much as one method’s chunk did.

get_nodes() → Iterator[BioCypherNode]

Yield BioCypher nodes from every enabled node method in config order.

get_edges() → Iterator[BioCypherEdge]

Yield BioCypher edges from every enabled edge method in config order.

property supported_node_methods: list[str]

Return the names of all registered node methods.

property supported_edge_methods: list[str]

Return the names of all registered edge methods.