amachine.am_create.am_star_composition

View Source

  1from collections import defaultdict
  2
  3from sympy.core.operations import str_signature
  4
  5from ..am_hmm          import HMM
  6from ..am_causal_state import CausalState
  7from ..am_transition   import Transition
  8from ..am_id_generator import next_composition_id
  9
 10def star_composition(
 11    exit_symbol                 : str,
 12    enter_symbols               : list[str],
 13    component_residency_factor  : float,
 14    component_groups            : dict[str, list[HMM]],
 15    instances_per_component     : int | dict[ str, list[int] ] = 1,
 16    composition_id              : str | None = None, ) -> HMM :
 17        
 18    if component_residency_factor <= 0.0 or component_residency_factor >= 1.0 :
 19        raise ValueError( f"component_residency_factor must be in (0,1), got {component_residency_factor}." )
 20
 21    if isinstance( instances_per_component, int ):
 22        ic_dict = {}
 23        for group_id, components in component_groups.items():
 24            ic_dict[ group_id ] = [ instances_per_component ] * len(components)
 25        instances_per_component = ic_dict
 26
 27    if composition_id is None :
 28        composition_id = str(next_composition_id())
 29
 30    component_pool : list[ tuple[ HMM, str, int ] ] = []
 31    for group_id, components in component_groups.items() :
 32        for c_i, c in enumerate( components ) :
 33            component_pool.append( ( c, group_id, c_i ) )
 34
 35    total_instances = sum( [
 36        sum( counts ) for counts in instances_per_component.values()
 37    ] )
 38
 39    if total_instances > len( enter_symbols ) :
 40        raise Exception( f"Too few enter symbols ({len(enter_symbols)}) given number of instances ({total_instances})." )
 41
 42    ##############################################################
 43    # Build the alphabet
 44
 45    alphabet       = [ exit_symbol ]
 46    symbol_idx_map = { exit_symbol : 0 }
 47    symbol_set     = { exit_symbol }
 48
 49    for m, _, _ in component_pool :
 50        for s in m.alphabet :
 51            if s not in symbol_set :
 52                symbol_idx_map[ s ] = len( alphabet )
 53                alphabet.append( s )
 54                symbol_set.add( s )
 55
 56    for s in enter_symbols[ :total_instances ] :
 57        if s not in symbol_set :
 58            symbol_idx_map[ s ] = len( alphabet )
 59            alphabet.append( s )
 60            symbol_set.add( s )
 61
 62    ##############################################################
 63    # Add the states and transitions
 64
 65    cstate_name_to_global_idxs = {
 66        group_id : defaultdict( list )
 67        for group_id in component_groups.keys()
 68    }
 69
 70    cstate_instance_to_global_idx = [ {} for _ in range( total_instances ) ]
 71    global_idx_to_component_state = {}
 72    global_idx_to_group           = {}
 73    g_sidx       = 0
 74    instance_idx = 0
 75    states       = []
 76    transitions  = []
 77
 78    composition_depth = max( (  
 79        m.composition_depth for gr in component_groups.values() for m in gr
 80    ) ) + 1
 81
 82    connector_state = CausalState( name=f"/c_{composition_id}" )
 83    connector_state.add_class( composition_depth, "connector" )
 84
 85    states.append( connector_state )
 86    connector_g_sidx = g_sidx
 87    g_sidx += 1
 88
 89    for m_idx, ( m, group_id, group_offset ) in enumerate( component_pool ) :
 90        
 91        # Instance count for this component
 92        n_instances = instances_per_component[ group_id ][ group_offset ]
 93        
 94        for _ in range( n_instances ):
 95
 96            for s_idx, state in enumerate( m.states ) :
 97
 98                m_classes = {
 99                    composition_depth : {
100                        f"cm_{composition_id}_{m_idx}",
101                        f"cmi_{composition_id}_{instance_idx}",
102                        f"c_{composition_id}",
103                        f"g_{group_id}"
104                    }
105                }
106
107                # Isomorphs are remapped in a later pass
108                states.append( CausalState(
109                    name=f"{g_sidx}/{state.name}",
110                    classes=( state.classes | m_classes )
111                ) )
112
113                global_idx_to_component_state[ g_sidx ] = state
114                global_idx_to_group[ g_sidx ]           = group_id
115                cstate_name_to_global_idxs[ group_id ][ state.name ].append( g_sidx )
116                cstate_instance_to_global_idx[ instance_idx ][ s_idx ] = g_sidx
117
118                g_sidx += 1
119
120            for tr in m.transitions :
121                
122                origin = cstate_instance_to_global_idx[ instance_idx ][ tr.origin_state_idx ]
123                target = cstate_instance_to_global_idx[ instance_idx ][ tr.target_state_idx ]
124                
125                transitions.append( tr.modified_deep_copy(
126                    origin_state_idx=origin,
127                    target_state_idx=target,
128                    symbol_idx=symbol_idx_map[ m.alphabet[ tr.symbol_idx ] ]
129                ) )
130
131            m_entry_g_sidx = cstate_instance_to_global_idx[ instance_idx ][ m.start_state ]
132
133            # all of the transitions from the entry state need to have their probabilities adjusted
134            # since we will add a new transition to the connector
135            entry_tr_ids = [
136                i for i, tr in enumerate( transitions )
137                if tr.origin_state_idx == m_entry_g_sidx
138            ]
139
140            for i in entry_tr_ids :
141                transitions[ i ] = transitions[ i ].modified_deep_copy(
142                    prob = transitions[ i ].prob * component_residency_factor
143                )
144
145            # Entry state -> connector  (escape with remaining probability mass)
146            transitions.append( Transition(
147                origin_state_idx=m_entry_g_sidx,
148                target_state_idx=connector_g_sidx,
149                prob=( 1.0 - component_residency_factor ),
150                symbol_idx=symbol_idx_map[ exit_symbol ],
151                composition_depth=composition_depth 
152            ) )
153
154            # Connector -> entry state
155            # Probability is 1.0 / total_instances, mapping to unique enter symbol per instance
156            transitions.append( Transition(
157                origin_state_idx=connector_g_sidx,
158                target_state_idx=m_entry_g_sidx,
159                prob=( 1.0 / total_instances ),
160                symbol_idx=symbol_idx_map[ enter_symbols[ instance_idx ] ],
161                composition_depth=composition_depth 
162            ) )
163
164            # Increment global instance tracker
165            instance_idx += 1
166
167    # Remap and add the isomorphs
168    for g_idx, s in enumerate( states ) :
169
170        if g_idx not in global_idx_to_component_state :
171            continue  # skip the connector state
172
173        group_id = global_idx_to_group[ g_idx ]
174        c_state  = global_idx_to_component_state[ g_idx ]
175
176        isomorphs = set()
177        pseudo_isomorphs = set()
178
179        for c_iso in c_state.isomorphs :
180            # Isomorphs are scoped to the group
181            g_isos = cstate_name_to_global_idxs[ group_id ][ c_iso ]
182            for g_iso in g_isos :
183                isomorphs.add( states[ g_iso ].name )
184
185        for c_iso in c_state.pseudo_isomorphs :
186            g_isos = cstate_name_to_global_idxs[ group_id ][ c_iso ]
187            for g_iso in g_isos :
188                pseudo_isomorphs.add( states[ g_iso ].name )
189
190        states[ g_idx ] = states[ g_idx ].modified_deep_copy(
191            isomorphs=isomorphs,
192            pseudo_isomorphs=pseudo_isomorphs
193        )
194
195    ##############################################################
196
197    res = HMM(
198        states=states,
199        transitions=transitions,
200        alphabet=alphabet,
201        start_state=connector_g_sidx,
202        composition_depth=composition_depth
203    )
204
205    if not res.is_row_stochastic() :
206        raise Exception( "Start composition is not row stochastic" )
207
208    if not  res.is_unifilar() :
209        raise Exception( "Start composition is not unifilar" )
210
211    return res

