amachine.am_create.am_star_join

View Source

  1from collections import defaultdict
  2import copy
  3import random
  4
  5import numpy as np
  6
  7from ..am_hmm          import HMM
  8from ..am_causal_state import CausalState
  9from ..am_transition   import Transition
 10
 11from ..am_random import exp_uniform_blend, resolve_rng
 12from ..am_vocabulary import Vocabulary
 13
 14def star_join(
 15    exit_symbol : str, 
 16    enter_symbols : list[str],
 17    machines : list[HMM],
 18    mode_residency_factor : float ) -> HMM :
 19
 20    machine = HMM()
 21
 22    isomorphic_groups = defaultdict(list)
 23    for i, m in enumerate( machines ) :
 24        if m.isoclass is not None :
 25            isomorphic_groups[ m.isoclass ].append( i )
 26
 27    # since we are merging multuple machines which might have name collision
 28    # we need to rename the states to ensure uniqueness
 29    def rename_state( base_name : str, g : int ) :
 30        return f"{g}/{base_name}"
 31
 32    def get_gid( idx : int, isoclass : int | None ) :
 33        return idx if isoclass is None else isoclass
 34
 35    machine.set_alphabet( [ exit_symbol ] )
 36
 37    for m in machines :
 38        machine.extend_alphabet( alphabet=m.alphabet )
 39
 40    # create a connector state and connector state class
 41    connector_state = CausalState(
 42        name=f"/c", 
 43        classes=set({"connector"}) 
 44    )
 45    
 46    # initial states before adding each machines states
 47    machine.set_states( [ connector_state ] )
 48    machine.start_state = 0
 49
 50    # number of machines (groups of states)
 51    n_groups = len( machines )
 52
 53    # make sure we have enough symbols (otherwise connector can't be unifilar)
 54    if n_groups > len(enter_symbols) :
 55        raise Exception(
 56            f"Too few enter symbols given number of machines"
 57        )
 58
 59    # for each given machine
 60    for m_idx, m in enumerate( machines ) : 
 61
 62        # default to the index of the machine in the list
 63        m_gid = get_gid( m_idx, m.isoclass )
 64
 65        # give the states from this machine a class name
 66        m_classes = { 
 67            f"m_{m_idx}", 
 68            f"isoclass_{m.isoclass}" 
 69        }
 70
 71        added_states = []
 72        
 73        # create a state and extend our existing machine to include it
 74        for s_idx, state in enumerate( m.states ) :
 75
 76            isomorphs=set()
 77            if m.isoclass is not None and m.isoclass in isomorphic_groups :
 78
 79                for other_idx in isomorphic_groups[ m.isoclass ] :
 80                    
 81                    if other_idx == m_idx : 
 82                        continue
 83                    
 84                    other_m = machines[ other_idx ]
 85                    isomorphs.add(  
 86                        rename_state( 
 87                            other_m.states[ s_idx ].name, 
 88                            get_gid( other_idx, other_m.isoclass ) )
 89                    )
 90
 91            added_states.append( 
 92                CausalState( 
 93                    name=rename_state(state.name, m_gid),
 94                    classes=( m_classes | state.classes ),
 95                    isomorphs=isomorphs
 96                ) 
 97            )
 98
 99        machine.extend_states( added_states )
