torchcell.data.Aggregator

class Aggregator(root: str)[source]

Bases: ABC

Abstract base that groups experiment pairs into an aggregated LMDB store.

Return a grouping key for the experiment pair.

The key determines which experiments are aggregated together.

abstractmethod aggregate_key_raw(record: dict[str, Any]) → str[source]

Return the grouping key for ONE raw record dict (pre-pydantic).

Streaming counterpart of aggregate_check(): process calls this per record straight off the stored JSON, so grouping never requires the whole dataset in memory. Must produce the same key aggregate_check would for that record.

create_aggregate_entry(experiments_to_aggregate: list[list[dict[str, Experiment | FitnessExperiment | GeneInteractionExperiment | GeneEssentialityExperiment | SyntheticLethalityExperiment | SyntheticRescueExperiment | CalMorphExperiment | MicroarrayExpressionExperiment | MicroarrayExpressionScoreExperiment | MicroarrayExpressionRatioExperiment | InductionTimecourseExpressionExperiment | RNASeqExpressionExperiment | PseudobulkExpressionExperiment | VisualScoreExperiment | MetaboliteExperiment | ProteinAbundanceExperiment | ProteinFoldChangeExperiment | EnvironmentResponseExperiment | StrainEnvironmentResponseExperiment | SegregantGrowthExperiment | SegregantPerturbationExperiment | ProductTiterExperiment | ProteinTurnoverExperiment | ProteinSynthesisRateExperiment | FluxExperiment | PromoterActivityExperiment | PooledSortSeqExpressionExperiment | BiosensorFluorescenceExperiment | GradedGeneEssentialityExperiment | PhosphositeIntensityExperiment | MorphologyExperiment | MrnaNumberFractionExperiment | FitnessContrastExperiment | SegregantExpressionExperiment | LifeCycleExperiment | PseudobulkUmiCountExperiment | GenomeArchitectureExperiment | ExperimentReference | FitnessExperimentReference | GeneInteractionExperimentReference | GeneEssentialityExperimentReference | SyntheticLethalityExperimentReference | SyntheticRescueExperimentReference | CalMorphExperimentReference | MicroarrayExpressionExperimentReference | MicroarrayExpressionScoreExperimentReference | MicroarrayExpressionRatioExperimentReference | InductionTimecourseExpressionExperimentReference | RNASeqExpressionExperimentReference | PseudobulkExpressionExperimentReference | VisualScoreExperimentReference | MetaboliteExperimentReference | ProteinAbundanceExperimentReference | ProteinFoldChangeExperimentReference | EnvironmentResponseExperimentReference | StrainEnvironmentResponseExperimentReference | SegregantGrowthExperimentReference | SegregantPerturbationExperimentReference | ProductTiterExperimentReference | ProteinTurnoverExperimentReference | ProteinSynthesisRateExperimentReference | FluxExperimentReference | PromoterActivityExperimentReference | PooledSortSeqExpressionExperimentReference | BiosensorFluorescenceExperimentReference | GradedGeneEssentialityExperimentReference | PhosphositeIntensityExperimentReference | MorphologyExperimentReference | MrnaNumberFractionExperimentReference | FitnessContrastExperimentReference | SegregantExpressionExperimentReference | LifeCycleExperimentReference | PseudobulkUmiCountExperimentReference | GenomeArchitectureExperimentReference]]]) → list[dict[str, Experiment | FitnessExperiment | GeneInteractionExperiment | GeneEssentialityExperiment | SyntheticLethalityExperiment | SyntheticRescueExperiment | CalMorphExperiment | MicroarrayExpressionExperiment | MicroarrayExpressionScoreExperiment | MicroarrayExpressionRatioExperiment | InductionTimecourseExpressionExperiment | RNASeqExpressionExperiment | PseudobulkExpressionExperiment | VisualScoreExperiment | MetaboliteExperiment | ProteinAbundanceExperiment | ProteinFoldChangeExperiment | EnvironmentResponseExperiment | StrainEnvironmentResponseExperiment | SegregantGrowthExperiment | SegregantPerturbationExperiment | ProductTiterExperiment | ProteinTurnoverExperiment | ProteinSynthesisRateExperiment | FluxExperiment | PromoterActivityExperiment | PooledSortSeqExpressionExperiment | BiosensorFluorescenceExperiment | GradedGeneEssentialityExperiment | PhosphositeIntensityExperiment | MorphologyExperiment | MrnaNumberFractionExperiment | FitnessContrastExperiment | SegregantExpressionExperiment | LifeCycleExperiment | PseudobulkUmiCountExperiment | GenomeArchitectureExperiment | ExperimentReference | FitnessExperimentReference | GeneInteractionExperimentReference | GeneEssentialityExperimentReference | SyntheticLethalityExperimentReference | SyntheticRescueExperimentReference | CalMorphExperimentReference | MicroarrayExpressionExperimentReference | MicroarrayExpressionScoreExperimentReference | MicroarrayExpressionRatioExperimentReference | InductionTimecourseExpressionExperimentReference | RNASeqExpressionExperimentReference | PseudobulkExpressionExperimentReference | VisualScoreExperimentReference | MetaboliteExperimentReference | ProteinAbundanceExperimentReference | ProteinFoldChangeExperimentReference | EnvironmentResponseExperimentReference | StrainEnvironmentResponseExperimentReference | SegregantGrowthExperimentReference | SegregantPerturbationExperimentReference | ProductTiterExperimentReference | ProteinTurnoverExperimentReference | ProteinSynthesisRateExperimentReference | FluxExperimentReference | PromoterActivityExperimentReference | PooledSortSeqExpressionExperimentReference | BiosensorFluorescenceExperimentReference | GradedGeneEssentialityExperimentReference | PhosphositeIntensityExperimentReference | MorphologyExperimentReference | MrnaNumberFractionExperimentReference | FitnessContrastExperimentReference | SegregantExpressionExperimentReference | LifeCycleExperimentReference | PseudobulkUmiCountExperimentReference | GenomeArchitectureExperimentReference]][source]: Flatten nested lists of experiments into a single list.

property phenotype_info: list[type[Phenotype]]: Return the phenotype classes present in the aggregated store (cached).

close_lmdb() → None[source]: Close the LMDB environment if open and reset the handle.

process(input_path: str, output_path: str, key_groups: dict[str, list[bytes]] | None = None) → None[source]

Read pairs from input LMDB, group by aggregate key, and write groups out.

Two streaming passes, never the whole dataset in memory. The previous implementation held every record as pydantic objects inside aggregated_data – its own comment conceded “can potentially be huge”, and at the 025 build’s scale that is hundreds of GB.

Pass 1 groups input KEYS by aggregate_key_raw computed off the raw JSON. Pass 2 assembles each output group as a JSON array by joining the STORED record bytes – they were written by upstream model_dump calls, so a reconstruct-and-redump would be an identity round trip. Group order is first-occurrence (dict insertion order), matching the old output.

key_groups, when given, IS pass 1’s result and pass 1 is not run: the grouping key to the input keys, groups in the order a cursor walk first meets them and keys in cursor order within each group (cursor order is lexicographic in the key bytes). Neo4jCellDataset.process builds it from aggregate_key_raw while the raw stage writes the input (RawStageGrouping). Both reads open with readahead: pass 1 is a cursor walk, and pass 2’s gets run in cursor order too, which on a cold store read 2,630 MB/s with readahead against 213 MB/s without (experiments/tcdb-002-build-speed/results/query_build_cost_probe.csv).