1+
2+ from typing import Union
3+
4+ import sqlalchemy as sa
5+ from sqlalchemy .orm import Session
6+
7+ # SQLAlchemy needs access to all models to properly map relationships.
8+ from mavedb .models import *
9+
10+ from mavedb .db .session import SessionLocal
11+ from mavedb .models .score_set import ScoreSet
12+ from mavedb .models .score_calibration import ScoreCalibration as ScoreCalibrationDBModel
13+ from mavedb .models .publication_identifier import PublicationIdentifier
14+ from mavedb .view_models .score_range import (
15+ ScoreRangeCreate ,
16+ ScoreSetRangesAdminCreate ,
17+ ZeibergCalibrationScoreRangesAdminCreate ,
18+ ScottScoreRangesAdminCreate ,
19+ InvestigatorScoreRangesAdminCreate ,
20+ IGVFCodingVariantFocusGroupControlScoreRangesAdminCreate ,
21+ IGVFCodingVariantFocusGroupMissenseScoreRangesAdminCreate ,
22+ )
23+
24+ score_range_kinds : dict [
25+ str ,
26+ Union [
27+ type [ZeibergCalibrationScoreRangesAdminCreate ],
28+ type [ScottScoreRangesAdminCreate ],
29+ type [InvestigatorScoreRangesAdminCreate ],
30+ type [IGVFCodingVariantFocusGroupControlScoreRangesAdminCreate ],
31+ type [IGVFCodingVariantFocusGroupMissenseScoreRangesAdminCreate ],
32+ ],
33+ ] = {
34+ "zeiberg_calibration" : ZeibergCalibrationScoreRangesAdminCreate ,
35+ "scott_calibration" : ScottScoreRangesAdminCreate ,
36+ "investigator_provided" : InvestigatorScoreRangesAdminCreate ,
37+ "cvfg_all_variants" : IGVFCodingVariantFocusGroupControlScoreRangesAdminCreate ,
38+ "cvfg_missense_variants" : IGVFCodingVariantFocusGroupMissenseScoreRangesAdminCreate ,
39+ }
40+
41+ EVIDENCE_STRENGTH_FROM_POINTS = {
42+ 8 : "Very Strong" ,
43+ 4 : "Strong" ,
44+ 3 : "Moderate+" ,
45+ 2 : "Moderate" ,
46+ 1 : "Supporting" ,
47+ }
48+
49+
50+ def do_migration (session : Session ) -> None :
51+ score_sets_with_ranges = (
52+ session .execute (sa .select (ScoreSet ).where (ScoreSet .score_ranges .isnot (None ))).scalars ().all ()
53+ )
54+
55+ for score_set in score_sets_with_ranges :
56+ if not score_set .score_ranges :
57+ continue
58+
59+ score_set_ranges = ScoreSetRangesAdminCreate .model_validate (score_set .score_ranges )
60+
61+ for field in score_set_ranges .model_fields_set :
62+ if field == "record_type" :
63+ continue
64+
65+ ranges = getattr (score_set_ranges , field )
66+ if not ranges :
67+ continue
68+
69+ range_model = score_range_kinds .get (field )
70+ inferred_ranges = range_model .model_validate (ranges )
71+
72+ model_thresholds = []
73+ for range in inferred_ranges .ranges :
74+ model_thresholds .append (ScoreRangeCreate .model_validate (range .__dict__ ).model_dump ())
75+
76+ # We should migrate the zeiberg evidence classifications to be explicitly part of the calibration ranges.
77+ if field == "zeiberg_calibration" :
78+ for inferred_range , model_range in zip (
79+ inferred_ranges .ranges ,
80+ model_thresholds ,
81+ ):
82+ model_range ["label" ] = f"PS3 { EVIDENCE_STRENGTH_FROM_POINTS .get (inferred_range .evidence_strength , 'Unknown' )} if inferred_range .evidence_strength > 0 else f"BS3 { EVIDENCE_STRENGTH_FROM_POINTS .get (abs (inferred_range .evidence_strength ), 'Unknown' )}
83+ model_range ["acmg_classification" ] = {"points" : inferred_range .evidence_strength }
84+
85+ # Reliant on existing behavior that these sources have been created already.
86+ # If not present, no sources will be associated.
87+ if "odds_path_source" in inferred_ranges .model_fields_set and inferred_ranges .odds_path_source :
88+ oddspaths_sources = (
89+ session .execute (
90+ sa .select (PublicationIdentifier ).where (
91+ PublicationIdentifier .identifier .in_ (
92+ [src .identifier for src in (inferred_ranges .odds_path_source or [])]
93+ )
94+ )
95+ )
96+ .scalars ()
97+ .all ()
98+ )
99+ else :
100+ oddspaths_sources = []
101+
102+ if "source" in inferred_ranges .model_fields_set and inferred_ranges .source :
103+ range_sources = (
104+ session .execute (
105+ sa .select (PublicationIdentifier ).where (
106+ PublicationIdentifier .identifier .in_ (
107+ [src .identifier for src in (inferred_ranges .source or [])]
108+ )
109+ )
110+ )
111+ .scalars ()
112+ .all ()
113+ )
114+ else :
115+ range_sources = []
116+
117+ sources = set ()
118+ for publication in oddspaths_sources :
119+ setattr (publication , "relation" , "method" )
120+ sources .add (publication )
121+ for publication in range_sources :
122+ setattr (publication , "relation" , "threshold" )
123+ sources .add (publication )
124+
125+ score_calibration = ScoreCalibrationDBModel (
126+ score_set_id = score_set .id ,
127+ title = inferred_ranges .title ,
128+ research_use_only = inferred_ranges .research_use_only ,
129+ primary = inferred_ranges .primary ,
130+ private = False , # All migrated calibrations are public.
131+ investigator_provided = True if field == "investigator_provided" else False ,
132+ baseline_score = inferred_ranges .baseline_score
133+ if "baseline_score" in inferred_ranges .model_fields_set
134+ else None ,
135+ baseline_score_description = inferred_ranges .baseline_score_description
136+ if "baseline_score_description" in inferred_ranges .model_fields_set
137+ else None ,
138+ functional_ranges = None if not model_thresholds else model_thresholds ,
139+ calibration_metadata = None ,
140+ publication_identifiers = sources ,
141+ # If investigator_provided, set to creator of score set, else set to default system user (1).
142+ created_by_id = score_set .created_by_id if field == "investigator_provided" else 1 ,
143+ modified_by_id = score_set .created_by_id if field == "investigator_provided" else 1 ,
144+ )
145+ session .add (score_calibration )
146+
147+
148+ if __name__ == "__main__" :
149+ db = SessionLocal ()
150+ db .current_user = None # type: ignore
151+
152+ do_migration (db )
153+
154+ db .commit ()
155+ db .close ()
0 commit comments