100
101        added_transitions = []
102
103        # add all of the transitions from the machine
104        for tr in m.transitions :
105
106            # get the names of the states for the transition
107            origin_state_name = rename_state( m.states[ tr.origin_state_idx ].name, m_gid )
108            target_state_name = rename_state( m.states[ tr.target_state_idx ].name, m_gid )
109
110            # idx of the symbol remaped to this machines alphabet list
111            new_symbol_idx = machine.symbol_idx_map[ m.alphabet[ tr.symbol_idx ] ]
112
113            # create and add the new transition
114            added_transitions.append( Transition(
115                origin_state_idx=machine.state_idx_map[ origin_state_name ],
116                target_state_idx=machine.state_idx_map[ target_state_name ],
117                prob=tr.prob,
118                symbol_idx=new_symbol_idx,
119                pq=None,
120                cross_component=False
121            ) )
122
123        machine.extend_transitions( added_transitions )
124
125        # Add connector transitions, and adjust transition probabilities to sum to 1
126
127        # the name of the state that is the entry point to this group from the connector
128        m_entry_state_name = rename_state( m.states[ m.start_state ].name, m_gid )
129
130        # get the index of the entry state for this machine
131        m_entry_state_idx = machine.state_idx_map[ m_entry_state_name ]
132
133
134        # Get the within group transitions from m's entry state
135        # ( the probabilities will need to be adjusted )
136        transition_ids_from_m_entry = set()
137        for i, tr in enumerate( machine.transitions ) : 
138            if tr.origin_state_idx == m_entry_state_idx :
139                transition_ids_from_m_entry.add( i )
140        
141        n_from_entry = len( transition_ids_from_m_entry )
142
143        # Pr of staying in this group is distributed over the within group outgoing edges from the entry state 
144        for i in transition_ids_from_m_entry :
145
146            machine.transitions[ i ] = Transition(
147                origin_state_idx=machine.transitions[ i ].origin_state_idx,
148                target_state_idx=machine.transitions[ i ].target_state_idx,
149                prob=mode_residency_factor / n_from_entry,
150                symbol_idx=machine.transitions[ i ].symbol_idx,
151                pq=None,
152                cross_component=False
153            )
154
155        # from m's entry state back to connector
156        escape_pr = 1.0 - mode_residency_factor
157
158        machine.extend_transitions( transitions=[
159            Transition(
160                origin_state_idx=m_entry_state_idx,
161                target_state_idx=machine.start_state,
162                prob=escape_pr,
163                symbol_idx=machine.symbol_idx_map[ exit_symbol ],
164                pq=None,
165                cross_component=False
166            )
167        ] )
168
169        # from the connector to m's entry state
170        machine.extend_alphabet( alphabet=[ enter_symbols[ m_idx ] ] )
171        
172        machine.extend_transitions( transitions=[
173            Transition(
174                origin_state_idx=machine.start_state,
175                target_state_idx=m_entry_state_idx,
176                prob=( 1.0 / n_groups ),
177                symbol_idx=machine.symbol_idx_map[ enter_symbols[ m_idx ] ],
178                pq=None,
179                cross_component=False
180            )
181        ] )
182
183    return machine

def star_join( exit_symbol: str, enter_symbols: list[str], machines: list[amachine.am_hmm.HMM], mode_residency_factor: float) -> amachine.am_hmm.HMM: View Source

 15def star_join(
 16    exit_symbol : str, 
 17    enter_symbols : list[str],
 18    machines : list[HMM],
 19    mode_residency_factor : float ) -> HMM :
 20
 21    machine = HMM()
 22
 23    isomorphic_groups = defaultdict(list)
 24    for i, m in enumerate( machines ) :
 25        if m.isoclass is not None :
 26            isomorphic_groups[ m.isoclass ].append( i )
 27
 28    # since we are merging multuple machines which might have name collision
 29    # we need to rename the states to ensure uniqueness
 30    def rename_state( base_name : str, g : int ) :
 31        return f"{g}/{base_name}"
 32
 33    def get_gid( idx : int, isoclass : int | None ) :
 34        return idx if isoclass is None else isoclass
 35
 36    machine.set_alphabet( [ exit_symbol ] )
 37
 38    for m in machines :
 39        machine.extend_alphabet( alphabet=m.alphabet )
 40
 41    # create a connector state and connector state class
 42    connector_state = CausalState(
 43        name=f"/c", 
 44        classes=set({"connector"}) 
 45    )
 46    
 47    # initial states before adding each machines states
 48    machine.set_states( [ connector_state ] )
 49    machine.start_state = 0
 50
 51    # number of machines (groups of states)
 52    n_groups = len( machines )
 53
 54    # make sure we have enough symbols (otherwise connector can't be unifilar)
 55    if n_groups > len(enter_symbols) :
 56        raise Exception(
 57            f"Too few enter symbols given number of machines"
 58        )
 59
 60    # for each given machine
 61    for m_idx, m in enumerate( machines ) : 
 62
 63        # default to the index of the machine in the list
 64        m_gid = get_gid( m_idx, m.isoclass )
 65
 66        # give the states from this machine a class name
 67        m_classes = { 
 68            f"m_{m_idx}", 
 69            f"isoclass_{m.isoclass}" 
 70        }
 71
 72        added_states = []
 73        
 74        # create a state and extend our existing machine to include it
 75        for s_idx, state in enumerate( m.states ) :
 76
 77            isomorphs=set()
 78            if m.isoclass is not None and m.isoclass in isomorphic_groups :
 79
 80                for other_idx in isomorphic_groups[ m.isoclass ] :
 81                    
 82                    if other_idx == m_idx : 
 83                        continue
 84                    
 85                    other_m = machines[ other_idx ]
 86                    isomorphs.add(  
 87                        rename_state( 
 88                            other_m.states[ s_idx ].name, 
 89                            get_gid( other_idx, other_m.isoclass ) )
 90                    )
 91
 92            added_states.append( 
 93                CausalState( 
 94                    name=rename_state(state.name, m_gid),
 95                    classes=( m_classes | state.classes ),
 96                    isomorphs=isomorphs
 97                ) 
 98            )
 99
