amachine.am_create.am_structured_isomorphic_permutations_of

View Source

  1from typing import Literal
  2import warnings 
  3import math
  4import itertools
  5
  6import numpy as np
  7
  8from ..am_hmm    import HMM
  9from ..am_random import ( 
 10    resolve_rng,
 11    unique_random_permutations,
 12    unique_random_combinations
 13)
 14
 15from ..am_random import resolve_rng
 16
 17from ..am_vocabulary import Vocabulary
 18from ..am_structured_symbol_set import StructuredSymbolSet
 19
 20from .am_isomorphic_to_with_category_permutations import isomorphic_to_with_category_permutations
 21
 22def structured_isomorphic_permutations_of(
 23    m : HMM,
 24    symbol_set : StructuredSymbolSet,
 25    n_permutations : int,
 26    permutation_group : Literal[ "cyclic_product", "symmetric" ] = "symmetric",
 27    selection_mode : Literal[ "sequential", "random", "cohesive" ] = "cohesive",
 28    invariant_categories : set[int] | None = None,
 29    np_rng : np.random.Generator | None = None ) -> list[HMM]:
 30
 31    np_rng = resolve_rng( np_rng )
 32
 33    if permutation_group not in { "cyclic_product", "symmetric" } :
 34        raise ValueError( f"Invalid permutation group {permutation_group}, options are cyclic_product or symmetric" )
 35
 36    if selection_mode not in { "sequential", "random", "cohesive"  } :
 37        raise ValueError( f"Invalid selection_mode {selection_mode} options are sequential, random, or cohesive" )
 38
 39    if invariant_categories is None :
 40        invariant_categories = set()
 41
 42    if n_permutations == 1 :
 43        return [ m ]
 44
 45    categories_to_permute = symbol_set.categories - invariant_categories
 46    ordered_categories = sorted(symbol_set.categories)
 47
 48    category_sizes = symbol_set.symbol_counts_by_category()
 49    category_symbols = symbol_set.category_symbols()
 50
 51    print( f"Creating {n_permutations} total isomorphic HMMs" )
 52
 53    m_isos = []
 54
 55    enter_symbol_pool = list( set( 
 56        Vocabulary.digits()
 57      + Vocabulary.letters_lower()
 58      + Vocabulary.letters_upper()
 59      + Vocabulary.greek_lower()
 60      + Vocabulary.greek_upper() ) - set( m.alphabet ) )
 61
 62    effective_sizes = [ 
 63        category_sizes[c] if c not in invariant_categories else 1 
 64        for c in ordered_categories
 65    ]
 66
 67    if permutation_group == "cyclic_product" :
 68
 69        def shifts_to_permutations( shift_sequence ) :
 70            permutation_sequence = []
 71            for shifts in shift_sequence :
 72                permutations = {}
 73                for c, shift in shifts.items() :
 74                    permutations[ c ] = {}
 75                    for s in category_symbols[ c ] :
 76                        idx = symbol_set.symbols.index( s )
 77                        new_idx = ( idx + shift ) % len( category_symbols[ c ] )
 78                        permutations[ c ][ s ] = category_symbols[ c ][ new_idx ]
 79                permutation_sequence.append( permutations )
 80            return permutation_sequence
 81
 82        if selection_mode == "sequential" :
 83            shift_sequence = [ 
 84                { 
 85                    c : i if c in categories_to_permute else 0
 86                    for c in ordered_categories 
 87                }
 88                for i in range( 1, n_permutations )
 89            ]
 90
 91            permutation_sequence = shifts_to_permutations( shift_sequence )
 92        
 93        elif selection_mode == "random": 
 94
 95            combinations = unique_random_combinations( 
 96                bin_sizes=effective_sizes,
 97                n_samples=n_permutations-1,
 98                np_rng=np_rng
 99            )
