Cartridge genotype plane: persist population genotype priors on a cartridge

site_docs/guides/genotype.md

@@ -316,6 +316,99 @@ excludes an allele), and unspecified genes fall back to uniform.
     explicit opt-in that reuses the cartridge's recombination `allele_usage` as a
     frequency proxy — convenient but biologically approximate.
 
+## Population genotype models on a cartridge
+
+The `allele_frequencies` / `haplotype_deletion_prob` you pass to
+`Genotype.sample` can instead be **authored once on the cartridge** as a
+*population genotype model* — a donor-population germline prior. It is a distinct
+plane from `reference_models.allele_usage`: `allele_usage` weights how often each
+allele is *expressed* during recombination, whereas a genotype prior describes
+which alleles a *donor population carries* (frequencies, gene-deletion rates, and
+population novel alleles). It lives on `DataConfig.genotype_priors`.
+
+### Authoring and attaching a model
+
+```python
+from GenAIRR.genotype_priors import PopulationGenotypeModel, PopulationNovelAllele
+
+model = PopulationGenotypeModel(
+    model_id="IGH-toy-1", source="hand-authored",     # identity is required
+    allele_frequencies={"V": {"IGHVF1-G1": {"IGHVF1-G1*01": 3.0, "IGHVF1-G1*02": 1.0}}},
+    haplotype_deletion_prob={"V": {"IGHVF1-G1": 0.1}},
+)
+# Attach via the cartridge builder (validated against the chain type + catalogue):
+#   cfg = builder.set_genotype_priors(model).build()
+```
+
+`set_genotype_priors` validates the model against the cartridge: unknown
+genes/alleles raise, and any population novel allele goes through the same
+functional validation as `add_novel_allele` (conserved anchor, stop-free frame,
+base-allele match). A non-`None` plane becomes part of cartridge identity (it
+folds into `compute_checksum()` and the manifest).
+
+### Estimating a model from observed genotypes
+
+Given a list of observed `Genotype` objects (e.g. one per donor), estimate a
+prior directly:
+
+```python
+model = PopulationGenotypeModel.from_genotypes(
+    [g_donor1, g_donor2, g_donor3],
+    cfg=cfg, model_id="cohort-est", source="my-cohort",
+)
+# or, attaching to a builder in one chained step:
+#   builder.estimate_genotype_priors([g_donor1, g_donor2, g_donor3], source="my-cohort")
+```
+
+Estimator conventions: allele frequencies are counted **per carried chromosome**
+(homozygous contributes 2, hemizygous 1, deleted 0); gene deletion probability is
+`deleted_haplotypes / (2 × n_subjects)`. `pseudocount` smooths the per-gene
+catalogue-allele counts only (deletion gets none). Genotypes carrying a
+duplicated gene are rejected — the plane is deletion-only.
+
+### Drawing from the cartridge plane
+
+When a cartridge carries a plane, `Genotype.sample(cfg)` **auto-uses it** and
+records where every input came from:
+
+```python
+g = Genotype.sample(cfg, seed=7)        # plane supplies freqs / deletion / weights
+print(g.prior_provenance)
+# {'allele_frequencies': 'cartridge', 'haplotype_deletion_prob': 'cartridge',
+#  'chromosome_weights': 'cartridge', 'novel_alleles': 'none',  # 'cartridge' if the model has novels
+#  'model_id': 'IGH-toy-1', 'model_checksum': '…'}
+print(g.to_metadata())                  # subject_id + provenance + source/effective refdata hashes
+```
+
+Pass `use_cartridge_priors=False` for a clean uniform, catalogue-only draw (all
+plane consumption — including novels — disabled). Each input is sourced
+**independently**, so you can mix explicit and cartridge values:
+
+```python
+g = Genotype.sample(
+    cfg, seed=7,
+    allele_frequencies={"V": {"IGHVF1-G1": {"IGHVF1-G1*01": 1.0}}},  # explicit
+    # haplotype_deletion_prob and chromosome_weights left to the plane
+)
+print(g.prior_provenance["allele_frequencies"])      # 'explicit'
+print(g.prior_provenance["haplotype_deletion_prob"]) # 'cartridge'
+print(g.prior_provenance["chromosome_weights"])      # 'cartridge'
+```
+
+**Population novel alleles** on the plane are *candidate* alleles for the draw: a
+novel that gets sampled is carried (and flows into `v_call` / reads / truth like
+any allele); a novel that isn't drawn never pollutes the output reference. By
+default (`include_cartridge_novel_alleles="auto"`) novels are injected only when
+the allele frequencies are cartridge- or uniform-sourced — supplying an
+**explicit** frequency table keeps it explicit. Pass
+`include_cartridge_novel_alleles=True` to inject them anyway, or `False` to never.
+
+### Auditing the plane
+
+`cfg.cartridge_manifest()["models"]["genotype_priors"]` reports availability,
+`model_id` / `source` / `version`, the plane's `model_checksum`, per-segment gene
+counts, the novel-allele count, and `source_field` (`"DataConfig.genotype_priors"`).
+
 ## Novel / private alleles
 
 Individuals carry germline alleles that aren't in any reference — *private* or