100        machine.extend_states( added_states )
101
102        added_transitions = []
103
104        # add all of the transitions from the machine
105        for tr in m.transitions :
106
107            # get the names of the states for the transition
108            origin_state_name = rename_state( m.states[ tr.origin_state_idx ].name, m_gid )
109            target_state_name = rename_state( m.states[ tr.target_state_idx ].name, m_gid )
110
111            # idx of the symbol remaped to this machines alphabet list
112            new_symbol_idx = machine.symbol_idx_map[ m.alphabet[ tr.symbol_idx ] ]
113
114            # create and add the new transition
115            added_transitions.append( Transition(
116                origin_state_idx=machine.state_idx_map[ origin_state_name ],
117                target_state_idx=machine.state_idx_map[ target_state_name ],
118                prob=tr.prob,
119                symbol_idx=new_symbol_idx,
120                pq=None,
121                cross_component=False
122            ) )
123
124        machine.extend_transitions( added_transitions )
125
126        # Add connector transitions, and adjust transition probabilities to sum to 1
127
128        # the name of the state that is the entry point to this group from the connector
129        m_entry_state_name = rename_state( m.states[ m.start_state ].name, m_gid )
130
131        # get the index of the entry state for this machine
132        m_entry_state_idx = machine.state_idx_map[ m_entry_state_name ]
133
134
135        # Get the within group transitions from m's entry state
136        # ( the probabilities will need to be adjusted )
137        transition_ids_from_m_entry = set()
138        for i, tr in enumerate( machine.transitions ) : 
139            if tr.origin_state_idx == m_entry_state_idx :
140                transition_ids_from_m_entry.add( i )
141        
142        n_from_entry = len( transition_ids_from_m_entry )
143
144        # Pr of staying in this group is distributed over the within group outgoing edges from the entry state 
145        for i in transition_ids_from_m_entry :
146
147            machine.transitions[ i ] = Transition(
148                origin_state_idx=machine.transitions[ i ].origin_state_idx,
149                target_state_idx=machine.transitions[ i ].target_state_idx,
150                prob=mode_residency_factor / n_from_entry,
151                symbol_idx=machine.transitions[ i ].symbol_idx,
152                pq=None,
153                cross_component=False
154            )
155
156        # from m's entry state back to connector
157        escape_pr = 1.0 - mode_residency_factor
158
159        machine.extend_transitions( transitions=[
160            Transition(
161                origin_state_idx=m_entry_state_idx,
162                target_state_idx=machine.start_state,
163                prob=escape_pr,
164                symbol_idx=machine.symbol_idx_map[ exit_symbol ],
165                pq=None,
166                cross_component=False
167            )
168        ] )
169
170        # from the connector to m's entry state
171        machine.extend_alphabet( alphabet=[ enter_symbols[ m_idx ] ] )
172        
173        machine.extend_transitions( transitions=[
174            Transition(
175                origin_state_idx=machine.start_state,
176                target_state_idx=m_entry_state_idx,
177                prob=( 1.0 / n_groups ),
178                symbol_idx=machine.symbol_idx_map[ enter_symbols[ m_idx ] ],
179                pq=None,
180                cross_component=False
181            )
182        ] )
183
184    return machine