A second part of the package, under core//utility/, independent of Armatura
and Orca — none of it participates in the anomaly shield. Generic tools for JAX/NumPy
pipelines. Each is tested on its own (test/test_chunk.py, test_compiler.py, test_memory.py,
test_preset.py, test_tensor.py, test_noise.py, test_vector.py, test_profiler.py,
test_visualizer.py, test_logger.py, test_anwav.py, test_diagnostic.py, test_iodat.py,
test_resonance_search.py); every example below was run for real before being written down.
DynamicAICodegen compiles a list of operation names (relu, sigmoid, tanh, scale,
dropout, clip, l2_normalize, identity) into a single JIT-compiled JAX pipeline. Useful
when you want to describe a small transformation pipeline declaratively (as data, not as
hand-written JAX code) and still get one compiled kernel plus a gradient for free.
fromdense_armor.coreimportDynamicAICodegencodegen=DynamicAICodegen()ops=codegen.compile_pipeline(["relu","l2_normalize"])out=codegen.run_dynamic_pipeline([-2.0,3.0,-1.0,4.0],ops)# out -> [0. 0.6 0. 0.8]
DynamicAICodegen — compila ricette testuali in pipeline JAX eseguibili.
Le operazioni supportate (relu/sigmoid/tanh/scale/dropout/clip/l2_normalize)
sono eseguite con jax.lax.switch dentro un unico kernel JIT, e i tipi
sono forzati a float64 lato CPU prima di ogni esecuzione per evitare
eccezioni quando la pipeline viene salvata/ricaricata da file binario.
def__init__(self,seed:int=42)->None:"""seed — seme iniziale della chiave PRNG usata dalle istruzioni stocastiche (es. dropout)."""self.cmd_map=CMD_MAPself.base_key=jax.random.PRNGKey(seed)
defcompile_pipeline(self,text_instructions:list)->np.ndarray:""" Converte una lista di istruzioni testuali/tuple in matrice float64 di shape (N, 4): [cmd_id, p1, p2, reserved]. """compiled=[]forcmdintext_instructions:ifisinstance(cmd,tuple):name=str(cmd[0]).lower().strip()p1=float(cmd[1])iflen(cmd)>1else0.0p2=float(cmd[2])iflen(cmd)>2else0.0else:name=str(cmd).lower().strip()ifname=='dropout':p1,p2=0.8,0.0elifname=='clip':p1,p2=-1.0,1.0elifname=='scale':p1,p2=float(_PHI),float(_ALPHA)# <-- Sintonizzazione d'Asse Geometricaelse:p1,p2=0.0,0.0cmd_id=self.cmd_map.get(name,0)compiled.append([float(cmd_id),p1,p2,0.0])returnnp.array(compiled,dtype=np.float64)
defrun_dynamic_pipeline(self,input_data:np.ndarray,compiled_ops:np.ndarray,)->np.ndarray:"""Esegue la pipeline compilata in un singolo lax.scan JIT."""res_data,updated_key=_run_pipeline_jit(jnp.array(input_data,dtype=jnp.float64),jnp.array(compiled_ops,dtype=jnp.float64),self.base_key,)self.base_key=updated_keyreturnnp.array(res_data)
[ADVANCED ENGINE]: Suddivide la pipeline in blocchi ed esegue il chunking
interamente all'interno dell'acceleratore hardware senza colli di bottiglia CPU.
defrun_pipeline_with_chunking(self,input_data:np.ndarray,compiled_ops:np.ndarray,chunk_size:int=500,)->np.ndarray:""" [ADVANCED ENGINE]: Suddivide la pipeline in blocchi ed esegue il chunking interamente all'interno dell'acceleratore hardware senza colli di bottiglia CPU. """n_ops=len(compiled_ops)# Calcolo dei blocchi necessari preservando la conformazione statican_chunks=(n_ops+chunk_size-1)//chunk_sizetotal_slots=n_chunks*chunk_size# Allocazione della matrice di padding condizionatapadded_ops=np.zeros((total_slots,4),dtype=np.float64)padded_ops[:n_ops]=compiled_ops# Riorganizzazione geometrica in tensore 3D (N_Chunks x Chunk_Size x 4)structured_chunks=padded_ops.reshape(n_chunks,chunk_size,4)j_data=jnp.array(input_data,dtype=jnp.float64)j_chunks=jnp.array(structured_chunks,dtype=jnp.float64)res_data=_run_pipeline_chunked_jit(j_data,j_chunks,self.base_key)returnnp.array(res_data)
defcompute_gradients(self,input_data:np.ndarray,compiled_ops:np.ndarray,)->np.ndarray:"""Calcola i gradienti AD della loss, normalizzata da una costante fissa (_PHI_FOUR)."""j_input=jnp.array(input_data,dtype=jnp.float64)j_ops=jnp.array(compiled_ops,dtype=jnp.float64)grads=_grad_engine(j_input,j_ops,self.base_key)self.base_key=jax.random.split(self.base_key)[0]returnnp.array(grads)
defsave_compiled_pipeline(self,compiled_ops:np.ndarray,filename:str="compiled_recipe.npy",):"""Salva la matrice delle operazioni compilate in formato binario compresso .npy."""np.save(filename,compiled_ops)logger.info("Advanced Pipeline salvata con successo: '%s'",filename)
defload_compiled_pipeline(self,filename:str="compiled_recipe.npy",)->np.ndarray:"""Carica una matrice di operazioni precedentemente salvata."""ifnotos.path.exists(filename):raiseFileNotFoundError(f"File pipeline assente: '{filename}'")ops=np.load(filename)logger.info("Advanced Pipeline caricata con successo: '%s' shape=%s",filename,ops.shape)returnops
ImageChunker splits a large batch (or a long list of compiled operations) into fixed-size
blocks, and merges the results back. Useful when a batch doesn't fit in memory in one shot, or
when a long instruction list would otherwise force XLA to recompile every time its length
changes.
Segmentazione a blocchi (chunking) per batch di dati e per liste di
istruzioni compilate, per evitare ricompilazioni JIT quando cambia il
numero di elementi/istruzioni.
Divide/ricompone batch di dati e liste di istruzioni in blocchi a dimensione fissa.
Utile quando un batch è troppo grande per stare in memoria in un colpo
solo, o quando una lista di istruzioni compilate è troppo lunga per
essere eseguita senza far ricompilare XLA ad ogni cambio di lunghezza.
chunk_size — dimensione fissa di ogni blocco (batch o istruzioni).
defsplit_array(self,array:np.ndarray)->list:"""Spezza un array multidimensionale in una lista di sotto-chunk."""total_samples=array.shape[0]num_chunks=int(np.ceil(total_samples/self.chunk_size))chunks=[]forbinrange(num_chunks):start_idx=b*self.chunk_sizeend_idx=min(start_idx+self.chunk_size,total_samples)chunks.append(array[start_idx:end_idx])returnchunks
defmerge_chunks(self,chunks_list:list)->np.ndarray:"""Ricombina una lista di sotto-chunk in un unico array compatto."""ifnotchunks_list:returnnp.array([],dtype=np.float32)# Se l'input è monodimensionale flat, usa concatenate invece di vstackifchunks_list[0].ndim==1:returnnp.concatenate(chunks_list)returnnp.vstack(chunks_list)
Esegue le istruzioni del compilatore a blocchi fissi (chunk_size).
Impedisce a XLA di ricompilare la pipeline se cambia il numero di
istruzioni, delegando l'esecuzione di ogni blocco a
codegen_engine.run_pipeline_with_chunking.
defexecute_pipeline_chunked(self,codegen_engine,input_vector:np.ndarray,compiled_ops:list)->np.ndarray:"""Esegue le istruzioni del compilatore a blocchi fissi (chunk_size). Impedisce a XLA di ricompilare la pipeline se cambia il numero di istruzioni, delegando l'esecuzione di ogni blocco a ``codegen_engine.run_pipeline_with_chunking``. """output=jnp.array(input_vector,dtype=jnp.float64)# Spezza ed esegue la lista di comandi/operazioniforiinrange(0,len(compiled_ops),self.chunk_size):chunk_ops=compiled_ops[i:i+self.chunk_size]output=codegen_engine.run_pipeline_with_chunking(output,chunk_ops,chunk_size=self.chunk_size)returnnp.array(output,dtype=np.float64)
Sostituisce i marcatori -1.0 in un template di operazioni con parametri dinamici.
Ogni elemento di template_ops diverso da -1.0 viene ripetuto
com'è nelle 4 colonne dell'operazione patchata; ogni marcatore -1.0
viene sostituito, in ordine, col prossimo valore di
dynamic_parameters. Implementato con jax.lax.scan per
restare compatibile con JIT.
@staticmethoddefpatch_and_scan_parameters(template_ops:jnp.ndarray,dynamic_parameters:jnp.ndarray)->jnp.ndarray:"""Sostituisce i marcatori -1.0 in un template di operazioni con parametri dinamici. Ogni elemento di ``template_ops`` diverso da -1.0 viene ripetuto com'è nelle 4 colonne dell'operazione patchata; ogni marcatore -1.0 viene sostituito, in ordine, col prossimo valore di ``dynamic_parameters``. Implementato con ``jax.lax.scan`` per restare compatibile con JIT. """defpatch_single_op(carry:jnp.ndarray,op:jnp.ndarray)->tuple:"""Un passo di scan: se op e' un marcatore -1.0 lo sostituisce col prossimo parametro dinamico."""idx=carry# Slot parametrico attivo, da riempire col prossimo valore dinamicois_parametric=op==-1.0final_param=jnp.where(is_parametric,dynamic_parameters[idx],op)next_idx=jnp.where(is_parametric,idx+jnp.int32(1),idx)# Restituisce l'operazione patchata a basso livello XLApatched_op=jnp.array([op,op,op,final_param],dtype=jnp.float64)returnnext_idx,patched_op_,patched_compiled_ops=jax.lax.scan(patch_single_op,jnp.int32(0),template_ops)returnpatched_compiled_ops
UniversalMemoryGuard checks free RAM (and VRAM, if an NVIDIA GPU is present) before a
heavy allocation, and computes how many chunks a batch needs to fit safely. Useful as a guard
rail right before a large jax/numpy allocation you don't want to OOM on.
fromdense_armor.coreimportUniversalMemoryGuardguard=UniversalMemoryGuard(min_free_ram_percentage=0.10)guard.check_memory_safety()# raises MemoryPressureError if RAM is too low
UniversalMemoryGuard — controlla RAM e VRAM prima di ogni allocazione pesante.
MemoryPressureError — eccezione lanciata quando la memoria è insufficiente.
def__init__(self,min_free_ram_percentage:float=0.15,force_gc:bool=True,)->None:"""min_free_ram_percentage — soglia minima di RAM libera richiesta; force_gc — se True tenta un soft garbage-collect prima di bloccare."""self.min_free_ram=min_free_ram_percentageself.force_gc=force_gc
defcheck_memory_safety(self)->None:"""Verifica lo stato della RAM e della VRAM prima di allocazioni critiche."""vm=psutil.virtual_memory()free_pct=vm.available/vm.total# Soft GC se vicini alla sogliaifself.force_gcandfree_pct<(self.min_free_ram+0.10):gc.collect()ifHAS_JAX:try:jax.clear_caches()exceptException:# pulizia best-effort e non critica: gli interni di JAX# possono fallire in troppi modi diversi per elencarli,# ma non deve bloccare il check di sicurezza memoria --# loggato (non piu' silenzioso) per restare tracciabile.logger.debug("jax.clear_caches() fallito durante il soft GC",exc_info=True)vm=psutil.virtual_memory()free_pct=vm.available/vm.totaliffree_pct<self.min_free_ram:raiseMemoryPressureError(f"RAM insufficiente: {free_pct:.1%} disponibile — "f"richiesta minima: {self.min_free_ram:.1%}")# VRAM check (solo se JAX con backend GPU)ifHAS_JAX:try:fordevinjax.devices():ifdev.platform=="gpu":vram_free=self._get_gpu_free_memory_nvidia()ifvram_free<0.05:raiseMemoryPressureError(f"VRAM esaurita su {dev.device_kind}: "f"{vram_free:.1%} libera.")exceptMemoryPressureError:raiseexcept(RuntimeError,AttributeError):# query driver/dispositivo JAX fallita (es. driver GPU non# inizializzato correttamente): stesso principio del check# RAM, non e' un errore dell'utente -- si prosegue senza# bloccare su un dato VRAM che non si riesce a leggere.pass
defcalculate_optimal_chunks(self,total_items:int,item_size_bytes:int)->int:"""Calcola il partizionamento ottimale basato sulla RAM e sul sovraccarico XLA."""self.check_memory_safety()vm=psutil.virtual_memory()# [FIX XLA-PADDING]: Riduciamo la finestra allocabile al 40% per compensare # i buffer temporanei generati durante il tracciamento dei grafi staticisafe_allocatable_bytes=int(vm.available*0.40)total_size_bytes=total_items*item_size_bytesiftotal_size_bytes<=safe_allocatable_bytes:return1returnmax(math.ceil(total_size_bytes/safe_allocatable_bytes),1)
AIHardwareProfiler detects the host's CPU/RAM/JAX backend and computes a safe maximum
tensor size for it. Honest caveat: the RAM tiers behind max_tensor_dim (2048/4096/8192,
doubled on GPU/TPU) are a rough heuristic, not calibrated against anything specific to this
package -- treat it as a starting guess, not a guarantee.
StochasticAdversarialNoise injects synthetic noise (bitflip, dropout, gaussian blur) into
a tensor while preserving its norm. Honest caveat: this is a generic noise injector, not a real
adversarial-example generator -- for actually testing the shield's robustness, the attacks in
test/test_boundA.py–test_boundE.py
(PGD/BIM/MI-FGSM, Carlini-Wagner, DeepFool, Fourier) are the real, calibrated benchmark; this
module overlaps with that suite rather than adding to it.
AIHardwareProfiler — profila l'architettura host per soglie di carico ottimali.
StochasticAdversarialNoise — inietta perturbazioni avversariali su CPU e GPU.
@staticmethoddefinject_noise(data_vector:np.ndarray,noise_type:str,intensity:float,seed:int=42,)->np.ndarray:"""Applica alterazioni probabilistiche preservando la norma del tensore."""noise_type=noise_type.lower().strip()ifintensity<=0.0ornoise_type=="clean":returndata_vector# ── JAX path ifHAS_JAXandisinstance(data_vector,(jnp.ndarray,jax.Array)):key=jax.random.PRNGKey(seed)key,subkey=jax.random.split(key)trigger_mask=jax.random.uniform(subkey,shape=data_vector.shape)<intensityifnoise_type=="bitflip":output=jnp.where(trigger_mask,-data_vector,data_vector)elifnoise_type=="dropout_noise":output=jnp.where(trigger_mask,0.0,data_vector)elifnoise_type=="gaussian_blur":key,subkey2=jax.random.split(key)noise=jax.random.normal(subkey2,shape=data_vector.shape)*intensityoutput=data_vector+noiseelse:output=data_vectornorm=jnp.linalg.norm(output)returnjnp.where(norm>0,output/(norm+1e-15),output)# ── NumPy fallback output=np.array(data_vector,copy=True)rng=np.random.default_rng(seed)trigger_mask=rng.random(size=output.shape)<intensityifnoise_type=="bitflip":output=np.where(trigger_mask,-output,output)elifnoise_type=="dropout_noise":output=np.where(trigger_mask,0.0,output)elifnoise_type=="gaussian_blur":noise=rng.normal(0.0,intensity,size=output.shape)output=output+noisenorm=np.linalg.norm(output)returnoutput/(norm+1e-15)ifnorm>0elseoutput
PipelineProfiler measures JIT latency in microseconds, with the first (compilation)
call timed separately from steady-state calls. This is the module that caught a real bug:
DynamicAICodegen's kernels used to be re-defined (and re-jax.jit-wrapped) on every single
call, so they never reused XLA's compilation cache -- warm-up and steady-state timed almost
identically. Once fixed, the split is real: warm-up is 1700x+ slower than steady-state on a
small pipeline.
@staticmethoddefmeasure_microseconds(codegen_instance,input_data:np.ndarray,compiled_ops:np.ndarray,repetitions:int=100,)->dict:""" Misura latenza JIT della pipeline DynamicAICodegen. Returns ------- dict con chiavi: warmup_compilation_us — prima esecuzione (compilazione XLA) mean_execution_us — media a regime repetitions — numero di run """start_warmup=time.perf_counter_ns()warmup_res=codegen_instance.run_dynamic_pipeline(input_data,compiled_ops)_=jax.block_until_ready(warmup_res)warmup_us=(time.perf_counter_ns()-start_warmup)/1_000.0latencies=[]for_inrange(repetitions):t0=time.perf_counter_ns()res=codegen_instance.run_dynamic_pipeline(input_data,compiled_ops)_=jax.block_until_ready(res)latencies.append((time.perf_counter_ns()-t0)/1_000.0)return{"warmup_compilation_us":warmup_us,"mean_execution_us":float(np.mean(latencies)),"std_execution_us":float(np.std(latencies)),"min_execution_us":float(np.min(latencies)),"repetitions":repetitions,}
@staticmethoddefmeasure_stabilizer_microseconds(stabilizer_instance,raw_batch:np.ndarray,repetitions:int=100,)->dict:""" Misura latenza vmap del filtro AdaptiveSignalStabilizer. Returns ------- dict con chiavi: warmup_compilation_us — prima esecuzione (compilazione vmap) mean_execution_us — media a regime repetitions — numero di run """start_warmup=time.perf_counter_ns()warmup_res=stabilizer_instance.filter_batch_scenarios(raw_batch)_=jax.block_until_ready(warmup_res)warmup_us=(time.perf_counter_ns()-start_warmup)/1_000.0latencies=[]for_inrange(repetitions):t0=time.perf_counter_ns()res=stabilizer_instance.filter_batch_scenarios(raw_batch)_=jax.block_until_ready(res)latencies.append((time.perf_counter_ns()-t0)/1_000.0)return{"warmup_compilation_us":warmup_us,"mean_execution_us":float(np.mean(latencies)),"std_execution_us":float(np.std(latencies)),"min_execution_us":float(np.min(latencies)),"repetitions":repetitions,}
TensorVault is a small library of static (invert, identity, edge_detector, blend)
and parametric (scale_project, amplify, bias_shift) transformation matrices, with backend
(JAX/NumPy) and precision auto-detected. Honest caveat: these are tiny, fixed matrices (2x2 or
a 3-element kernel) -- writing one inline is a single line of code. The real value here is the
backend/precision auto-detection, not the matrix catalog itself.
TensorVault — custodisce matrici di trasformazione statiche e parametriche.
Rileva automaticamente il backend (JAX/NumPy) e la precisione (float32/float64).
def__init__(self)->None:"""Rileva il backend disponibile (JAX o NumPy) e la precisione attiva."""self.xp=jnpifHAS_JAXelsenp# Lettura sicura di jax_enable_x64 senza usare il vecchio metodo .get()is_x64=getattr(jax.config,"jax_enable_x64",False)ifHAS_JAXelseFalseself.dtype=self.xp.float64if(is_x64ornotHAS_JAX)elseself.xp.float32
defget_static_transform(self,name:str)->np.ndarray:"""Sintetizza e restituisce la matrice statica richiesta."""xp,dt=self.xp,self.dtypetransforms={'invert':lambda:xp.array([[0.,1.],[1.,0.]],dtype=dt),'identity':lambda:xp.eye(2,dtype=dt),'edge_detector':lambda:xp.array([-1.,2.,-1.],dtype=dt),'blend':lambda:xp.array([[0.5,0.5],[0.5,0.5]],dtype=dt),}key=name.lower()ifkeynotintransforms:raiseKeyError(f"Trasformazione statica '{name}' non disponibile. "f"Disponibili: {list(transforms.keys())}")returntransforms[key]()
defget_backend_info(self)->str:"""Descrizione leggibile del backend/precisione attivi (es. 'JAX / GPU Accelerato (32-bit)')."""precision="64-bit"ifself.dtypein(np.float64,getattr(jnp,"float64",None))else"32-bit"backend="JAX / GPU Accelerato"ifHAS_JAXelse"NumPy / CPU Standard"returnf"{backend} ({precision})"
ParametricScenarioSimulator runs parallel Monte Carlo simulations over time (via
jax.vmap), plus a stochastic decision collapse driven by a probability distribution. Honest
caveat: the per-step update (next_state = current_state * 0.95 + param * 0.05) is a fixed
exponential-moving-average weighting, not a configurable simulation model -- useful mainly if
that specific dynamic matches your scenario, not as a general-purpose simulator.
fromdense_armor.coreimportParametricScenarioSimulatorimportnumpyasnpsim=ParametricScenarioSimulator()result,collapsed=sim.collapse_decision(np.array([0.1,0.2,0.3,0.4]),target_idx=2)# result -> 0 or 1 (stochastic); collapsed -> the vector renormalized after the choice
BitwisePermutationEngine swaps elements of a combinatorial vector (a 2^n-sized space)
based on target/control bit masks. Honest caveat: each call performs exactly one
controlled-swap between one pair of indices -- a single primitive, not a general permutation
engine. Narrower than the name suggests.
ParametricScenarioSimulator — simulazioni Monte Carlo massive via JAX vmap.
BitwisePermutationEngine — manipolazione vettori combinatori via maschere di bit.
collapse_decision non modifica in-place l'array del chiamante: opera su
una copia interna e restituisce (result, collapsed_vector) — il
chiamante può ignorare il vettore collassato se non gli serve.
Manipolazione di vettori combinatori (spazio 2^n) via maschere di bit.
Scambia elementi in array multidimensionali in base a coppie di bit
target/control.
n_elements — numero di bit del vettore combinatorio (spazio 2^n_elements).
def__init__(self,n_elements:int):"""n_elements — numero di bit del vettore combinatorio (spazio 2^n_elements)."""self.n=n_elementsself.size=1<<n_elements# 2^N stati possibili
defapply_bitwise_swap(self,data:np.ndarray,target_bit:int,control_bit:int,)->np.ndarray:"""Permuta gli elementi del vettore in base a maschere binarie."""output=data.copy()t_stride=1<<(self.n-1-target_bit)c_stride=1<<(self.n-1-control_bit)foriinrange(self.size):if(i&c_stride)andnot(i&t_stride):idx_0=iidx_1=i+t_strideoutput[idx_0],output[idx_1]=data[idx_1],data[idx_0]returnoutput
defrun_parallel_scenarios(self,base_state:float,parameters_batch:np.ndarray,)->np.ndarray:""" Esegue in parallelo (vmap) tutti i batch con jax.lax.scan per l'asse temporale. Parameters ---------- base_state — stato scalare iniziale per tutti gli scenari parameters_batch — array (N_scenari, T_steps) di parametri temporali Returns ------- np.ndarray di shape (N_scenari, T_steps) """result=_parallel_engine(jnp.float64(base_state),jnp.asarray(parameters_batch))returnnp.array(result)
defcollapse_decision(self,distribution_vector:np.ndarray,target_idx:int,)->tuple:""" Collasso decisionale stocastico condizionato dalla distribuzione. FIX BUG: non modifica più l'array originale in-place. Opera su una copia interna. Returns ------- (result: int, collapsed_vector: np.ndarray) result — 0 o 1 (scelta stocastica) collapsed_vector — vettore normalizzato post-collasso """vec=np.array(distribution_vector,copy=True,dtype=np.float64)prob_0=np.sum(np.abs(vec[:target_idx]))prob_1=np.sum(np.abs(vec[target_idx:]))total=prob_0+prob_1iftotal<1e-12:raiseRuntimeError("Vettore decisionale a energia zero — impossibile calcolare la scelta.")prob_0/=totalprob_1/=totalresult=int(np.random.choice([0,1],p=[prob_0,prob_1]))zero_slot=1-resultifzero_slot==0:vec[:target_idx]=0.0else:vec[target_idx:]=0.0new_total=np.sum(np.abs(vec))ifnew_total>0:vec/=new_totalreturnresult,vec
SIGNAL_STABILIZER_PRESETS are 4 empirically-calibrated parameter sets
(balanced_v2, cifar10_best_v1, pure_1d_time_v1, cifar10_hardened_lyapunov) for
AdaptiveSignalStabilizer (Orca's Stage 1). Verified, not just declared: on the
same noisy series with an outlier, pure_1d_time_v1 (tuned for a more reactive regime) leaves
over 2x the residual variance of balanced_v2 -- the presets genuinely configure different
filtering behavior, not just different numbers that happen to look distinct.
MinimalConsoleFormatter / CompactJsonFormatter are two logging.Formatter
subclasses — one human-readable for the console, one compact JSON for a log file. Honest
caveat: fairly thin wrappers around logging.Formatter -- CompactJsonFormatter's structured
fields (module/filename/line number, one JSON object per event) are the main reason to reach
for this over writing a one-line formatter yourself.
importloggingfromdense_armor.core.loggerimportMinimalConsoleFormatterhandler=logging.StreamHandler()handler.setFormatter(MinimalConsoleFormatter())log=logging.getLogger("demo")log.addHandler(handler)log.setLevel(logging.INFO)log.info("esempio")# [13:07:27] [INFO] esempio
defformat(self,record:logging.LogRecord)->str:"""Formatta un log record come `[HH:MM:SS] [LEVEL] messaggio`."""timestamp=datetime.fromtimestamp(record.created).strftime("%H:%M:%S")returnf"[{timestamp}] [{record.levelname}] {record.getMessage()}"
defformat(self,record:logging.LogRecord)->str:"""Formatta un log record come riga JSON compatta (una per evento)."""log_payload={"timestamp":datetime.fromtimestamp(record.created).isoformat(),"level":record.levelname,"module":record.module,"filename":record.filename,"line_number":record.lineno,"message":record.getMessage(),"framework":"dense-armor",}ifrecord.exc_info:log_payload["exception"]=self.formatException(record.exc_info)returnjson.dumps(log_payload,ensure_ascii=False)
defget_json_file_logger(name:str="dense_armor")->logging.Logger:"""Logger JSON su file (`dense_armor.log`), niente output su console (vedi commento sotto). Non usato dal resto della libreria."""logger=logging.getLogger(name)ifnotlogger.handlers:logger.setLevel(logging.INFO)# niente console_handler: le carte girano ad ogni passo del train,# duplicare ogni riga anche su stdout inonda il terminale -- il# file resta la fonte per il dashboard/i log.file_handler=logging.FileHandler("dense_armor.log",encoding="utf-8")file_handler.setFormatter(CompactJsonFormatter())logger.addHandler(file_handler)returnlogger
AIEngineVisualizer exports a SHA-256-signed provenance archive (parameters, execution
environment, integrity hash) and plain-text trend reports comparing raw vs. filtered variance.
Useful when you need an auditable record of a run, not just its output.
fromdense_armor.coreimportAIEngineVisualizerviz=AIEngineVisualizer(output_dir=".")sha256=viz.export_provenance_archive([{"step":1,"value":0.5}],filename="archive.json")# sha256 -> "b40db71b3b16d081..." (64 hex chars, matches the hash written into archive.json)
defexport_provenance_archive(self,run_history:list,filename:str="ai_provenance_archive.json",)->str:""" Genera un archivio di tracciabilità scientifica con firma SHA-256. Returns ------- sha256_hash : str """filepath=os.path.join(self.output_dir,filename)provenance_payload={"metadata":{# FIX BUG: engine_signature non più hardcoded, usa la versione del package"engine_signature":self.ENGINE_SIGNATURE,"export_timestamp_utc":time.strftime('%Y-%m-%d %H:%M:%S',time.gmtime()),"execution_environment":{"os":platform.system(),"architecture":platform.machine(),"python":platform.python_version(),"hardware":{"cpu_cores_logical":psutil.cpu_count(logical=True),"total_ram_gb":round(psutil.virtual_memory().total/(1024**3),2),},},},"records":run_history,}raw_bytes=json.dumps(provenance_payload,sort_keys=True,indent=4).encode('utf-8')sha256=hashlib.sha256(raw_bytes).hexdigest()provenance_payload["metadata"]["integrity_sha256"]=sha256withopen(filepath,"w",encoding="utf-8")asf:json.dump(provenance_payload,f,indent=4)returnsha256
anwav(fpath) analyzes a WAV file: peak, RMS, estimated loudness (LUFS), crest factor,
with a plain-text compliance verdict. Useful as a quick sanity check on an audio file's levels.
fromdense_armor.utility.anwavimportanwavanwav("track.wav")# -> File : track.wav# -> Picco Massimo : -6.02 dBFS# ...# [VERDETTO STANDARD]:# CONFORME (Peak): Picco in sicurezza sotto i -1.0 dB.
defanwav(fpath:str)->None:"""Analizza il file wav verificando i parametri di picco e dinamica."""ifnotos.path.exists(fpath):print(f"[ERR] File {fpath} non trovato!")returnsrate,data=wavfile.read(fpath)ifnp.issubdtype(data.dtype,np.floating):scala=1.0elifdata.dtype==np.int16:scala=32768.0elifdata.dtype==np.int32:scala=2147483648.0else:scala=float(np.iinfo(data.dtype).max)+1.0audio=data.astype(np.float32)/scala# Calcolo parametri essenzialimxval=np.max(np.abs(audio))p_db=20*np.log10(mxval)ifmxval>0else-99.0rms=np.sqrt(np.mean(audio**2))r_db=20*np.log10(rms)ifrms>0else-99.0lufs=r_db+3.0crest=p_db-r_dbprint(f" -> File : {fpath}")print(f" -> Picco Massimo : {p_db:.2f} dBFS")print(f" -> Volume Medio RMS : {r_db:.2f} dBFS")print(f" -> Loudness (LUFS) : {lufs:.1f} LUFS")print(f" -> Fattore Cresta (Dinamica) : {crest:.2f} dB")print("-"*85)print("[VERDETTO STANDARD]:")ifp_db>-1.0:print(" AVVISO: Il picco supera i -1.0 dB. Rischio distorsione.")else:print(" CONFORME (Peak): Picco in sicurezza sotto i -1.0 dB.")iflufs>-7.0:print(" AVVISO: Volume molto spinto da Club.")eliflufs<-16.0:print(" AVVISO: Traccia troppo silenziosa.")else:print(" CONFORME (Loudness): Rispetta i target standard.")ifcrest<6.0:print(" AVVISO: Traccia troppo schiacciata. Manca impatto.")else:print(" CONFORME (Dinamica): Mantiene l'impatto analogico.")
diag(iorig, ifilt) compares two audio signals (file paths or NumPy arrays): structural
fidelity, removed energy, distortion peak. Useful for checking how much an audio filter/process
actually changed a signal, beyond just listening to it.
defdiag(iorig:Union[str,np.ndarray],ifilt:Union[str,np.ndarray])->Optional[Dict[str,float]]:"""Esegue un'analisi differenziale profonda accettando sia percorsi file (str) che array NumPy."""ifisinstance(iorig,str)andisinstance(ifilt,str):ifnotos.path.exists(iorig)ornotos.path.exists(ifilt):print("[ERR] Uno dei file audio non è presente.")returnsr1,d_ori=wavfile.read(iorig)sr2,d_flt=wavfile.read(ifilt)v_ori=d_ori.astype(np.float32)/32768.0v_flt=d_flt.astype(np.float32)/32768.0else:v_ori=iorig.astype(np.float32)/32768.0ifiorig.dtype!=np.float32elseiorig.copy()v_flt=ifilt.astype(np.float32)/32768.0ififilt.dtype!=np.float32elseifilt.copy()iflen(v_ori.shape)>1:v_ori=np.mean(v_ori,axis=1)iflen(v_flt.shape)>1:v_flt=np.mean(v_flt,axis=1)mlen=min(v_ori.shape[0],v_flt.shape[0])v_ori=v_ori[:mlen]v_flt=v_flt[:mlen]v_dff=v_ori-v_fltv_rem=float(np.var(v_dff))fdel=(1.0-(np.sum(v_dff**2)/np.sum(v_ori**2)))*100.0ifnp.sum(v_ori**2)>0else0.0pk_df=float(np.max(np.abs(v_dff)))iflen(v_dff)>0else0.0pk_db=20*np.log10(pk_df)ifpk_df>0else-99.0alter=float(np.mean(np.abs(v_dff)>0.05)*100.0)iflen(v_dff)>0else0.0print("="*85)print("[DIAGNOSTICA DIFFERENZIALE STEREO] RE-ALLINEAMENTO COMPLETATO")print("="*85)print(f" -> Indice Strutturale di Fedeltà : {fdel:.4f}% (Portante preservata)")print(f" -> Energia Totale Rimossa (Var) : {v_rem:.4e}")print(f" -> Picco di Distorsione Segato : {pk_db:.2f} dBFS (Transiente massimo)")print(f" -> Tasso Modulazione Reticolo : {alter:.2f}% (Campioni modificati)")print("-"*85)iffdel>99.5:print("[VERDETTO DIAG] INTERVENTO CHIRURGICO: Solo micro-fruscii rimossi.")eliffdel>=95.0:print("[VERDETTO DIAG] RESTAURO EQUILIBRATO: Ottimo bilanciamento inter-canale.")else:print("[VERDETTO DIAG] MUTAZIONE AGGRESSIVA STEREO: Il Test 2 ha riscritto lo spazio dinamico.")print("="*85)return{"fedelta":fdel,"energia_rimossa":v_rem,"picco_distorsione_db":pk_db,"tasso_modulazione":alter}
lodat(fpath, dname) reads a named tensor out of an HDF5 or NetCDF file. Useful as a thin,
uniform loader when a pipeline needs to accept either format without branching on the caller's
side.
deflodat(fpath:str,dname:str)->np.ndarray:""" Rileva l'estensione del file ed estrae il tensore di produzione garantendo la massima compatibilità di I/O. """ifnotos.path.exists(fpath):raiseFileNotFoundError(f"File non trovato: {fpath}")exten=os.path.splitext(fpath)[1].lower()ifextenin['.h5','.hdf5']:withh5py.File(fpath,'r')asf:data=np.array(f[dname])logger.info("Estratto HDF5: %s | Shape: %s",fpath,data.shape)returndataelifextenin['.nc','.netcdf']:withnetCDF4.Dataset(fpath,'r')asf:data=np.array(f.variables[dname][:])logger.info("Estratto NetCDF: %s | Shape: %s",fpath,data.shape)returndataelse:raiseValueError(f"Formato file non supportato: {exten}")
apply_fast_resonance(matrix, query) scores cosine similarity between a query vector and
each row of a matrix, modulated by apply_damping_blend (the same operator Orca's gating
uses). Verified, not just declared: the modulation is load-bearing, not decorative --
kappa (the damping weight) measurably changes the score (kappa=0 vs. kappa=1 differ well
beyond floating-point noise on the same inputs), so this is genuinely different from plain
cosine similarity, not a rebrand of it.
fromdense_armor.utility.resonance_searchimportapply_fast_resonanceimportnumpyasnprng=np.random.default_rng(0)db=rng.standard_normal((5,8)).astype(np.float32)query=db[2].copy()# an exact copy of row 2scores=apply_fast_resonance(db,query)# int(scores.argmax()) -> 2 (the matching row scores highest)
defapply_fast_resonance(matrix_np:np.ndarray,query_np:np.ndarray,kappa:float=0.86210,delta_eff:float=0.043410,stress_segnale:float=9.42194e-04)->np.ndarray:"""Punteggio di risonanza tra ogni riga di matrix_np e query_np, gestendo input vuoti/degeneri."""ifmatrix_npisNoneorquery_npisNone:returnnp.array([],dtype=np.float32)ifmatrix_np.size==0orquery_np.size==0:returnnp.zeros(len(matrix_np),dtype=np.float32)q=np.asarray(query_np,dtype=np.float32).squeeze()ifq.ndim==0orq.size==0:returnnp.zeros(len(matrix_np),dtype=np.float32)ifq.ndim>1:q=q.flatten()qn=np.linalg.norm(q)ifqn<1e-8:returnnp.zeros(len(matrix_np),dtype=np.float32)q=q/qnj_matrix=jnp.array(matrix_np,dtype=jnp.float32)j_query=jnp.array(q,dtype=jnp.float32)scores=_resonance_scores(j_matrix,j_query,float(kappa),float(delta_eff),float(stress_segnale))returnnp.array(scores,dtype=np.float32)
defsmoke_test()->bool:"""Auto-test rapido: True se apply_fast_resonance produce un risultato sensato su dati sintetici."""try:N,D=4,8rng=np.random.default_rng(42)m=rng.standard_normal((N,D)).astype(np.float32)q=rng.standard_normal(D).astype(np.float32)s=apply_fast_resonance(m,q)asserts.shape==(N,)assertnotnp.all(np.isnan(s))assertnotnp.all(s==0.0)returnTrueexceptException:# broad by design: uno smoke test deve catturare QUALUNQUE# fallimento (import, shape, NaN, assert) e ridurlo a True/False --# non sta nascondendo un bug, e' la sua funzione.returnFalse