100
101            shift_sequence = [
102                {  c : p[c] for c in ordered_categories }
103                for p in combinations
104            ]
105
106            permutation_sequence = shifts_to_permutations( shift_sequence )
107
108        elif selection_mode == "cohesive" :
109
110            pool_size = min( math.prod( effective_sizes)-1, 1_000_000 )
111
112            pool = unique_random_combinations( 
113                bin_sizes=effective_sizes,
114                n_samples=pool_size,
115                np_rng=np_rng
116            )
117
118            bigrams = []
119            for tra in  m.transitions :
120                for trb in m.transitions :
121                    if tra == trb :
122                        continue
123                    if tra.target_state_idx == trb.origin_state_idx :
124                        bigrams.append( ( 
125                            m.alphabet[ tra.symbol_idx ], 
126                            m.alphabet[ trb.symbol_idx ] ) )
127
128            category_symbols = symbol_set.category_symbols()
129
130            scored_combinations = []
131            for combination in pool :
132                score = 0
133                for a, b in bigrams :
134
135                    c_a = symbol_set.symbol_categories[ a ]
136                    c_b = symbol_set.symbol_categories[ b ]
137
138                    cat_a = category_symbols[ c_a ]
139                    cat_b = category_symbols[ c_b ]
140
141                    new_a = cat_a[ ( cat_a.index( a ) + combination[ c_a ] ) % len( cat_a ) ]
142                    new_b = cat_b[ ( cat_b.index( b ) + combination[ c_b ] ) % len( cat_b ) ]
143
144                    score += symbol_set.symbol_cohesion[ 
145                        symbol_set.symbols.index( new_a ), 
146                        symbol_set.symbols.index( new_b ) 
147                    ]
148                    
149                scored_combinations.append( ( score, combination ) )
150        
151            scored_combinations = sorted( scored_combinations, reverse=True )
152            scored_combinations = scored_combinations[ 0 : n_permutations-1 ]
153
154            shift_sequence = [
155                {  c : p[c] for c in ordered_categories }
156                for score, p in scored_combinations
157            ]
158
159            permutation_sequence = shifts_to_permutations( shift_sequence )
160
161    elif permutation_group == "symmetric" :
162
163
164        if selection_mode == "sequential":
165            
166            category_iters = []
167            for c in ordered_categories:
168                if c in categories_to_permute:
169                    category_iters.append(itertools.permutations(range(len(category_symbols[c]))))
170                else:
171                    category_iters.append([tuple(range(len(category_symbols[c])))])
172            
173            product_iter = itertools.product(*category_iters)
174            
175            try:
176                next(product_iter)
177            except StopIteration:
178                pass 
179                
180            permutation_sequence = []
181            for _ in range(n_permutations-1):
182                try:
183                    indices_tuple = next(product_iter)
184                except StopIteration:
185                    warnings.warn(f"Requested {n_permutations} permutations, but only {len(permutation_sequence)} symmetric combinations asides from identity exist.")
186                    break
187    
188                permutations = {}
189                for c, perm_indices in enumerate(indices_tuple):
190                    permutations[c] = {}
191                    for old_idx, new_idx in enumerate(perm_indices):
192                        old_sym = category_symbols[c][old_idx]
193                        new_sym = category_symbols[c][new_idx]
194                        permutations[c][old_sym] = new_sym
195                        
196                permutation_sequence.append(permutations)
197
198        elif selection_mode == "random":
199            
200            pool_size = min( math.prod( effective_sizes)-1, 1_000_000 )
201
202            index_permutation_sequence = unique_random_permutations( 
203                bin_sizes=effective_sizes,
204                n_samples=pool_size,
205                np_rng=np_rng )
206
207            permutation_sequence = [
208                {
209                    c: {
210                        category_symbols[c][old_idx] : category_symbols[c][new_idx]
211                        for old_idx, new_idx in index_perm.items()
212                    }
213                    for c, index_perm in index_permutation.items()
214                }
215                for index_permutation in index_permutation_sequence
216            ]
217
218        elif selection_mode == "cohesive":
219
220            bigram_indices = []
221            for tra in m.transitions:
222                for trb in m.transitions:
223                    if tra != trb and tra.target_state_idx == trb.origin_state_idx:
224                        c_a = symbol_set.symbol_categories[ m.alphabet[tra.symbol_idx] ]
225                        c_b = symbol_set.symbol_categories[ m.alphabet[trb.symbol_idx] ]
226                        
227                        # Store the category and the original index within that category
228                        orig_idx_a = category_symbols[c_a].index( m.alphabet[tra.symbol_idx] )
229                        orig_idx_b = category_symbols[c_b].index( m.alphabet[trb.symbol_idx] )
230                        
231                        bigram_indices.append( (c_a, orig_idx_a, c_b, orig_idx_b) )
232
233            def calculate_score(perm_state):
234                score = 0
235                for c_a, idx_a, c_b, idx_b in bigram_indices:
236
237                    new_sym_a = category_symbols[c_a][ perm_state[c_a][idx_a] ]
238                    new_sym_b = category_symbols[c_b][ perm_state[c_b][idx_b] ]
239                    
240                    global_idx_a = symbol_set.symbols.index(new_sym_a)
241                    global_idx_b = symbol_set.symbols.index(new_sym_b)
242                    
243                    score += symbol_set.symbol_cohesion[global_idx_a, global_idx_b]
244                return score
245
246            permutation_sequence = []
247            
248            # Hill Climbing Loop
249            for _ in range(n_permutations-1):
250                
251                # Start with a random permutation state for each category
252                current_perm = {
253                    c: list(np_rng.permutation(len(category_symbols[c]))) if c in categories_to_permute else list(range(len(category_symbols[c])))
254                    for c in ordered_categories
255                }
256                
257                current_score = calculate_score(current_perm)
258                improved = True
259                
260                # Keep optimizing until we hit a local maximum
261                while improved:
262                    improved = False
263                    
264                    # Try swapping every possible pair in every active category
265                    for c in categories_to_permute:
266                        n_sym = len(category_symbols[c])
267                        for i in range(n_sym):
268                            for j in range(i + 1, n_sym):
269                                
270                                # Perform swap
271                                current_perm[c][i], current_perm[c][j] = current_perm[c][j], current_perm[c][i]
272                                
273                                new_score = calculate_score(current_perm)
274                                
275                                if new_score > current_score:
276                                    current_score = new_score
277                                    improved = True
278                                else:
279                                    # Revert swap if it didn't improve the score
280                                    current_perm[c][i], current_perm[c][j] = current_perm[c][j], current_perm[c][i]
281
282                # Translate the optimized integer lists back into the expected dictionary format
283                final_permutation = {}
284                for c in ordered_categories :
285                    final_permutation[c] = {
286                        category_symbols[c][old_idx] : category_symbols[c][new_idx]
287                        for old_idx, new_idx in enumerate(current_perm[c])
288                    }
289                    
290                permutation_sequence.append(final_permutation)
291
292    for idx, permutations in enumerate( permutation_sequence ) :
293
294        m_iso = isomorphic_to_with_category_permutations(
295            m=m,
296            symbol_set=symbol_set,
297            permutations=permutations,
298            decorator=f"@{idx}" 
299        )
300
301        m_isos.append( m_iso )
302
303    all_machines = [ m ] + m_isos
304
305    for ma in all_machines :
306        for mb in all_machines :
307            if ma != mb :
308                for j, state in enumerate( ma.states ) : 
309                    ma.states[ j ].add_isomorph( mb.states[ j ].name )
310                    mb.states[ j ].add_isomorph( ma.states[ j ].name )
311
312    return all_machines