def star_composition( exit_symbol: str, enter_symbols: list[str], component_residency_factor: float, component_groups: dict[str, list[amachine.am_hmm.HMM]], instances_per_component: int | dict[str, list[int]] = 1, composition_id: str | None = None) -> amachine.am_hmm.HMM: View Source

 11def star_composition(
 12    exit_symbol                 : str,
 13    enter_symbols               : list[str],
 14    component_residency_factor  : float,
 15    component_groups            : dict[str, list[HMM]],
 16    instances_per_component     : int | dict[ str, list[int] ] = 1,
 17    composition_id              : str | None = None, ) -> HMM :
 18        
 19    if component_residency_factor <= 0.0 or component_residency_factor >= 1.0 :
 20        raise ValueError( f"component_residency_factor must be in (0,1), got {component_residency_factor}." )
 21
 22    if isinstance( instances_per_component, int ):
 23        ic_dict = {}
 24        for group_id, components in component_groups.items():
 25            ic_dict[ group_id ] = [ instances_per_component ] * len(components)
 26        instances_per_component = ic_dict
 27
 28    if composition_id is None :
 29        composition_id = str(next_composition_id())
 30
 31    component_pool : list[ tuple[ HMM, str, int ] ] = []
 32    for group_id, components in component_groups.items() :
 33        for c_i, c in enumerate( components ) :
 34            component_pool.append( ( c, group_id, c_i ) )
 35
 36    total_instances = sum( [
 37        sum( counts ) for counts in instances_per_component.values()
 38    ] )
 39
 40    if total_instances > len( enter_symbols ) :
 41        raise Exception( f"Too few enter symbols ({len(enter_symbols)}) given number of instances ({total_instances})." )
 42
 43    ##############################################################
 44    # Build the alphabet
 45
 46    alphabet       = [ exit_symbol ]
 47    symbol_idx_map = { exit_symbol : 0 }
 48    symbol_set     = { exit_symbol }
 49
 50    for m, _, _ in component_pool :
 51        for s in m.alphabet :
 52            if s not in symbol_set :
 53                symbol_idx_map[ s ] = len( alphabet )
 54                alphabet.append( s )
 55                symbol_set.add( s )
 56
 57    for s in enter_symbols[ :total_instances ] :
 58        if s not in symbol_set :
 59            symbol_idx_map[ s ] = len( alphabet )
 60            alphabet.append( s )
 61            symbol_set.add( s )
 62
 63    ##############################################################
 64    # Add the states and transitions
 65
 66    cstate_name_to_global_idxs = {
 67        group_id : defaultdict( list )
 68        for group_id in component_groups.keys()
 69    }
 70
 71    cstate_instance_to_global_idx = [ {} for _ in range( total_instances ) ]
 72    global_idx_to_component_state = {}
 73    global_idx_to_group           = {}
 74    g_sidx       = 0
 75    instance_idx = 0
 76    states       = []
 77    transitions  = []
 78
 79    composition_depth = max( (  
 80        m.composition_depth for gr in component_groups.values() for m in gr
 81    ) ) + 1
 82
 83    connector_state = CausalState( name=f"/c_{composition_id}" )
 84    connector_state.add_class( composition_depth, "connector" )
 85
 86    states.append( connector_state )
 87    connector_g_sidx = g_sidx
 88    g_sidx += 1
 89
 90    for m_idx, ( m, group_id, group_offset ) in enumerate( component_pool ) :
 91        
 92        # Instance count for this component
 93        n_instances = instances_per_component[ group_id ][ group_offset ]
 94        
 95        for _ in range( n_instances ):
 96
 97            for s_idx, state in enumerate( m.states ) :
 98
 99                m_classes = {
100                    composition_depth : {
101                        f"cm_{composition_id}_{m_idx}",
102                        f"cmi_{composition_id}_{instance_idx}",
103                        f"c_{composition_id}",
104                        f"g_{group_id}"
105                    }
106                }
107
108                # Isomorphs are remapped in a later pass
109                states.append( CausalState(
110                    name=f"{g_sidx}/{state.name}",
111                    classes=( state.classes | m_classes )
112                ) )
113
114                global_idx_to_component_state[ g_sidx ] = state
115                global_idx_to_group[ g_sidx ]           = group_id
116                cstate_name_to_global_idxs[ group_id ][ state.name ].append( g_sidx )
117                cstate_instance_to_global_idx[ instance_idx ][ s_idx ] = g_sidx
118
119                g_sidx += 1
120
121            for tr in m.transitions :
122                
123                origin = cstate_instance_to_global_idx[ instance_idx ][ tr.origin_state_idx ]
124                target = cstate_instance_to_global_idx[ instance_idx ][ tr.target_state_idx ]
125                
126                transitions.append( tr.modified_deep_copy(
127                    origin_state_idx=origin,
128                    target_state_idx=target,
129                    symbol_idx=symbol_idx_map[ m.alphabet[ tr.symbol_idx ] ]
130                ) )
131
132            m_entry_g_sidx = cstate_instance_to_global_idx[ instance_idx ][ m.start_state ]
133
134            # all of the transitions from the entry state need to have their probabilities adjusted
135            # since we will add a new transition to the connector
136            entry_tr_ids = [
137                i for i, tr in enumerate( transitions )
138                if tr.origin_state_idx == m_entry_g_sidx
139            ]
140
141            for i in entry_tr_ids :
142                transitions[ i ] = transitions[ i ].modified_deep_copy(
143                    prob = transitions[ i ].prob * component_residency_factor
144                )
145
146            # Entry state -> connector  (escape with remaining probability mass)
147            transitions.append( Transition(
148                origin_state_idx=m_entry_g_sidx,
149                target_state_idx=connector_g_sidx,
150                prob=( 1.0 - component_residency_factor ),
151                symbol_idx=symbol_idx_map[ exit_symbol ],
152                composition_depth=composition_depth 
153            ) )
154
155            # Connector -> entry state
156            # Probability is 1.0 / total_instances, mapping to unique enter symbol per instance
157            transitions.append( Transition(
158                origin_state_idx=connector_g_sidx,
159                target_state_idx=m_entry_g_sidx,
160                prob=( 1.0 / total_instances ),
161                symbol_idx=symbol_idx_map[ enter_symbols[ instance_idx ] ],
162                composition_depth=composition_depth 
163            ) )
164
165            # Increment global instance tracker
166            instance_idx += 1
167
168    # Remap and add the isomorphs
169    for g_idx, s in enumerate( states ) :
170
171        if g_idx not in global_idx_to_component_state :
172            continue  # skip the connector state
173
174        group_id = global_idx_to_group[ g_idx ]
175        c_state  = global_idx_to_component_state[ g_idx ]
176
177        isomorphs = set()
178        pseudo_isomorphs = set()
179
180        for c_iso in c_state.isomorphs :
181            # Isomorphs are scoped to the group
182            g_isos = cstate_name_to_global_idxs[ group_id ][ c_iso ]
183            for g_iso in g_isos :
184                isomorphs.add( states[ g_iso ].name )
185
186        for c_iso in c_state.pseudo_isomorphs :
187            g_isos = cstate_name_to_global_idxs[ group_id ][ c_iso ]
188            for g_iso in g_isos :
189                pseudo_isomorphs.add( states[ g_iso ].name )
190
191        states[ g_idx ] = states[ g_idx ].modified_deep_copy(
192            isomorphs=isomorphs,
193            pseudo_isomorphs=pseudo_isomorphs
194        )
195
196    ##############################################################
197
198    res = HMM(
199        states=states,
200        transitions=transitions,
201        alphabet=alphabet,
202        start_state=connector_g_sidx,
203        composition_depth=composition_depth
204    )
205
206    if not res.is_row_stochastic() :
207        raise Exception( "Start composition is not row stochastic" )
208
209    if not  res.is_unifilar() :
210        raise Exception( "Start composition is not unifilar" )
211
212    return res