1- """ Calls the tHMM functions and outputs the parameters needed to generate the Figures. """
1+ """Calls the tHMM functions and outputs the parameters needed to generate the Figures."""
22
33import itertools
4- from typing import Any , Tuple
4+ from typing import Any
5+
56import numpy as np
67from scipy .optimize import linear_sum_assignment
78from sklearn .metrics import rand_score
8- from . tHMM import tHMM
9+
910from .BaumWelch import (
10- do_E_step ,
1111 calculate_log_likelihood ,
12- do_M_step ,
12+ do_E_step ,
1313 do_M_E_step ,
1414 do_M_E_step_atonce ,
15+ do_M_step ,
1516)
17+ from .tHMM import tHMM
1618
1719
1820def fit_list (
1921 tHMMobj_list : list [tHMM ], tolerance : float = 1e-6 , max_iter : int = 200 , rng = None
20- ) -> Tuple [list [np .ndarray ], list [np .ndarray ], float ]:
22+ ) -> tuple [list [list [ np .ndarray ]] , list [list [ list [ np .ndarray ]] ], float ]:
2123 """
2224 Runs the tHMM function through Baum Welch fitting for a list containing a set of data for different concentrations.
2325
@@ -34,10 +36,7 @@ def fit_list(
3436
3537 # Step 0: initialize with random assignments and do an M step
3638 # when there are no fixed emissions, we need to randomize the start
37- init_gam = [
38- [rng .dirichlet (np .ones (tO .num_states ), size = len (lin )) for lin in tO .X ]
39- for tO in tHMMobj_list
40- ]
39+ init_gam = [[rng .dirichlet (np .ones (tO .num_states ), size = len (lin )) for lin in tO .X ] for tO in tHMMobj_list ]
4140
4241 if len (tHMMobj_list ) > 1 : # it means we are fitting several concentrations at once.
4342 do_M_E_step_atonce (tHMMobj_list , init_gam )
@@ -46,15 +45,15 @@ def fit_list(
4645
4746 # Step 1: first E step
4847 MSD_list , NF_list , betas_list , gammas_list = map (
49- list , zip (* [do_E_step (tHMM ) for tHMM in tHMMobj_list ])
48+ list , zip (* [do_E_step (tHMM ) for tHMM in tHMMobj_list ], strict = False )
5049 )
5150 old_LL = calculate_log_likelihood (NF_list )
5251
5352 # first stopping condition check
5453 for _ in range (max_iter ):
5554 do_M_step (tHMMobj_list , MSD_list , betas_list , gammas_list )
5655 MSD_list , NF_list , betas_list , gammas_list = map (
57- list , zip (* [do_E_step (tHMM ) for tHMM in tHMMobj_list ])
56+ list , zip (* [do_E_step (tHMM ) for tHMM in tHMMobj_list ], strict = False )
5857 )
5958 new_LL = calculate_log_likelihood (NF_list )
6059 if new_LL - old_LL < tolerance :
@@ -67,7 +66,7 @@ def fit_list(
6766
6867def Analyze_list (
6968 pop_list : list , num_states : int , fpi = None , fT = None , rng = None , write_states = False
70- ) -> Tuple [list [tHMM ], float , list [np .ndarray ]]:
69+ ) -> tuple [list [tHMM ], float , list [list [ list [ np .ndarray ]] ]]:
7170 """This function runs the analyze function for the case when we want to fit multiple conditions at the same time.
7271 :param pop_list: The list of cell populations to run the analyze function on.
7372 :param num_states: The number of states that we want to run the model for.
@@ -110,7 +109,7 @@ def run_Analyze_over(
110109 atonce = False ,
111110 list_of_fpi = None ,
112111 list_of_fT = None ,
113- ) -> list [Tuple [list [tHMM ], float , list [np .ndarray ]]]:
112+ ) -> list [tuple [list [tHMM ], float , list [np .ndarray ]]]:
114113 """
115114 A function that can be parallelized to speed up figure creation.
116115
@@ -180,37 +179,23 @@ def Results(tHMMobj: tHMM, LL: float) -> dict[str, Any]:
180179
181180 true_states_by_lineage = [lin .states for lin in tHMMobj .X ]
182181
183- results_dict ["transition_matrix_similarity" ] = np .linalg .norm (
184- tHMMobj .estimate .T - tHMMobj .X [0 ].T
185- )
182+ results_dict ["transition_matrix_similarity" ] = np .linalg .norm (tHMMobj .estimate .T - tHMMobj .X [0 ].T )
186183
187- results_dict ["pi_similarity" ] = np .linalg .norm (
188- tHMMobj .X [0 ].pi - tHMMobj .estimate .pi
189- )
184+ results_dict ["pi_similarity" ] = np .linalg .norm (tHMMobj .X [0 ].pi - tHMMobj .estimate .pi )
190185
191186 # Get the estimated parameter values
192- results_dict ["param_estimates" ] = [
193- tHMMobj .estimate .E [x ].params for x in range (tHMMobj .num_states )
194- ]
187+ results_dict ["param_estimates" ] = [tHMMobj .estimate .E [x ].params for x in range (tHMMobj .num_states )]
195188
196189 # Get the true parameter values
197- results_dict ["param_trues" ] = [
198- tHMMobj .X [0 ].E [x ].params for x in range (tHMMobj .num_states )
199- ]
190+ results_dict ["param_trues" ] = [tHMMobj .X [0 ].E [x ].params for x in range (tHMMobj .num_states )]
200191
201192 # Get the distance between distributions of two states
202- results_dict ["distribution distance 0" ] = tHMMobj .estimate .E [0 ].dist (
203- tHMMobj .X [0 ].E [0 ]
204- )
205- results_dict ["distribution distance 1" ] = tHMMobj .estimate .E [1 ].dist (
206- tHMMobj .X [0 ].E [1 ]
207- )
193+ results_dict ["distribution distance 0" ] = tHMMobj .estimate .E [0 ].dist (tHMMobj .X [0 ].E [0 ])
194+ results_dict ["distribution distance 1" ] = tHMMobj .estimate .E [1 ].dist (tHMMobj .X [0 ].E [1 ])
208195
209196 # 2. Calculate accuracy after switching states
210197 results_dict ["state_counter" ] = np .bincount (pred_states [0 ])
211- results_dict ["state_proportions" ] = [
212- 100.0 * i / len (pred_states [0 ]) for i in results_dict ["state_counter" ]
213- ]
198+ results_dict ["state_proportions" ] = [100.0 * i / len (pred_states [0 ]) for i in results_dict ["state_counter" ]]
214199 results_dict ["state_proportions_0" ] = results_dict ["state_proportions" ][0 ]
215200 results_dict ["state_similarity" ] = 100.0 * rand_score (
216201 list (itertools .chain (* true_states_by_lineage )),
@@ -223,7 +208,7 @@ def Results(tHMMobj: tHMM, LL: float) -> dict[str, Any]:
223208 return results_dict
224209
225210
226- def permute_states (tHMMobj : tHMM , switch_map : np .ndarray ) -> Tuple [Any , list ]:
211+ def permute_states (tHMMobj : tHMM , switch_map : np .ndarray ) -> tuple [Any , list ]:
227212 """
228213 This function takes the tHMMobj and the predicted states,
229214 and finds out whether we need to switch the state identities or not based on the likelihood.
@@ -234,9 +219,7 @@ def permute_states(tHMMobj: tHMM, switch_map: np.ndarray) -> Tuple[Any, list]:
234219 """
235220 pred_states = tHMMobj .predict ()
236221
237- pred_states_switched = [
238- np .array ([switch_map [st ] for sublist in pred_states for st in sublist ])
239- ]
222+ pred_states_switched = [np .array ([switch_map [st ] for sublist in pred_states for st in sublist ])]
240223
241224 # Rearrange the values in the transition matrix
242225 tHMMobj .estimate .T = tHMMobj .estimate .T [switch_map , :]
0 commit comments