@@ -527,8 +620,6 @@ The genotype foundation is deliberately scoped. Deferred to later work:
   (`with_genotype` is single-subject; `result.genotypes` is a one-element list).
 - **External loaders** — importing genotypes from VDJbase / TIgGER / IgDiscover /
   partis output.
-- **Cartridge genotype plane** — persisting a population genotype model in a
-  cartridge.
 - **Same-haplotype receptor revision** — `receptor_revision` with a genotype is
   rejected for now.
 

src/GenAIRR/cartridge_builder.py

@@ -90,6 +90,7 @@ class _BuilderDAllele(DAllele):
 from .dataconfig.config_info import ConfigInfo
 from .dataconfig.data_config import DataConfig
 from .dataconfig.enums import ChainType, Species
+from .genotype_priors import PopulationGenotypeModel
 from .reference_models import (
     AlleleUsageSpec,
     EmpiricalDistributionSpec,
@@ -346,6 +347,7 @@ def __init__(self, chain_type: ChainType) -> None:
         self._metadata: Optional[ConfigInfo] = None
         self._reference_rules: Optional[ReferenceRulesSpec] = None
         self._reference_models: Optional[ReferenceEmpiricalModels] = None
+        self._genotype_priors: Optional[PopulationGenotypeModel] = None
         self._report = CartridgeBuildReport()
 
     # ──────────────────────────────────────────────────────────
@@ -714,6 +716,81 @@ def with_models(
     # Estimators
     # ──────────────────────────────────────────────────────────
 
+    def _build_working_cfg(self):
+        """A lightweight DataConfig carrying only the parsed catalogues — used by
+        genotype-prior validation/estimation to resolve gene/allele names and
+        construct throwaway Genotypes for novel functional validation."""
+        return DataConfig(
+            name=self._name or "WORKING",
+            metadata=self._metadata,
+            v_alleles=self._v_alleles,
+            d_alleles=self._d_alleles or None,
+            j_alleles=self._j_alleles,
+            c_alleles=None,
+        )
+
+    def set_genotype_priors(
+        self, model: PopulationGenotypeModel
+    ) -> "ReferenceCartridgeBuilder":
+        """Attach a hand-authored population genotype prior, validated against
+        this cartridge's chain type and catalogue. Chainable."""
+        from GenAIRR.genotype import Genotype
+
+        if not isinstance(model, PopulationGenotypeModel):
+            raise TypeError(
+                f"set_genotype_priors expects a PopulationGenotypeModel, "
+                f"got {type(model).__name__}")
+        chain_label = "vdj" if self._chain_type.has_d else "vj"
+        model.validate(chain_type=chain_label)  # catalogue-free
+
+        # Catalogue-aware checks against the builder's pools.
+        cfg = self._build_working_cfg()
+        by_seg = {"V": cfg.v_alleles, "D": cfg.d_alleles or {}, "J": cfg.j_alleles}
+        for table_name, table in (("allele_frequencies", model.allele_frequencies),
+                                  ("haplotype_deletion_prob", model.haplotype_deletion_prob)):
+            for seg, genes in (table or {}).items():
+                catalogue = by_seg[seg]
+                for gene, payload in genes.items():
+                    if gene not in catalogue:
+                        raise ValueError(
+                            f"genotype_priors.{table_name}: {seg} gene {gene!r} is not "
+                            f"in the cartridge")
+                    if table_name == "allele_frequencies":
+                        names = {a.name for a in catalogue[gene]}
+                        for allele in payload:
+                            if allele not in names:
+                                raise ValueError(
+                                    f"genotype_priors.allele_frequencies: {allele!r} is "
+                                    f"not a known allele of {gene!r}")
+        # Novels: reuse Genotype.add_novel_allele for full functional validation.
+        helper = Genotype.from_dataconfig(cfg)
+        for nv in model.novel_alleles:
+            helper.add_novel_allele(
+                nv.name, base=nv.base_allele, sequence=nv.sequence.upper(),
+                segment=nv.segment, allow_nonfunctional=nv.allow_nonfunctional)
+
+        self._genotype_priors = model
+        self._report.stages.append({
+            "stage": "set_genotype_priors",
+            "inputs": {"model_id": model.model_id, "source": model.source,
+                       "chain_type_label": chain_label},
+            "inferred": {"model_checksum": model.content_checksum(),
+                         "novel_allele_count": len(model.novel_alleles)},
+            "warnings": [],
+        })
+        return self
+
+    def estimate_genotype_priors(
+        self, genotypes, **kwargs
+    ) -> "ReferenceCartridgeBuilder":
+        """Estimate a population genotype prior from observed ``Genotype`` objects
+        and attach it (chainable). Thin wrapper over
+        :meth:`PopulationGenotypeModel.from_genotypes` followed by
+        :meth:`set_genotype_priors`."""
+        model = PopulationGenotypeModel.from_genotypes(
+            genotypes, cfg=self._build_working_cfg(), **kwargs)
+        return self.set_genotype_priors(model)
+
     def estimate_allele_usage(
         self,
         rearrangements: Any,
@@ -2276,6 +2353,7 @@ def build(self) -> DataConfig:
             c_alleles=None,  # v1 boundary
             reference_rules=self._reference_rules,
             reference_models=self._reference_models,
+            genotype_priors=self._genotype_priors,
         )
         # Pull a manifest snapshot + checksum. The manifest call
         # runs before verify_integrity so the report carries the

src/GenAIRR/dataconfig/data_config.py

@@ -15,6 +15,7 @@
 from GenAIRR.alleles.allele import Allele
 from GenAIRR.reference_models import ReferenceEmpiricalModels
 from GenAIRR.reference_rules import ReferenceRulesSpec
+from GenAIRR.genotype_priors import PopulationGenotypeModel
 
 
 DEFAULT_P_NUCLEOTIDE_LENGTH_PROBS = {0: 0.50, 1: 0.25, 2: 0.15, 3: 0.07, 4: 0.03}
@@ -136,6 +137,84 @@ def _allele_usage_manifest_block(cfg):
     }
 
 
+def _genotype_priors_manifest_block(cfg):
+    """Build the ``models.genotype_priors`` manifest block (Slice — Cartridge
+    genotype plane). Audit-sized: counts and identity, never the full tables.
+    Reads the top-level ``DataConfig.genotype_priors`` plane (independent of
+    ``reference_models``)."""
+    def _empty(available, valid, error=None):
+        b = {
+            "available": available,
+            "valid": valid,
+            "model_id": None,
+            "source": None,
+            "version": None,
+            "model_checksum": None,
+            "segments_with_frequencies": [],
+            "freq_gene_counts": {"V": 0, "D": 0, "J": 0},
+            "deletion_gene_counts": {"V": 0, "D": 0, "J": 0},
+            "novel_allele_count": 0,
+            "chromosome_weights": None,
+            "source_field": "DataConfig.genotype_priors",
+        }
+        if error is not None:
+            b["validation_error"] = error
+        return b
+
+    model = getattr(cfg, "genotype_priors", None)
+    if model is None:
+        return _empty(False, None)
+    if not isinstance(model, PopulationGenotypeModel):
+        # Field is typed Optional[PopulationGenotypeModel] but Python won't enforce
+        # it; a garbage value must not crash the manifest (called from build()).
+        return _empty(True, False,
+                      f"genotype_priors is not a PopulationGenotypeModel "
+                      f"(got {type(model).__name__})")
+
+    # A plane may have been attached directly (bypassing builder validation).
+    # Report validity rather than leaking non-JSON-clean numerics (e.g. NaN
+    # chromosome weights) into the manifest.
+    try:
+        model.validate(
+            chain_type=getattr(getattr(cfg, "metadata", None), "chain_type", None))
+        valid, validation_error = True, None
+    except ValueError as exc:
+        valid, validation_error = False, str(exc)
+
+    # content_checksum / float() can themselves raise on a malformed model; never
+    # let that crash the manifest — report None and the validity flag instead.
+    try:
+        checksum = model.content_checksum()
+    except Exception:
+        checksum = None
+    cw = None
+    if valid:
+        try:
+            cw = [float(model.chromosome_weights[0]), float(model.chromosome_weights[1])]
+        except Exception:
+            cw = None
+    freq = model.allele_frequencies if isinstance(model.allele_frequencies, dict) else {}
+    dele = model.haplotype_deletion_prob if isinstance(model.haplotype_deletion_prob, dict) else {}
+    novels = model.novel_alleles if isinstance(model.novel_alleles, (list, tuple)) else []
+    block = {
+        "available": True,
+        "valid": valid,
+        "model_id": (model.model_id or None) if isinstance(model.model_id, str) else None,
+        "source": (model.source or None) if isinstance(model.source, str) else None,
+        "version": (model.version or None) if isinstance(model.version, str) else None,
+        "model_checksum": checksum,
+        "segments_with_frequencies": [s for s in ("V", "D", "J") if freq.get(s)],
+        "freq_gene_counts": {s: len(freq.get(s, {})) for s in ("V", "D", "J")},
+        "deletion_gene_counts": {s: len(dele.get(s, {})) for s in ("V", "D", "J")},
+        "novel_allele_count": len(novels),
+        "chromosome_weights": cw,
+        "source_field": "DataConfig.genotype_priors",
+    }
+    if validation_error is not None:
+        block["validation_error"] = validation_error
+    return block
+
+
 def _np_length_models_manifest_block(cfg):
     """Build the ``models.np_length_models`` manifest block
     per the NP Length Distribution Estimation v1 audit
@@ -301,6 +380,15 @@ class DataConfig:
     # ``reference_rules``.
     reference_models: Optional[ReferenceEmpiricalModels] = None
 
+    # Donor-population germline prior plane (Slice — Cartridge genotype plane).
+    # ``None`` means no population prior; ``Genotype.sample(cfg)`` then falls
+    # back to a uniform synthetic prior. A non-``None`` plane is cartridge
+    # identity (folds into compute_checksum). See
+    # ``site_docs/guides/genotype.md`` ("Population genotype models on a
+    # cartridge"). Same soft-transition checksum policy as ``reference_rules`` /
+    # ``reference_models``.
+    genotype_priors: Optional[PopulationGenotypeModel] = None
+
     def __getattr__(self, name):
         # Backward-compat shim for pickled DataConfigs missing post-v1
         # fields. Note: schema_version / schema_sha256 fall through to
@@ -319,6 +407,8 @@ def __getattr__(self, name):
             return None
         if name == 'reference_models':
             return None
+        if name == 'genotype_priors':
+            return None
         raise AttributeError(f"'{type(self).__name__}' object has no attribute '{name}'")
 
     def compute_checksum(self) -> str:
@@ -355,6 +445,8 @@ def compute_checksum(self) -> str:
         pop_rules = 'reference_rules' in self.__dict__ and rr_value is None
         rm_value = self.__dict__.get('reference_models')
         pop_models = 'reference_models' in self.__dict__ and rm_value is None
+        gp_value = self.__dict__.get('genotype_priors')
+        pop_priors = 'genotype_priors' in self.__dict__ and gp_value is None
 
         self.schema_sha256 = ""
         if had_report:
@@ -363,6 +455,8 @@ def compute_checksum(self) -> str:
             del self.__dict__['reference_rules']
         if pop_models:
             del self.__dict__['reference_models']
+        if pop_priors:
+            del self.__dict__['genotype_priors']
         try:
             blob = pickle.dumps(self, protocol=4)
             return hashlib.sha256(blob).hexdigest()
@@ -374,6 +468,8 @@ def compute_checksum(self) -> str:
                 self.__dict__['reference_rules'] = rr_value
             if pop_models:
                 self.__dict__['reference_models'] = rm_value
+            if pop_priors:
+                self.__dict__['genotype_priors'] = gp_value
 
     def verify_integrity(self) -> None:
         """Validate schema_version and schema_sha256.
@@ -917,6 +1013,10 @@ def cartridge_manifest(
             # ``np_length_keys`` / ``legacy_np_lengths_present``
             # entries above.
             "np_length_models": _np_length_models_manifest_block(self),
+            # Donor-population germline prior plane (Slice — Cartridge genotype
+            # plane). Read from the top-level ``DataConfig.genotype_priors``
+            # field (NOT ``reference_models``); ``source_field`` records that.
+            "genotype_priors": _genotype_priors_manifest_block(self),
         }
 
         # Bridge once (or accept the provided refdata) to read the