SeqMut.trace_lineage

static SeqMut.trace_lineage(lineage, candidate_id)

Reconstruct the chain of lineage records that leads to one candidate.

A design campaign is a chain: a candidate of one round becomes the parent of the next, and each record names its parent by content hash. This walks those links backwards from candidate_id to the root the export still contains, so the full parent-to-candidate path is recovered from the records alone – including after a JSON round trip, and including records that came from different runs or different classes.

Parameters:

• lineage (list of dict) – Lineage records collected over one or more design rounds, as produced by SeqMut.combine() or SeqOpt.run() with lineage opted in (read from SeqMut.lineage_ / SeqOpt.lineage_, or reloaded from JSON). Records of unrelated candidates may be present; only the traced chain is returned.

• candidate_id (str) – Identifier of the candidate to trace, as found in the candidate_id column of the returned table or in the candidate_id field of a record.

Returns:

list_lineage – The records from the root ancestor to candidate_id, root first, so concatenating their mutations replays the design from the root sequence to the candidate. The chain holds one record when the candidate descends directly from its source.

Return type:

list of dict

Raises:

• ValueError – If lineage is not a non-empty list of lineage records, or if candidate_id is not the identifier of one of them.

• RuntimeError – If the records are cyclic, so a candidate would be its own ancestor.

See also

• SeqMut.combine(): whose lineage parameter produces the records.

• SeqOpt.trace_lineage(): the same tracer on the optimizer.

Added in version 1.2.0.

Examples

:meth:SeqMut.trace_lineage reads the candidate-lineage record that :meth:SeqMut.combine attaches when lineage is opted in. Each candidate gets a plain, JSON-serializable record: a content-hash candidate_id over its source sequence and its ordered mutations, its parent_id, the parent-relative mutations, the generating method, the objective_values, the effective seed and a digest of the applied design limits. The record is opt-in and purely additive: the default table is unchanged.

import json

import pandas as pd
import aaanalysis as aa
aa.options["verbose"] = False

# Data + CPP features the delta-CPP engine scores the candidates on
df_seq = aa.load_dataset(name="DOM_GSEC", n=10)
labels = df_seq["label"].to_list()
sf = aa.SequenceFeature()
df_parts = sf.get_df_parts(df_seq=df_seq)
df_scales = aa.load_scales()
cpp = aa.CPP(df_parts=df_parts, split_kws=sf.get_split_kws(), df_scales=df_scales)
df_feat = cpp.run(labels=labels, n_filter=25)

entry = df_seq["entry"].iloc[0]
seq_wt = df_seq["sequence"].iloc[0]
ts = int(df_seq.set_index("entry").loc[entry, "tmd_start"])

# Round 1: two candidates of the wild-type, with the lineage record switched on
variants = pd.DataFrame({"entry": [entry] * 3,
                         "variant": ["double", "double", "single"],
                         "pos": [ts, ts + 1, ts + 3],
                         "to_aa": ["W", "P", "K"]})
seqm = aa.SeqMut()
df_variant = seqm.combine(df_seq=df_seq, variants=variants, df_feat=df_feat, lineage=True)
aa.display_df(df_variant[["variant", "n_mut", "delta_cpp", "candidate_id"]],
              n_rows=10, show_shape=True)
print(json.dumps(seqm.lineage_[0], indent=1))

DataFrame shape: (2, 4)

	variant	n_mut	delta_cpp	candidate_id
1	G40K	1	0.000000	sha256:8e74a619...c3215b217ecd139
2	L37W+Q38P	2	0.020000	sha256:06f78686...7d77de73cad02e6

