Add constant "hla_compatibility_score"

tests/database/services/test_file_upload.py

@@ -14,6 +14,7 @@
 from txmatching.database.sql_alchemy_schema import (
     AppUserModel, ConfigModel, DonorModel, HLAAntibodyRawModel,
     PairingResultModel, RecipientAcceptableBloodModel, RecipientModel)
+from txmatching.scorers.scorer_constants import HLA_SCORE
 from txmatching.scorers.split_hla_additive_scorer import SplitScorer
 from txmatching.solve_service.solve_from_configuration import \
     solve_from_configuration
@@ -148,7 +149,7 @@ def test_saving_patients_from_obfuscated_excel(self):
             for recipient_enum, recipient, expected_score in zip(
                     recipient_enums, txm_event.active_and_valid_recipients_dict.values(), expected_score_row):
                 if expected_score >= 0:
-                    calculated_score = calculated_scores[(donor_enum, recipient_enum)]["hla_compatibility_score"]
+                    calculated_score = calculated_scores[(donor_enum, recipient_enum)][HLA_SCORE]
                     self.assertEqual(expected_score, calculated_score,
                                      f'Not true for expected {expected_score} vs real {calculated_score} '
                                      f'{[code.raw_code for code in donor.parameters.hla_typing.hla_types_raw_list]} and '

tests/solvers/all_solutions_solver/test_solver_from_compatibility_graph.py

@@ -10,6 +10,7 @@
 from txmatching.patients.patient import Donor, Recipient
 from txmatching.patients.patient_parameters import PatientParameters
 from txmatching.scorers.compatibility_graph import CompatibilityGraph
+from txmatching.scorers.scorer_constants import HLA_SCORE
 from txmatching.solvers.all_solutions_solver.compatibility_graph_solver import \
     find_optimal_paths
 from txmatching.solvers.donor_recipient_pair_idx_only import \
@@ -49,7 +50,7 @@ def test_solve(self):
         all_scores = []
         for solution in all_solutions:
             solution_score = sum(
-                [scorer.score_transplant_ij(pair.donor_idx, pair.recipient_idx)['hla_compatibility_score']
+                [scorer.score_transplant_ij(pair.donor_idx, pair.recipient_idx)[HLA_SCORE]
                  for pair in solution])
             all_scores.append(solution_score)
         self.assertEqual(len(all_solutions[0]), 9)
@@ -90,8 +91,8 @@ def test_handling_correctly_multiple_donors_with_the_same_recipient(self):
 
         original_donor_idx_to_recipient_idx = {0: 0, 1: 0, 2: 1, 3: 2}
 
-        compatibility_graph_test = {(0, 1): {"hla_compatibility_score": 11}, (1, 2): {"hla_compatibility_score": 10},
-                                    (2, 0): {"hla_compatibility_score": 10}, (3, 0): {"hla_compatibility_score": 10}}
+        compatibility_graph_test = {(0, 1): {HLA_SCORE: 11}, (1, 2): {HLA_SCORE: 10},
+                                    (2, 0): {HLA_SCORE: 10}, (3, 0): {HLA_SCORE: 10}}
 
         solutions = list(find_optimal_paths(
             compatibility_graph_test,
@@ -127,7 +128,7 @@ def test_handling_correctly_multiple_donors_with_the_same_recipient_at_the_end_o
 
         original_donor_idx_to_recipient_idx = {0: 0, 1: 0, 2: -1}
 
-        compatibility_graph_test = {(2, 0): {"hla_compatibility_score": 15}}
+        compatibility_graph_test = {(2, 0): {HLA_SCORE: 15}}
 
         solutions = list(find_optimal_paths(
             compatibility_graph_test,
@@ -155,7 +156,7 @@ def test_works_with_one_cycle_only(self):
 
         original_donor_idx_to_recipient_idx = {0: -1, 1: 0}
 
-        compatibility_graph_test = {(0, 0): {'hla_compatibility_score': 0}}
+        compatibility_graph_test = {(0, 0): {HLA_SCORE: 0}}
 
         solutions = list(find_optimal_paths(
             compatibility_graph_test,
@@ -210,7 +211,7 @@ def _get_compatibility_graph_from_score_matrix(score_matrix: np.array) -> Compat
     for row_enum, row in enumerate(score_matrix):
         for score_enum, score in enumerate(row):
             if score >= 0:
-                compatibility_graph[(row_enum, score_enum)] = {"hla_compatibility_score": score}
+                compatibility_graph[(row_enum, score_enum)] = {HLA_SCORE: score}
     return compatibility_graph
 
 

tests/web/test_optimizer_api.py

@@ -10,6 +10,7 @@
     get_txm_event_complete, get_txm_event_db_id_by_name
 from txmatching.optimizer.optimizer_functions import get_compatibility_graph_for_optimizer_api
 from txmatching.optimizer.optimizer_request_object import CompatibilityGraphEntry
+from txmatching.scorers.scorer_constants import HLA_SCORE
 from txmatching.solve_service.solve_from_configuration import \
     solve_from_configuration
 from txmatching.web import API_VERSION, OPTIMIZER_NAMESPACE
@@ -26,13 +27,13 @@ def test_optimizer_api_works(self):
                     {
                         "donor_id": 1,
                         "recipient_id": 2,
-                        "weights": {"hla_compatibility_score": 17,
+                        "weights": {HLA_SCORE: 17,
                                     "donor_age_difference": 1}
                     },
                     {
                         "donor_id": 3,
                         "recipient_id": 4,
-                        "weights": {"hla_compatibility_score": 10,
+                        "weights": {HLA_SCORE: 10,
                                     "donor_age_difference": 17}
                     }
                 ],
@@ -65,7 +66,7 @@ def test_optimizer_api_works(self):
                         ],
                         [
                             {
-                                "hla_compatibility_score": 3
+                                HLA_SCORE: 3
                             },
                             {
                                 "donor_age_difference": 10
@@ -81,7 +82,7 @@ def test_optimizer_api_works(self):
         self.assertEqual(1, res.json['statistics']['number_of_found_cycles_and_chains'])
         self.assertEqual(2, res.json['statistics']['number_of_found_transplants'])
 
-        total_score = sum([dic["weights"]["hla_compatibility_score"] for dic in json_data["compatibility_graph"]])
+        total_score = sum([dic["weights"][HLA_SCORE] for dic in json_data["compatibility_graph"]])
         self.assertEqual(total_score, res.json['cycles_and_chains'][0]['scores'][0])
         self.assertGreaterEqual(total_score, 0)
 
@@ -162,5 +163,5 @@ def test_optimizer_export_api_works(self):
 def _comp_graph_dataclass_list_to_dict_list(dataclass_list: List[CompatibilityGraphEntry]):
     return [{"donor_id": entry.donor_id,
              "recipient_id": entry.recipient_id,
-             "weights": {"hla_compatibility_score": entry.weights["hla_compatibility_score"]}}
+             "weights": {HLA_SCORE: entry.weights[HLA_SCORE]}}
             for entry in dataclass_list]

txmatching/data_transfer_objects/optimizer/optimizer_in_swagger.py

@@ -1,5 +1,6 @@
 from flask_restx import fields
 
+from txmatching.scorers.scorer_constants import HLA_SCORE
 from txmatching.web.web_utils.namespaces import optimizer_api
 
 
@@ -49,13 +50,13 @@ def output(self, key, obj, **kwargs):
     'limitations': fields.Nested(LimitationsJson, reqired=False),
     'scoring': fields.List(required=False, cls_or_instance=fields.List(requred=True, cls_or_instance=DictItem(
         attribute="calling_args")), example=[[{"transplant_count": 1}],
-                                             [{"hla_compatibility_score": 3}, {"donor_age_difference": 20}]])
+                                             [{HLA_SCORE: 3}, {"donor_age_difference": 20}]])
 })
 
 CompGraphEntry = optimizer_api.model("CompatibilityGraphEntry", {
     "donor_id": fields.Integer(required=True, example=1),
     "recipient_id": fields.Integer(required=True, example=2),
-    "weights": DictItem(attribute="calling_args", example={"hla_compatibility_score": 17,
+    "weights": DictItem(attribute="calling_args", example={HLA_SCORE: 17,
                                                            "donor_age_difference": 1
                                                            })
 })

txmatching/database/services/matching_service.py

@@ -30,6 +30,7 @@
     AntibodyMatchForHLAGroup, get_crossmatched_antibodies)
 from txmatching.utils.transplantation_warning import (TransplantWarningDetail,
                                                       TransplantWarnings)
+from txmatching.scorers.scorer_constants import HLA_SCORE
 
 logger = logging.getLogger(__name__)
 
@@ -67,9 +68,9 @@ def get_matchings_detailed_for_pairing_result_model(
                                                                              txm_event.active_and_valid_donors_dict,
                                                                              compatibility_graph)
 
-    number_of_possible_transplants = len([weights['hla_compatibility_score']
+    number_of_possible_transplants = len([weights[HLA_SCORE]
                                           for weights in compatibility_graph_of_db_ids.values() if
-                                          weights["hla_compatibility_score"] >= 0])
+                                          weights[HLA_SCORE] >= 0])
 
     number_of_possible_recipients = len([
         recipient_db_id for recipient_db_id in txm_event.active_and_valid_recipients_dict.keys()
@@ -149,7 +150,7 @@ def _create_transplant_dto(pair: DonorRecipientPair):
 
         return TransplantDTOOut(
             score=latest_matchings_detailed.scores_tuples[(pair.donor.db_id, pair.recipient.db_id)][
-                'hla_compatibility_score'],
+                HLA_SCORE],
             max_score=latest_matchings_detailed.max_transplant_score,
             compatible_blood=latest_matchings_detailed.blood_compatibility_tuples[
                 (pair.donor.db_id, pair.recipient.db_id)],
@@ -247,6 +248,6 @@ def recipient_has_at_least_one_donor(score_dict, recipient_db_id, active_and_val
     """
     Returns true if the recipient has at least one donor, otherwise returns false.
     """
-    return sum(score_dict.get((item, recipient_db_id), -1)['hla_compatibility_score'] >= 0 for item in
+    return sum(score_dict.get((item, recipient_db_id), -1)[HLA_SCORE] >= 0 for item in
                active_and_valid_donors_dict.keys()
                if isinstance(score_dict.get((item, recipient_db_id), -1), dict)) > 0

txmatching/database/services/scorer_service.py

@@ -6,6 +6,7 @@
 from txmatching.data_transfer_objects.matchings.matchings_model import \
     MatchingsModel
 from txmatching.scorers.compatibility_graph import CompatibilityGraph
+from txmatching.scorers.scorer_constants import HLA_SCORE
 
 
 @dataclass
@@ -14,12 +15,12 @@ class CompatibilityGraphDto:
 
 
 def compatibility_graph_to_dict(compatibility_graph: CompatibilityGraph) -> Dict[str, List[List[int]]]:
-    return {"compatibility_graph_dto": [[int(pair[0]), int(pair[1]), weighths["hla_compatibility_score"]]
+    return {"compatibility_graph_dto": [[int(pair[0]), int(pair[1]), weighths[HLA_SCORE]]
                                         for pair, weighths in compatibility_graph.items()]}
 
 
 def compatibility_graph_from_dict(compatibility_graph_dict: Dict[str, List[List[int]]]) -> CompatibilityGraph:
-    return {(pair[0], pair[1]): {"hla_compatibility_score": pair[2]} for pair in
+    return {(pair[0], pair[1]): {HLA_SCORE: pair[2]} for pair in
             compatibility_graph_dict["compatibility_graph_dto"]}
 
 

txmatching/optimizer/optimizer_functions.py

@@ -16,6 +16,7 @@
 from txmatching.scorers.compatibility_graph import OptimizerCompatibilityGraph
 from txmatching.scorers.scorer_from_config import scorer_from_configuration
 from txmatching.scorers.split_hla_additive_scorer import SplitScorer
+from txmatching.scorers.scorer_constants import HLA_SCORE
 from txmatching.solve_service.solver_lock import run_with_solver_lock
 from txmatching.solvers.ilp_solver.txm_configuration_for_ilp import \
     DataAndConfigurationForILPSolver
@@ -219,8 +220,8 @@ def _get_donor_score_matrix(comp_graph: List[CompatibilityGraphEntry], donor_to_
 
 def _hla_score_for_pair(donor_id: int, recipient_id: int, comp_graph: List[CompatibilityGraphEntry]) -> int:
     for row in comp_graph:
-        if row.donor_id == donor_id and row.recipient_id == recipient_id and "hla_compatibility_score" in row.weights:
-            return row.weights["hla_compatibility_score"]
+        if row.donor_id == donor_id and row.recipient_id == recipient_id and HLA_SCORE in row.weights:
+            return row.weights[HLA_SCORE]
     return -1
 
 
@@ -253,7 +254,7 @@ def get_optimizer_configuration(config: ConfigParameters) -> OptimizerConfigurat
         max_chain_length=config.max_sequence_length,
         custom_algorithm_settings={}
     )
-    scoring = [[{"hla_compatibility_score": 1}]]
+    scoring = [[{HLA_SCORE: 1}]]
     return OptimizerConfiguration(
         limitations=limitations,
         scoring=scoring
@@ -275,8 +276,8 @@ def get_compatibility_graph_for_optimizer_api(donors_dict: Dict[DonorDbId, Donor
                     donor_id=donor_id,
                     recipient_id=recipient_id,
                     weights={
-                        "hla_compatibility_score": int(
-                            compatibility_graph_from_scorer[(i, j)]['hla_compatibility_score'])
+                        HLA_SCORE: int(
+                            compatibility_graph_from_scorer[(i, j)][HLA_SCORE])
                     }
                 )
                 compatibility_graph.append(comp_graph_entry)

txmatching/scorers/additive_scorer.py

@@ -6,7 +6,7 @@
 from txmatching.patients.patient_types import DonorDbId, RecipientDbId
 from txmatching.scorers.compatibility_graph import CompatibilityGraph
 from txmatching.scorers.scorer_base import ScorerBase
-from txmatching.scorers.scorer_constants import ORIGINAL_DONOR_RECIPIENT_SCORE
+from txmatching.scorers.scorer_constants import ORIGINAL_DONOR_RECIPIENT_SCORE, HLA_SCORE
 from txmatching.solvers.matching.matching import Matching
 from txmatching.utils.hla_system.compatibility_index import CIConfiguration
 
@@ -76,7 +76,7 @@ def get_compatibility_graph(self,
                                                                                         recipient.related_donors_db_ids
                                                                                         if donor_db_id in donors_dict])
                 if score >= 0:
-                    compatibility_graph[(donor_enum, recipient_enum)] = {"hla_compatibility_score": int(score)}
+                    compatibility_graph[(donor_enum, recipient_enum)] = {HLA_SCORE: int(score)}
 
         return compatibility_graph
 

txmatching/scorers/scorer_constants.py

@@ -2,3 +2,5 @@
 TRANSPLANT_IMPOSSIBLE_SCORE = -1.0
 NEGATIVE_SCORE_BINARY_MODE = -1.0
 POSITIVE_SCORE_BINARY_MODE = 1.0
+
+HLA_SCORE = "hla_compatibility_score"

txmatching/solvers/all_solutions_solver/scoring_utils.py

@@ -3,8 +3,9 @@
 from txmatching.scorers.compatibility_graph import CompatibilityGraph
 from txmatching.solvers.donor_recipient_pair_idx_only import \
     DonorRecipientPairIdxOnly
+from txmatching.scorers.scorer_constants import HLA_SCORE
 
 
 def get_score_for_idx_pairs(compatibility_graph: CompatibilityGraph,
                             pairs: Iterable[DonorRecipientPairIdxOnly]):
-    return sum(compatibility_graph[(pair.donor_idx, pair.recipient_idx)]["hla_compatibility_score"] for pair in pairs)
+    return sum(compatibility_graph[(pair.donor_idx, pair.recipient_idx)][HLA_SCORE] for pair in pairs)