amachine.am_create.am_structured_isomorphic_rotation

  1from collections import defaultdict
  2import copy
  3import random
  4from typing import Literal
  5import warnings 
  6import math
  7
  8import numpy as np
  9
 10from ..am_hmm          import HMM
 11from ..am_causal_state import CausalState
 12from ..am_transition   import Transition
 13
 14from ..am_random import exp_uniform_blend, resolve_rng
 15from ..am_vocabulary import Vocabulary
 16from ..am_structured_symbol_set import StructuredSymbolSet
 17
 18from .am_random_structured import random_structured
 19from .am_isomorphic_to_with_category_rotation import isomorphic_to_with_category_rotation
 20from .am_star_join import star_join
 21
 22def get_unique_random_combinations(
 23    category_sizes,
 24    n_samples: int,
 25    rng: np.random.Generator,
 26) -> np.ndarray:
 27
 28    sizes = tuple(map(int, category_sizes))
 29    total = math.prod(sizes)
 30
 31    if n_samples > total:
 32        raise ValueError(
 33            f"Cannot draw {n_samples} unique samples: "
 34            f"only {total} combinations exist."
 35        )
 36
 37    if total < 2**63:
 38        flat = rng.choice( range( 1, total ), size=n_samples, replace=False )
 39        
 40        return np.stack(np.unravel_index(flat, shape=sizes), axis=1).tolist()
 41
 42    else:
 43        seen: set[tuple] = set()
 44        rows: list[tuple] = []
 45        while len(rows) < n_samples:
 46            combo = tuple(int(rng.integers(0, s)) for s in sizes)
 47            if combo not in seen:
 48                seen.add(combo)
 49                rows.append(combo)
 50
 51        return np.array(rows, dtype=np.intp).tolist()
 52
 53def structured_isomorphic_rotation(
 54    isoclass_name : int,
 55    n_states : int,
 56    symbol_set : StructuredSymbolSet,
 57    connectedness : float,
 58    randomness : float,
 59    star_joined : bool,
 60    n_rotations : int,
 61    rotation_mode : Literal[ "sequential", "cohesive", "random" ] = "sequential",
 62    mode_residency_factor : float | None = None,
 63    exclude_categories : set[int] | None = None,
 64    max_search_time : float = 120.0,
 65    n_search_workers = 3,
 66    random_seed : int | None = None
 67) -> HMM | list[HMM]:
 68
 69    if exclude_categories is None :
 70        exclude_categories = set()
 71
 72    if star_joined and mode_residency_factor is None :
 73        raise ValueError( "star_join requires mode_residency factor" )
 74
 75    if rotation_mode not in { "sequential", "cohesive", "random"  } :
 76        raise ValueError( "Invalid rotation mode" )
 77
 78    if random_seed is not None:
 79        py_rng = random.Random(random_seed)
 80        np_rng = np.random.default_rng(random_seed)
 81    else:
 82        py_rng = random
 83        np_rng = resolve_rng(None)
 84
 85    m = random_structured(
 86        n_states=n_states,
 87        symbol_set=symbol_set,
 88        connectedness=connectedness,
 89        randomness=randomness,
 90        ensure_strongly_connected=True,
 91        ensure_minimal=True,
 92        max_search_time=max_search_time,
 93        n_search_workers=n_search_workers,
 94        random_seed=random_seed
 95    )
 96
 97    # Collapse to the largest recurrent subgraph
 98    m.collapse_to_largest_strongly_connected_subgraph()
 99