{
 "candidate_id": "sha256:8e74a61925dc714836c3f0472ef91e2a123fd75637e205799c3215b217ecd139",
 "parent_id": "sha256:6292830e38478ba10f73868d050a2d353cb2011b5e40b9022a80db325eb43f23",
 "source_seq_id": "sha256:6292830e38478ba10f73868d050a2d353cb2011b5e40b9022a80db325eb43f23",
 "mutations": [
  {
   "pos": 40,
   "from_aa": "G",
   "to_aa": "K"
  }
 ],
 "method": "SeqMut.combine",
 "objective_values": {
  "delta_cpp": 0.0,
  "shift_score": 0.0
 },
 "seed": null,
 "constraints_digest": null
}

Round 2: a candidate becomes the parent. Passing the record of a round-1 candidate instead of True declares that parent, so the new records carry its candidate_id as their parent_id. That is how the rounds of a multi-generation design chain up. trace_lineage then walks the chain back: pass the collected records as lineage and the candidate to trace as candidate_id.

# The round-1 double becomes the parent of round 2
record_parent = [r for r in seqm.lineage_ if len(r["mutations"]) == 2][0]
seq_parent = df_variant[df_variant["n_mut"] == 2]["sequence_mut"].iloc[0]
df_seq_parent = pd.DataFrame({"entry": [entry], "sequence": [seq_parent],
                              "tmd_start": [ts], "tmd_stop": [int(df_seq["tmd_stop"].iloc[0])]})

variants_2 = pd.DataFrame({"entry": [entry] * 2, "variant": ["r2a", "r2b"],
                           "pos": [ts + 5, ts + 6], "to_aa": ["C", "Y"]})
seqm = aa.SeqMut()
df_variant_2 = seqm.combine(df_seq=df_seq_parent, variants=variants_2, df_feat=df_feat,
                            lineage=record_parent)
lineage = [record_parent] + seqm.lineage_

list_lineage = aa.SeqMut.trace_lineage(lineage=lineage,
                                       candidate_id=seqm.lineage_[0]["candidate_id"])
df_chain = pd.DataFrame([{"step": i, "method": r["method"],
                          "mutations": "+".join(f"{m['from_aa']}{m['pos']}{m['to_aa']}"
                                                for m in r["mutations"]),
                          "candidate_id": r["candidate_id"][:19] + "..."}
                         for i, r in enumerate(list_lineage)])
aa.display_df(df_chain, n_rows=10, show_shape=True)

DataFrame shape: (2, 4)

	step	method	mutations	candidate_id
1	0	SeqMut.combine	L37W+Q38P	sha256:06f78686198d...
2	1	SeqMut.combine	L42C	sha256:f7e250a9c157...

The chain is root first, so concatenating the mutations of its records replays the design from the wild-type to the candidate. Because every identifier is a content hash, the records survive a JSON round trip unchanged, and two runs that build the same mutant from the same source assign the same candidate_id, which is what makes duplicates detectable across runs.

# Replay the path: apply every mutation of the chain to the wild-type
sequence = seq_wt
for record in list_lineage:
    for m in record["mutations"]:
        sequence = sequence[:m["pos"] - 1] + m["to_aa"] + sequence[m["pos"]:]
print("replay reproduces the candidate:",
      sequence == df_variant_2["sequence_mut"].iloc[0])

# Loss-free JSON round trip, then trace again from the reloaded records
reloaded = json.loads(json.dumps(lineage))
print("round trip is loss-free:", reloaded == lineage)
print("same chain after reload:",
      aa.SeqMut.trace_lineage(lineage=reloaded,
                              candidate_id=seqm.lineage_[0]["candidate_id"]) == list_lineage)

# The same mutant, reached in a second run, gets the same identifier
seqm = aa.SeqMut()
seqm.combine(df_seq=df_seq, df_feat=df_feat, lineage=True,
             variants=pd.DataFrame({"entry": [entry] * 2, "variant": ["again"] * 2,
                                    "pos": [ts + 1, ts], "to_aa": ["P", "W"]}))
print("duplicate detected:",
      seqm.lineage_[0]["candidate_id"] == record_parent["candidate_id"])

replay reproduces the candidate: True
round trip is loss-free: True
same chain after reload: True
duplicate detected: True