amachine.am_create.am_isomorphic_to_with_superstate_permutations

  1import copy
  2from collections import defaultdict
  3
  4from ..am_hmm    import HMM
  5from ..am_random import unique_random_permutations
  6
  7import numpy as np
  8
  9
 10def isomorphic_to_with_superstate_permutations(
 11    m : HMM,
 12    superstate_connectors : list[ tuple[int,int] ],
 13    n_permutations : int,
 14    decorator : str,
 15    np_rng : np.random.Generator | None = None ) -> list[ HMM ] :
 16
 17    n_superstates = len( superstate_connectors )
 18
 19    permutations = unique_random_permutations(
 20        [ n_superstates ],
 21        n_samples=n_permutations-1,
 22        np_rng=np_rng
 23    )
 24
 25    permutations = [
 26        p[ 0 ] for p in permutations
 27    ]
 28
 29    enter_state_to_superstate : dict[ int, int ] = {}
 30    exit_state_to_superstate  : dict[ int, int ] = {}
 31    connector_states : set[int] = set()
 32
 33    for i, ( entry_state, exit_state ) in enumerate( superstate_connectors ) :
 34        enter_state_to_superstate[ entry_state ] = i
 35        exit_state_to_superstate[   exit_state ] = i
 36        connector_states.add( entry_state )
 37        connector_states.add( exit_state  )
 38
 39    result = [ copy.deepcopy( m ) ]
 40
 41    def is_external( tr ) :
 42        return ( tr.origin_state_idx in exit_state_to_superstate 
 43             and tr.target_state_idx in enter_state_to_superstate 
 44             and exit_state_to_superstate[ tr.origin_state_idx ] 
 45              != enter_state_to_superstate[ tr.target_state_idx ] )
 46
 47    for p_idx, permutation in enumerate( permutations ) :
 48
 49        alphabet = tuple(m.alphabet)
 50
 51        states = [ 
 52            s.modified_deep_copy( name=f"{s.name}{decorator}" )
 53            for s in m.states
 54        ]
 55
 56        exit_external_mass          : dict[ int, float ] = defaultdict( float )
 57        exit_external_mass_permuted : dict[ int, float ] = defaultdict( float )
 58
 59        transitions = []
 60        for i, tr in enumerate( m.transitions ) :
 61
 62            new_origin_idx = tr.origin_state_idx
 63            new_target_idx = tr.target_state_idx
 64
 65            if is_external( tr ) :
 66                 
 67                origin_superstate =  exit_state_to_superstate[ tr.origin_state_idx ]
 68                target_superstate = enter_state_to_superstate[ tr.target_state_idx ]
 69
 70                new_origin_superstate = permutation[ origin_superstate ]
 71                new_target_superstate = permutation[ target_superstate ]
 72
 73                new_origin_idx = superstate_connectors[ new_origin_superstate ][ 1 ]
 74                new_target_idx = superstate_connectors[ new_target_superstate ][ 0 ]
 75
 76                exit_external_mass[     tr.origin_state_idx ] += tr.prob
 77                exit_external_mass_permuted[ new_origin_idx ] += tr.prob
 78
 79            transitions.append( 
 80                tr.modified_deep_copy( 
 81                    origin_state_idx=new_origin_idx,
 82                    target_state_idx=new_target_idx 
 83                ) 
 84            )
 85
 86        # adjust the probabilities of the between superstate nodes
 87        # to preserve row-stochasticity
 88        for i, tr in enumerate( transitions ) :
 89            if is_external( tr ) :
 90                original_mass = exit_external_mass[ tr.origin_state_idx ]
 91                new_mass      = exit_external_mass_permuted[ tr.origin_state_idx ]
 92                transitions[ i ] = tr.modified_deep_copy(
 93                    prob = tr.prob * ( original_mass / new_mass )
 94                )
 95
 96        res = HMM(
 97            states=states,
 98            transitions=transitions,
 99            start_state=0,
100            alphabet=alphabet
101        )
102
103        if not res.is_row_stochastic() :
104            raise Exception( "Superstate permutation is not row stochastic" )
105
106        if not  res.is_unifilar() :
107            raise Exception( "Superstate permutation is not unifilar" )
108
109        result.append( res )
110
111    return result