100    # Minimize the machine -> epsilon-machine.
101    m.minimize()
102
103    categories_to_rotate = symbol_set.categories - exclude_categories
104
105    category_sizes = symbol_set.symbol_counts_by_category()
106    n_categories = len( category_sizes )
107
108    m.isoclass = isoclass_name
109    full_orbit_rotations = math.lcm( *( category_sizes[ i ] for i in categories_to_rotate ) ) - 1
110
111    print( f"Creating {n_rotations+1} total isomorphic HMMs" )
112
113    m_isos = []
114
115    enter_symbol_pool = list( set( 
116        Vocabulary.digits()
117      + Vocabulary.letters_lower()
118      + Vocabulary.letters_upper()
119      + Vocabulary.greek_lower()
120      + Vocabulary.greek_upper() ) - set( m.alphabet ) )
121
122    if star_joined and len( enter_symbol_pool ) < n_rotations + 1 :
123        raise ValueError( "Not enough symbols left over to construct star join." )
124
125    effective_sizes = [ 
126        category_sizes[i] if i not in exclude_categories else 1 
127        for i in range(n_categories) 
128    ]
129
130    if rotation_mode == "sequential" :
131
132        shift_sequence = [ 
133            { 
134                c : i if c in categories_to_rotate else 0
135                for c in range( n_categories ) 
136            }
137            for i in range( 1, n_rotations+1 )
138        ]
139    
140    elif rotation_mode == "random": 
141
142        combinations = get_unique_random_combinations( 
143            category_sizes=effective_sizes,
144            n_samples=n_rotations,
145            rng=np_rng
146        )
147
148        shift_sequence = [
149            {  c : p[c] for c in range(n_categories) }
150            for p in combinations
151        ]
152
153    elif rotation_mode == "cohesive" :
154
155        pool_size = min( math.prod( effective_sizes)-1, 1_000_000 )
156
157        pool = get_unique_random_combinations( 
158            category_sizes=effective_sizes,
159            n_samples=pool_size,
160            rng=np_rng
161        )
162
163        bigrams = []
164        for tra in  m.transitions :
165            for trb in m.transitions :
166                if tra == trb :
167                    continue
168                if tra.target_state_idx == trb.origin_state_idx :
169                    bigrams.append( ( 
170                        m.alphabet[ tra.symbol_idx ], 
171                        m.alphabet[ trb.symbol_idx ] ) )
172            
173
174        s_cats = {}
175        for c in set( symbol_set.symbol_categories.values() ) :
176            s_cats[ c ] = sorted( [ 
177                y for y in m.alphabet 
178                if y in symbol_set.symbol_categories and symbol_set.symbol_categories[y] == c 
179            ] )
180
181        scored_combinations = []
182        for combination in pool :
183            score = 0
184            for a, b in bigrams :
185
186                c_a = symbol_set.symbol_categories[ a ]
187                c_b = symbol_set.symbol_categories[ b ]
188
189                cat_a = s_cats[ c_a ]
190                cat_b = s_cats[ c_b ]
191
192                new_a = cat_a[ ( cat_a.index( a ) + combination[ c_a ] ) % len( cat_a ) ]
193                new_b = cat_b[ ( cat_b.index( b ) + combination[ c_b ] ) % len( cat_b ) ]
194
195                score += symbol_set.symbol_cohesion[ 
196                    symbol_set.symbols.index( new_a ), 
197                    symbol_set.symbols.index( new_b ) 
198                ]
199                
200            scored_combinations.append( ( score, combination ) )
201    
202        scored_combinations = sorted( scored_combinations, reverse=True )
203        scored_combinations = scored_combinations[ 0 : n_rotations ]
204
205        shift_sequence = [
206            {  c : p[c] for c in range(n_categories) }
207            for score, p in scored_combinations
208        ]
209
210    for categories_shifts in shift_sequence :
211
212        cs = sorted( [ item for item in categories_shifts.items() ] )
213        cs = [ str(x[1]) for x in cs ]
214
215        m_iso = isomorphic_to_with_category_rotation(
216            m=m,
217            symbol_categories=symbol_set.symbol_categories,
218            categories_shifts=categories_shifts,
219            decorator=f"@{"-".join(cs)}" 
220        )
221
222        m_iso.isoclass = isoclass_name
223        m_isos.append( m_iso )
224
225    all_machines = [ m ] + m_isos
226
227    for ma in all_machines :
228        for mb in all_machines :
229            if ma != mb :
230                for j, state in enumerate( ma.states ) : 
231                    ma.states[ j ].add_isomorph( mb.states[ j ].name )
232                    mb.states[ j ].add_isomorph( ma.states[ j ].name )
233
234    if star_joined :
235
236        # Join them together
237        m_star = star_join(
238            exit_symbol='x', 
239            enter_symbols=enter_symbol_pool[ 0:len(all_machines) ],
240            machines=all_machines,
241            mode_residency_factor=mode_residency_factor 
242        )
243
244        return m_star
245    
246    else :
247        return all_machines