def structured_isomorphic_permutations_of( m: amachine.am_hmm.HMM, symbol_set: amachine.am_structured_symbol_set.StructuredSymbolSet, n_permutations: int, permutation_group: Literal['cyclic_product', 'symmetric'] = 'symmetric', selection_mode: Literal['sequential', 'random', 'cohesive'] = 'cohesive', invariant_categories: set[int] | None = None, np_rng: numpy.random._generator.Generator | None = None) -> list[amachine.am_hmm.HMM]: View Source

 23def structured_isomorphic_permutations_of(
 24    m : HMM,
 25    symbol_set : StructuredSymbolSet,
 26    n_permutations : int,
 27    permutation_group : Literal[ "cyclic_product", "symmetric" ] = "symmetric",
 28    selection_mode : Literal[ "sequential", "random", "cohesive" ] = "cohesive",
 29    invariant_categories : set[int] | None = None,
 30    np_rng : np.random.Generator | None = None ) -> list[HMM]:
 31
 32    np_rng = resolve_rng( np_rng )
 33
 34    if permutation_group not in { "cyclic_product", "symmetric" } :
 35        raise ValueError( f"Invalid permutation group {permutation_group}, options are cyclic_product or symmetric" )
 36
 37    if selection_mode not in { "sequential", "random", "cohesive"  } :
 38        raise ValueError( f"Invalid selection_mode {selection_mode} options are sequential, random, or cohesive" )
 39
 40    if invariant_categories is None :
 41        invariant_categories = set()
 42
 43    if n_permutations == 1 :
 44        return [ m ]
 45
 46    categories_to_permute = symbol_set.categories - invariant_categories
 47    ordered_categories = sorted(symbol_set.categories)
 48
 49    category_sizes = symbol_set.symbol_counts_by_category()
 50    category_symbols = symbol_set.category_symbols()
 51
 52    print( f"Creating {n_permutations} total isomorphic HMMs" )
 53
 54    m_isos = []
 55
 56    enter_symbol_pool = list( set( 
 57        Vocabulary.digits()
 58      + Vocabulary.letters_lower()
 59      + Vocabulary.letters_upper()
 60      + Vocabulary.greek_lower()
 61      + Vocabulary.greek_upper() ) - set( m.alphabet ) )
 62
 63    effective_sizes = [ 
 64        category_sizes[c] if c not in invariant_categories else 1 
 65        for c in ordered_categories
 66    ]
 67
 68    if permutation_group == "cyclic_product" :
 69
 70        def shifts_to_permutations( shift_sequence ) :
 71            permutation_sequence = []
 72            for shifts in shift_sequence :
 73                permutations = {}
 74                for c, shift in shifts.items() :
 75                    permutations[ c ] = {}
 76                    for s in category_symbols[ c ] :
 77                        idx = symbol_set.symbols.index( s )
 78                        new_idx = ( idx + shift ) % len( category_symbols[ c ] )
 79                        permutations[ c ][ s ] = category_symbols[ c ][ new_idx ]
 80                permutation_sequence.append( permutations )
 81            return permutation_sequence
 82
 83        if selection_mode == "sequential" :
 84            shift_sequence = [ 
 85                { 
 86                    c : i if c in categories_to_permute else 0
 87                    for c in ordered_categories 
 88                }
 89                for i in range( 1, n_permutations )
 90            ]
 91
 92            permutation_sequence = shifts_to_permutations( shift_sequence )
 93        
 94        elif selection_mode == "random": 
 95
 96            combinations = unique_random_combinations( 
 97                bin_sizes=effective_sizes,
 98                n_samples=n_permutations-1,
 99                np_rng=np_rng
100            )
101
102            shift_sequence = [
103                {  c : p[c] for c in ordered_categories }
104                for p in combinations
105            ]
106
107            permutation_sequence = shifts_to_permutations( shift_sequence )
108
109        elif selection_mode == "cohesive" :
110
111            pool_size = min( math.prod( effective_sizes)-1, 1_000_000 )
112
113            pool = unique_random_combinations( 
114                bin_sizes=effective_sizes,
115                n_samples=pool_size,
116                np_rng=np_rng
117            )
118
119            bigrams = []
120            for tra in  m.transitions :
121                for trb in m.transitions :
122                    if tra == trb :
123                        continue
124                    if tra.target_state_idx == trb.origin_state_idx :
125                        bigrams.append( ( 
126                            m.alphabet[ tra.symbol_idx ], 
127                            m.alphabet[ trb.symbol_idx ] ) )
128
129            category_symbols = symbol_set.category_symbols()
130
131            scored_combinations = []
132            for combination in pool :
133                score = 0
134                for a, b in bigrams :
135
136                    c_a = symbol_set.symbol_categories[ a ]
137                    c_b = symbol_set.symbol_categories[ b ]
138
139                    cat_a = category_symbols[ c_a ]
140                    cat_b = category_symbols[ c_b ]
141
142                    new_a = cat_a[ ( cat_a.index( a ) + combination[ c_a ] ) % len( cat_a ) ]
143                    new_b = cat_b[ ( cat_b.index( b ) + combination[ c_b ] ) % len( cat_b ) ]
144
145                    score += symbol_set.symbol_cohesion[ 
146                        symbol_set.symbols.index( new_a ), 
147                        symbol_set.symbols.index( new_b ) 
148                    ]
149                    
150                scored_combinations.append( ( score, combination ) )
151        
152            scored_combinations = sorted( scored_combinations, reverse=True )
153            scored_combinations = scored_combinations[ 0 : n_permutations-1 ]
154
155            shift_sequence = [
156                {  c : p[c] for c in ordered_categories }
157                for score, p in scored_combinations
158            ]
159
160            permutation_sequence = shifts_to_permutations( shift_sequence )
161
162    elif permutation_group == "symmetric" :
163
164
165        if selection_mode == "sequential":
166            
167            category_iters = []
168            for c in ordered_categories:
169                if c in categories_to_permute:
170                    category_iters.append(itertools.permutations(range(len(category_symbols[c]))))
171                else:
172                    category_iters.append([tuple(range(len(category_symbols[c])))])
173            
174            product_iter = itertools.product(*category_iters)
175            
176            try:
177                next(product_iter)
178            except StopIteration:
179                pass 
180                
181            permutation_sequence = []
182            for _ in range(n_permutations-1):
183                try:
184                    indices_tuple = next(product_iter)
185                except StopIteration:
186                    warnings.warn(f"Requested {n_permutations} permutations, but only {len(permutation_sequence)} symmetric combinations asides from identity exist.")
187                    break
188    
189                permutations = {}
190                for c, perm_indices in enumerate(indices_tuple):
191                    permutations[c] = {}
192                    for old_idx, new_idx in enumerate(perm_indices):
193                        old_sym = category_symbols[c][old_idx]
194                        new_sym = category_symbols[c][new_idx]
195                        permutations[c][old_sym] = new_sym
196                        
197                permutation_sequence.append(permutations)
198
199        elif selection_mode == "random":
200            
201            pool_size = min( math.prod( effective_sizes)-1, 1_000_000 )
202
203            index_permutation_sequence = unique_random_permutations( 
204                bin_sizes=effective_sizes,
205                n_samples=pool_size,
206                np_rng=np_rng )
207
208            permutation_sequence = [
209                {
210                    c: {
211                        category_symbols[c][old_idx] : category_symbols[c][new_idx]
212                        for old_idx, new_idx in index_perm.items()
213                    }
214                    for c, index_perm in index_permutation.items()
215                }
216                for index_permutation in index_permutation_sequence
217            ]
218
219        elif selection_mode == "cohesive":
220
221            bigram_indices = []
222            for tra in m.transitions:
223                for trb in m.transitions:
224                    if tra != trb and tra.target_state_idx == trb.origin_state_idx:
225                        c_a = symbol_set.symbol_categories[ m.alphabet[tra.symbol_idx] ]
226                        c_b = symbol_set.symbol_categories[ m.alphabet[trb.symbol_idx] ]
227                        
228                        # Store the category and the original index within that category
229                        orig_idx_a = category_symbols[c_a].index( m.alphabet[tra.symbol_idx] )
230                        orig_idx_b = category_symbols[c_b].index( m.alphabet[trb.symbol_idx] )
231                        
232                        bigram_indices.append( (c_a, orig_idx_a, c_b, orig_idx_b) )
233
234            def calculate_score(perm_state):
235                score = 0
236                for c_a, idx_a, c_b, idx_b in bigram_indices:
237
238                    new_sym_a = category_symbols[c_a][ perm_state[c_a][idx_a] ]
239                    new_sym_b = category_symbols[c_b][ perm_state[c_b][idx_b] ]
240                    
241                    global_idx_a = symbol_set.symbols.index(new_sym_a)
242                    global_idx_b = symbol_set.symbols.index(new_sym_b)
243                    
244                    score += symbol_set.symbol_cohesion[global_idx_a, global_idx_b]
245                return score
246
247            permutation_sequence = []
248            
249            # Hill Climbing Loop
250            for _ in range(n_permutations-1):
251                
252                # Start with a random permutation state for each category
253                current_perm = {
254                    c: list(np_rng.permutation(len(category_symbols[c]))) if c in categories_to_permute else list(range(len(category_symbols[c])))
255                    for c in ordered_categories
256                }
257                
258                current_score = calculate_score(current_perm)
259                improved = True
260                
261                # Keep optimizing until we hit a local maximum
262                while improved:
263                    improved = False
264                    
265                    # Try swapping every possible pair in every active category
266                    for c in categories_to_permute:
267                        n_sym = len(category_symbols[c])
268                        for i in range(n_sym):
269                            for j in range(i + 1, n_sym):
270                                
271                                # Perform swap
272                                current_perm[c][i], current_perm[c][j] = current_perm[c][j], current_perm[c][i]
273                                
274                                new_score = calculate_score(current_perm)
275                                
276                                if new_score > current_score:
277                                    current_score = new_score
278                                    improved = True
279                                else:
280                                    # Revert swap if it didn't improve the score
281                                    current_perm[c][i], current_perm[c][j] = current_perm[c][j], current_perm[c][i]
282
283                # Translate the optimized integer lists back into the expected dictionary format
284                final_permutation = {}
285                for c in ordered_categories :
286                    final_permutation[c] = {
287                        category_symbols[c][old_idx] : category_symbols[c][new_idx]
288                        for old_idx, new_idx in enumerate(current_perm[c])
289                    }
290                    
291                permutation_sequence.append(final_permutation)
292
293    for idx, permutations in enumerate( permutation_sequence ) :
294
295        m_iso = isomorphic_to_with_category_permutations(
296            m=m,
297            symbol_set=symbol_set,
298            permutations=permutations,
299            decorator=f"@{idx}" 
300        )
301
302        m_isos.append( m_iso )
303
304    all_machines = [ m ] + m_isos
305
306    for ma in all_machines :
307        for mb in all_machines :
308            if ma != mb :
309                for j, state in enumerate( ma.states ) : 
310                    ma.states[ j ].add_isomorph( mb.states[ j ].name )
311                    mb.states[ j ].add_isomorph( ma.states[ j ].name )
312
313    return all_machines