Repository navigation
Expand file tree
/
Copy pathstructural_diffusion.py
More file actions
3983 lines (3520 loc) · 159 KB
/
Copy pathstructural_diffusion.py
File metadata and controls
3983 lines (3520 loc) · 159 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
r"""Structural diffusion and scoped diagnostics for the nodal EPI channel.
For a fixed, nonnegative adjacency W, let D contain its row strengths and
L_rw = I - D^-1 W, with a zero row at an isolated node. The canonical EPI
pressure is DeltaNFR_epi = -L_rw EPI. With the capacity vector nu_f, the
isolated channel therefore evolves as
dEPI/dt = -A EPI, A = diag(nu_f) L_rw.
The full pressure also includes phase, frequency and topology channels.
Wrapped circular phase averaging is nonlinear and is not this scalar
Laplacian identity. These diagnostics hold the graph and capacity fixed;
they do not integrate a changing canonical operator sequence.
Transport and spectral scope
----------------------------
For symmetric adjacency and a common positive capacity nu, each mode
decays at nu*lambda_k. A connected graph relaxes to a uniform field and
conserves sum(d_i*EPI_i). With disconnected components, stationarity need
not be globally uniform. With positive heterogeneous capacity the conserved
weights are d_i/nu_i and the decay rates are eigenvalues of A; replacing
capacity by its mean changes the dynamics. Zero capacity freezes a node
even when its pressure is nonzero. Directed transport uses the actual
nonsymmetric A; real eigenvalue parts describe asymptotic damping, without
certifying normality or absence of transient growth.
The symmetric normalized Laplacian provides an orthonormal geometry basis
only for symmetric adjacency. Its zero mode is proportional to sqrt(d)
on a connected component, corresponding to a uniform EPI field after the
degree-coordinate transformation. Finite dimension gives a finite spectrum;
a nodal-domain upper bound does not imply monotonic domain counts for every
graph or every basis of a degenerate eigenspace.
Adding a scalar reaction rate r gives real growth rates
r - Re(eigenvalues(A)). On a connected homogeneous network the first
nonuniform threshold is nu*lambda_2. This is a spatial-mode threshold,
not a bound on all evolution: the uniform mode already grows for any r>0.
The correspondence is a linear diagnostic, not a derivation of grammar U2
for arbitrary sequences.
Drift, waves and certificates
----------------------------
Under held pressure F the nodal equation is the first-order mobility law
q_dot = nu_f*F. The damped graph-wave model q_ddot + gamma*q_dot + L*q = 0
has a slow diffusion limit with mobility 1/gamma. The wave-to-diffusion
checks below concern that specified graph-wave model. They do not prove
that the isotropic Hamiltonian implemented by symplectic_substrate, or all
13 engine operators, induces that graph wave.
verify_structural_diffusion checks the canonical pressure on a replica and
samples the actual frozen-capacity flow. It reports global uniformity and
degree-weighted conservation separately from stationarity and the complete
left-nullspace invariants of A. Finite-time residuals are diagnostics, not
proofs of eventual convergence.
Random walks and currents
------------------------
P = I - L_rw is row-stochastic, with an absorbing self-transition at an
isolated node. For symmetric adjacency the degree distribution is
stationary; convergence of a discrete walk additionally requires
aperiodicity within an irreducible component. Resistance geometry requires
symmetric nonnegative conductance. Disconnected pairs have infinite
resistance and commute time; finite commute times use the volume of the
pair's component.
For symmetric W, edge currents J_ij = W_ij*(EPI_i-EPI_j) have divergence
(D-W)*EPI. At nodes with positive degree and capacity, the nodal continuity
balance is (d_i/nu_i)*dEPI_i/dt + div(J)_i = 0. This EPI-channel identity
is distinct from the tetrad charge and currents in physics.conservation.
The same conductance model gives the Dirichlet energy
E_D = (1/4)*sum_ij W_ij*(EPI_i-EPI_j)^2. Its gradient is (D-W)*EPI,
and the mobility is diag(nu_i/d_i), zero at isolates. Hence the frozen
EPI-only flow satisfies dE_D/dt = -sum_i mobility_i*gradient_i^2 <= 0.
compute_diffusion_energy reports this balance without evolving the graph.
This energy is distinct from the tetrad potential in physics.variational.
References within the implementation: dynamics.dnfr (canonical pressure),
directed_diffusion (nonsymmetric transport), symplectic_substrate (specified
Hamiltonian), and physics.conservation (tetrad diagnostics).
"""
from __future__ import annotations
import math
import struct
import sys
from collections.abc import Mapping
from dataclasses import dataclass, field
from fractions import Fraction
from functools import lru_cache
from typing import Any
from ..alias import get_attr
from ..constants.aliases import ALIAS_DNFR, ALIAS_EPI, ALIAS_VF
from ..mathematics._weight_normalization import normalize_weights
from ..mathematics.unified_numerical import np
from ..types import real_scalar_epi
from ..utils._structural_signature import (
proof_stamps_are_identical,
structural_proof_signature,
)
from ._conductance import ConductanceSnapshot, read_conductance
from ._exact_linear_algebra import exact_matrix_inverse as _exact_matrix_inverse
from ._exact_linear_algebra import (
exact_symmetric_semidefinite as _exact_symmetric_semidefinite,
)
from ._helpers import finite_real_scalar
__all__ = [
"DiffusionEnergyBalance",
"HeterogeneousDiffusionStabilityCertificate",
"SwitchingDiffusionStabilityCertificate",
"EulerRelaxationWindowDiagnostic",
"TimeVaryingDiffusionStabilityBound",
"StructuralDiffusionCertificate",
"OverdampedRegimeCertificate",
"OverdampedProjectionCertificate",
"UndampedLimitCertificate",
"DiscreteModeCertificate",
"StructuralStabilityCertificate",
"RandomWalkCertificate",
"StructuralFlowCertificate",
"structural_diffusion_operator",
"symmetric_normalized_laplacian",
"structural_field",
"structural_diffusivity",
"relaxation_spectrum",
"structural_frequency_rank",
"degree_weighted_total",
"compute_diffusion_energy",
"derive_time_varying_diffusion_stability_bound",
"diagnose_euler_relaxation_window",
"verify_heterogeneous_diffusion_stability",
"verify_switching_diffusion_stability",
"structural_eigenvalues",
"structural_eigenmodes",
"nodal_domain_count",
"compute_emergent_pulse",
"compute_nodal_pulse",
"dispersion_relation",
"instability_threshold",
"fiedler_partition",
"random_walk_matrix",
"stationary_distribution",
"effective_resistance",
"commute_time",
"structural_current",
"current_divergence",
"verify_structural_diffusion",
"verify_overdamped_regime",
"damped_wave_rates",
"verify_overdamped_projection",
"verify_undamped_limit",
"verify_discrete_modes",
"verify_structural_stability",
"verify_structural_random_walk",
"verify_structural_flow",
]
# Two binary64 significands make the rational bisection finer than any
# representable rate near a well-scaled endpoint. Stopping early can only make
# the returned lower bound more conservative.
_EXACT_QUOTIENT_BISECTION_STEPS = 2 * sys.float_info.mant_dig
# Refinement is optional because the inverse-norm endpoint is already a proof.
# Limiting exact LDL bisection to four quotient coordinates keeps larger graph
# certificates practical while retaining tight bounds in small theorem tests.
_EXACT_QUOTIENT_REFINEMENT_MAX_DIMENSION = 4
def _reject_boolean_numeric(value: Any, name: str) -> None:
"""Reject booleans before NumPy can coerce them to zero or one."""
if isinstance(value, (bool, np.bool_)):
raise ValueError(f"{name} must be numeric, not boolean")
def _readonly_float_array(value: Any) -> Any:
"""Return a detached binary64 array whose ordinary writes are disabled."""
result = np.array(value, dtype=float, copy=True)
result.setflags(write=False)
return result
def _finite_float_signature(value: Any) -> tuple[str, ...]:
"""Encode one finite array exactly enough for an in-memory proof stamp."""
array = np.asarray(value, dtype=float)
if not np.all(np.isfinite(array)):
raise ValueError("proof data must remain finite")
return tuple(float(item).hex() for item in array.flat)
def _fixed_flow_proof_stamp(
nodes: Any,
metric_weights: Any,
exact_gap: Fraction,
certified_rate: float,
is_certified: bool,
preserves_consensus_subspace: bool,
preserves_uniform_fixed_points: bool,
preserves_weighted_mean: bool,
) -> tuple[Any, ...]:
"""Snapshot the fields on which hybrid fixed-flow composition relies."""
return (
"fixed_heterogeneous_diffusion_v1",
structural_proof_signature(tuple(nodes)),
np.asarray(metric_weights).shape,
_finite_float_signature(metric_weights),
Fraction(exact_gap),
float(certified_rate).hex(),
bool(is_certified),
bool(preserves_consensus_subspace),
bool(preserves_uniform_fixed_points),
bool(preserves_weighted_mean),
)
def _switching_flow_proof_stamp(
nodes: Any,
regime_count: int,
reference_metric_weights: Any,
normalized_metric_weights: Any,
exact_gaps: Any,
certified_rates: Any,
certified_rate: float,
shares_exact_common_metric: bool,
supports_exact_theorem: bool,
preserves_consensus_by_regime: Any,
preserves_consensus_subspace: bool,
preserves_uniform_fixed_points_by_regime: Any,
preserves_uniform_fixed_points: bool,
preserves_weighted_mean_by_regime: Any,
preserves_weighted_mean: bool,
) -> tuple[Any, ...]:
"""Snapshot the fields on which hybrid switching composition relies."""
return (
"exact_common_metric_switching_diffusion_v1",
structural_proof_signature(tuple(nodes)),
int(regime_count),
np.asarray(reference_metric_weights).shape,
_finite_float_signature(reference_metric_weights),
np.asarray(normalized_metric_weights).shape,
_finite_float_signature(normalized_metric_weights),
tuple(Fraction(value) for value in exact_gaps),
np.asarray(certified_rates).shape,
_finite_float_signature(certified_rates),
float(certified_rate).hex(),
bool(shares_exact_common_metric),
bool(supports_exact_theorem),
tuple(bool(value) for value in preserves_consensus_by_regime),
bool(preserves_consensus_subspace),
tuple(bool(value) for value in preserves_uniform_fixed_points_by_regime),
bool(preserves_uniform_fixed_points),
tuple(bool(value) for value in preserves_weighted_mean_by_regime),
bool(preserves_weighted_mean),
)
def _fraction_lower_float(value: Fraction) -> float:
"""Largest nearby binary64 value that is known not to exceed ``value``."""
if value < 0:
raise ValueError("a lower-rounded fraction must be nonnegative")
try:
rounded = float(value)
except OverflowError:
return math.nextafter(float("inf"), 0.0)
if math.isinf(rounded):
return math.nextafter(rounded, 0.0)
if Fraction.from_float(rounded) > value:
rounded = math.nextafter(rounded, 0.0)
return rounded
def _fraction_upper_float(value: Fraction) -> float:
"""Smallest nearby binary64 value that is known not to be below ``value``."""
if value < 0:
raise ValueError("an upper-rounded fraction must be nonnegative")
try:
rounded = float(value)
except OverflowError:
return float("inf")
if math.isinf(rounded):
return rounded
if Fraction.from_float(rounded) < value:
rounded = math.nextafter(rounded, float("inf"))
return rounded
def _fraction_sqrt_upper_float(value: Fraction) -> float:
"""Return a proved binary64 upper bound on ``sqrt(value)``.
The search compares squared binary64 candidates with the rational input,
so the result does not rely on a directed-rounding guarantee from libm.
Positive binary64 values have the same order as their unsigned bit
patterns, which makes a fixed 63-step binary search sufficient.
"""
if value < 0:
raise ValueError("a square-root bound must be nonnegative")
if value == 0:
return 0.0
low_bits = 0
high_bits = 0x7FF0000000000000 # positive infinity
while low_bits + 1 < high_bits:
middle_bits = (low_bits + high_bits) // 2
candidate = struct.unpack(">d", middle_bits.to_bytes(8, byteorder="big"))[0]
if math.isinf(candidate) or Fraction.from_float(candidate) ** 2 >= value:
high_bits = middle_bits
else:
low_bits = middle_bits
return struct.unpack(">d", high_bits.to_bytes(8, byteorder="big"))[0]
def _ordered_nodes(G: Any) -> list:
"""Stable node ordering for the matrix representation."""
return list(G.nodes())
def _weighted_adjacency(G: Any, nodes: list | None = None) -> tuple[list, Any]:
"""One nonnegative adjacency convention, including parallel edges/loops."""
conductance = read_conductance(G, nodes)
return conductance.nodes, conductance.dense()
def _nodal_frequencies(G: Any, nodes: list | None = None) -> Any:
"""Read the actual capacity vector; zero capacity freezes a node."""
if nodes is None:
nodes = _ordered_nodes(G)
try:
raw = [
get_attr(
G.nodes[node], ALIAS_VF, 0.0, conv=lambda value: value, strict=True
)
for node in nodes
]
except (TypeError, ValueError, OverflowError) as exc:
raise ValueError(
"Structural frequency values must be finite real numbers"
) from exc
if any(isinstance(value, (bool, np.bool_)) for value in raw):
raise ValueError(
"Structural frequency must be a finite real scalar, not boolean"
)
try:
frequency = np.array(
[finite_real_scalar(value, "Structural frequency") for value in raw],
dtype=float,
)
except (TypeError, ValueError, OverflowError) as exc:
raise ValueError(
"Structural frequency values must be finite real numbers"
) from exc
if np.any(frequency < 0.0):
raise ValueError("Structural frequency must be finite and nonnegative")
return frequency
def structural_diffusion_operator(G: Any) -> tuple[list, Any]:
r"""Return the random-walk graph Laplacian L_rw = I − D⁻¹W.
This is the operator whose action on a field is exactly the canonical
ΔNFR ``neighbour-mean − self`` gradient: g = −L_rw·field. Built from
the (optionally weighted) adjacency; isolated nodes (degree 0) get a
zero row (no diffusion).
Finite effective edge weights are normalized in scaled coordinates, so
a raw row sum need not fit in a float. Raw degree-weighted quantities
retain their separate representability requirements.
Parameters
----------
G : TNFRGraph
Returns
-------
(nodes, L_rw) : tuple[list, np.ndarray]
The node ordering and the N×N random-walk Laplacian.
"""
conductance = read_conductance(G)
probability, scale, _ = conductance.normalization()
laplacian = conductance.dense(-probability)
diagonal = np.diag_indices(len(conductance.nodes))
laplacian[diagonal] += scale > 0.0
return conductance.nodes, laplacian
def symmetric_normalized_laplacian(
G: Any, nodes: list | None = None
) -> tuple[list, Any]:
r"""Return the symmetric normalized Laplacian L_sym = I − D^{-1/2} W D^{-1/2}.
For symmetric adjacency, L_sym shares the spectrum of the diffusion operator
L_rw = I − D⁻¹W (:func:`structural_diffusion_operator`) but is symmetric, so
it has an orthonormal eigenbasis and real eigenvalues — the canonical choice
for the relaxation spectrum (its λ₂ is the structural ``diffusion_gap``).
Isolated nodes (degree 0) get a zero row. Asymmetric adjacency is rejected:
it cannot be passed to a symmetric eigensolver. Directed damping rates are
available through :func:`relaxation_spectrum`.
Scaled conductance avoids overflowing raw row sums; square-root ratios
retain representable symmetric coefficients at very unequal strengths.
Parameters
----------
G : TNFRGraph
nodes : list, optional
Node ordering; defaults to the stable ``list(G.nodes())`` order.
Returns
-------
(nodes, L_sym) : tuple[list, np.ndarray]
The node ordering and the N×N symmetric normalized Laplacian.
"""
conductance = read_conductance(G, nodes, symmetric=True)
normalized, positive = conductance.symmetric_normalized_weights()
lap = conductance.dense(-normalized)
lap[np.diag_indices(len(conductance.nodes))] += positive
return conductance.nodes, lap
def _finite_scalar_epi(value: Any) -> float:
"""Read raw real EPI or an exact uniform-real BEPI embedding."""
is_bepi_representation = isinstance(value, Mapping) or all(
hasattr(value, attribute)
for attribute in ("f_continuous", "a_discrete", "x_grid")
)
if not is_bepi_representation:
return finite_real_scalar(value, "EPI")
try:
scalar = real_scalar_epi(value)
except (KeyError, TypeError, ValueError, OverflowError) as exc:
raise ValueError(
"EPI must be a finite real scalar or uniform real BEPI embedding"
) from exc
if scalar is None:
raise ValueError(
"EPI must be a finite real scalar or uniform real BEPI embedding"
)
return scalar
def structural_field(G: Any, nodes: list | None = None) -> Any:
r"""Return the strict scalar EPI field aligned with ``nodes``.
Raw finite real values and exact uniform-real BEPI embeddings share the
signed scalar channel. Nonuniform or complex BEPI elements are rejected
because their magnitude projection is not an equivalent diffusion state.
"""
if nodes is None:
nodes = _ordered_nodes(G)
raw = [
get_attr(G.nodes[node], ALIAS_EPI, 0.0, conv=lambda value: value, strict=True)
for node in nodes
]
return np.array([_finite_scalar_epi(value) for value in raw], dtype=float)
def structural_diffusivity(G: Any) -> float:
r"""Mean structural frequency, a descriptive capacity summary.
In ∂EPI/∂t = −νf·L_rw·EPI, νf plays the role of the diffusivity: the
larger the structural frequency, the faster the form spreads. This mean
is a diffusion coefficient only when all nodal frequencies are equal;
heterogeneous transport uses diag(νf)·L_rw, never mean(νf)·L_rw.
"""
vf = _nodal_frequencies(G)
return float(np.mean(vf)) if len(vf) else 0.0
def degree_weighted_total(G: Any) -> float:
r"""The row-strength-weighted total Σ_i d_i·EPI(i).
For symmetric adjacency and a common frequency this total is conserved.
For fixed positive heterogeneous frequencies the conserved weights are
d_i/νf_i instead; a directed graph requires the generator's left nullspace.
This helper keeps its literal weighted-total meaning in every case.
Nonfinite EPI or an unrepresentable final total raises ValueError.
Edge-based summation allows finite cancellation even when an intermediate
degree or product exceeds float range.
Under the stated symmetric homogeneous restriction this is an exact
**EPI-channel** conserved quantity. It is distinct from the Noether-like
diagnostic ``Q = Σ(Φ_s + K_φ)``
(:func:`tnfr.physics.conservation.compute_noether_charge`). Grammar U1–U6
does not generally conserve that diagnostic; its drift must be evaluated on
the supplied trajectory.
"""
conductance = read_conductance(G)
field = structural_field(G, conductance.nodes)
if not np.all(np.isfinite(field)):
raise ValueError("Structural transport requires finite scalar EPI")
total = conductance.weighted_total(field)
if not np.isfinite(total):
raise ValueError("Degree-weighted total exceeds finite floating-point range")
return total
def _read_edge_flux(G: Any) -> tuple[ConductanceSnapshot, Any, Any]:
"""Share edge differences between current, divergence and energy gradient.
Capacity is deliberately absent: constitutive current can remain nonzero
at a frozen node. Effective zero edges are removed before subtraction.
"""
conductance = read_conductance(G, symmetric=True)
field = structural_field(G, conductance.nodes)
if not np.all(np.isfinite(field)):
raise ValueError("Structural transport requires finite scalar EPI")
try:
with np.errstate(over="raise", under="raise", invalid="raise"):
difference = field[conductance.source] - field[conductance.target]
flux = conductance.weight * difference
except FloatingPointError as exc:
raise ValueError(
"Structural current is outside finite floating-point range"
) from exc
return conductance, difference, flux
@dataclass(frozen=True)
class DiffusionEnergyBalance:
"""Instantaneous balance for the fixed symmetric EPI-only channel.
Arrays follow ``nodes`` and are detached from the graph. ``gradient``
is the Euclidean derivative of ``energy``; ``epi_rate`` includes the
actual nodal capacities. Zero mobility is permitted and does not imply
zero pressure. This is not a certificate for the full tetrad energy,
changing topology, or arbitrary finite integration steps.
"""
nodes: list
energy: float
gradient: Any
mobility: Any
epi_rate: Any
energy_rate: float
@dataclass(frozen=True)
class HeterogeneousDiffusionStabilityCertificate:
"""Lyapunov certificate for connected symmetric EPI diffusion.
``metric_weights`` is the binary64 evaluation of ``d_i / nu_f_i`` and
``lyapunov_value`` is the corresponding distance to consensus. The
eigensolver fields are estimates. ``exact_quotient_gap_lower_bound`` is
obtained from the actual represented generator and metric by rational
arithmetic; ``certified_exponential_rate_lower_bound`` is its
downward-rounded energy rate. Hybrid composition uses only the latter.
``exact_consensus_subspace_preservation`` says the represented generator
maps constant vectors back into that subspace.
``exact_uniform_fixed_point_preservation`` is the stronger canonical
diffusion requirement that it annihilate them; rounded Laplacian rows need
not satisfy this automatically.
``exact_weighted_mean_preservation`` separately records whether this
represented metric's mean is conserved exactly. ``conserved_total`` and
``equilibrium_value`` retain their compatibility names but are snapshot
values of ``h.T x`` and its projection center unless that flag is true.
The certificate concerns the frozen EPI-only channel; it does not certify
changing topology or an operator word.
"""
nodes: tuple[Any, ...]
equilibrium_value: float
metric_weights: Any
conserved_total: float
lyapunov_value: float
lyapunov_derivative: float
generalized_gap: float
exponential_rate: float
derivative_upper_bound: float
balance_residual: float
is_consensus: bool
is_equilibrium: bool
is_certified: bool
exact_quotient_gap_lower_bound: Fraction = Fraction(0)
certified_exponential_rate_lower_bound: float = 0.0
exact_consensus_subspace_preservation: bool = False
exact_uniform_fixed_point_preservation: bool = False
exact_weighted_mean_preservation: bool = False
_proof_stamp: tuple[Any, ...] = field(default=(), repr=False, compare=False)
def _proof_fields_are_intact(self) -> bool:
"""Detect ordinary construction, replacement, or payload mutation.
This private consistency stamp is not an authentication boundary;
deliberate reconstruction of private fields is outside its scope.
"""
try:
expected = _fixed_flow_proof_stamp(
self.nodes,
self.metric_weights,
self.exact_quotient_gap_lower_bound,
self.certified_exponential_rate_lower_bound,
self.is_certified,
self.exact_consensus_subspace_preservation,
self.exact_uniform_fixed_point_preservation,
self.exact_weighted_mean_preservation,
)
observed = object.__getattribute__(self, "_proof_stamp")
except BaseException:
return False
return proof_stamps_are_identical(observed, expected)
@dataclass(frozen=True)
class SwitchingDiffusionStabilityCertificate:
"""Common-metric certificate for a finite family of graph topologies.
Each regime may change conductance and structural frequency. The exact
common-Lyapunov theorem requires identical normalized weights
``d_i/nu_f_i`` in every regime and requires every represented generator to
annihilate constant fields exactly. Numerical agreement within a caller
chosen tolerance is reported separately and is not promoted to that
theorem. ``equilibrium_value``, ``lyapunov_value``, and
``derivative_upper_bound`` are binary64 diagnostics for the first snapshot
in the displayed normalized metric. The theorem is anchored instead in
``reference_metric_weights`` and the downward-rounded certified rate;
callers evaluating other snapshots must recenter them independently.
"""
nodes: tuple[Any, ...]
regime_count: int
reference_metric_weights: Any
normalized_metric_weights: Any
metric_mismatches: Any
generalized_gaps: Any
equilibrium_value: float
lyapunov_value: float
uniform_exponential_rate: float
derivative_upper_bound: float
common_metric_residual: float
shares_exact_common_metric: bool
common_metric_within_tolerance: bool
numerical_hypotheses_pass: bool
supports_exact_switching_theorem: bool
tolerance: float
scope: str
exact_quotient_gap_lower_bounds: tuple[Fraction, ...] = ()
certified_exponential_rate_lower_bounds: Any = field(
default_factory=lambda: _readonly_float_array(())
)
certified_uniform_exponential_rate_lower_bound: float = 0.0
exact_consensus_subspace_preservation_by_regime: tuple[bool, ...] = ()
exact_common_consensus_subspace_preservation: bool = False
exact_uniform_fixed_point_preservation_by_regime: tuple[bool, ...] = ()
exact_common_uniform_fixed_point_preservation: bool = False
exact_weighted_mean_preservation_by_regime: tuple[bool, ...] = ()
exact_common_weighted_mean_preservation: bool = False
_proof_stamp: tuple[Any, ...] = field(default=(), repr=False, compare=False)
@property
def shares_common_metric(self) -> bool:
"""Compatibility alias for exact common-metric equality."""
return self.shares_exact_common_metric
@property
def is_certified(self) -> bool:
"""Compatibility alias for the exact switching-theorem flag."""
return self.supports_exact_switching_theorem
def _proof_fields_are_intact(self) -> bool:
"""Detect ordinary construction, replacement, or payload mutation.
This private consistency stamp is not an authentication boundary;
deliberate reconstruction of private fields is outside its scope.
"""
try:
expected = _switching_flow_proof_stamp(
self.nodes,
self.regime_count,
self.reference_metric_weights,
self.normalized_metric_weights,
self.exact_quotient_gap_lower_bounds,
self.certified_exponential_rate_lower_bounds,
self.certified_uniform_exponential_rate_lower_bound,
self.shares_exact_common_metric,
self.supports_exact_switching_theorem,
self.exact_consensus_subspace_preservation_by_regime,
self.exact_common_consensus_subspace_preservation,
self.exact_uniform_fixed_point_preservation_by_regime,
self.exact_common_uniform_fixed_point_preservation,
self.exact_weighted_mean_preservation_by_regime,
self.exact_common_weighted_mean_preservation,
)
observed = object.__getattribute__(self, "_proof_stamp")
except BaseException:
return False
return proof_stamps_are_identical(observed, expected)
@dataclass(frozen=True)
class TimeVaryingDiffusionStabilityBound:
"""Conditional exact-real bound induced by represented conductances.
The supplied frequency arrays are assumptions on every instant of the
schedule, not observations made by this read-only function. Certification
interprets each effective binary64 conductance materialized by the shared
reader, and each frequency bound, as an exact real number; it forms degrees
and the Laplacian in rational arithmetic and proves a positive quotient gap
on the Euclidean disagreement subspace.
``materialized_binary64_uniform_fixed_point_preservation`` separately says
whether the ordinary binary64 ``diag(strength)-adjacency`` construction
annihilates the uniform field; it is diagnostic and does not define the
exact-real theorem.
``combinatorial_gap``, ``minimum_mobility``, ``maximum_mobility``,
``dirichlet_energy``, and the fields prefixed by ``spectral_`` are
compatibility estimates. ``energy_decay_rate``,
``energy_derivative_upper_bound``, and ``consensus_distance_bound`` use only
rational proof quantities and directed binary64 enclosures. An infinite
consensus-distance bound records safe abstention. The spectral estimates
may be nonfinite when the ordinary diagnostic path exceeds binary64 range.
``exact_real_continuous_time_model_certified`` records the rational theorem;
``operational_binary64_rate_available`` and its compatibility alias
``is_certified`` additionally require a positive public binary64 rate.
``runtime_integration_certified`` remains false because no numerical
integrator or future schedule is observed here.
"""
nodes: tuple[Any, ...]
frequency_lower_bounds: Any
frequency_upper_bounds: Any
combinatorial_gap: float
minimum_mobility: float
maximum_mobility: float
dirichlet_energy: float
energy_decay_rate: float
energy_derivative_upper_bound: float
consensus_distance_bound: float
is_certified: bool
spectral_energy_decay_rate_estimate: float
spectral_consensus_distance_estimate: float
exact_combinatorial_gap_lower_bound: Fraction
certified_combinatorial_gap_lower_bound: float
exact_minimum_mobility_lower_bound: Fraction
certified_minimum_mobility_lower_bound: float
exact_maximum_mobility_upper_bound: Fraction
certified_maximum_mobility_upper_bound: float
exact_energy_decay_rate_lower_bound: Fraction
exact_dirichlet_energy: Fraction
dirichlet_energy_lower_bound: float
dirichlet_energy_upper_bound: float
mobility_lower_bounds: Any
mobility_upper_bounds: Any
certified_mobility_lower_bounds: Any
certified_mobility_upper_bounds: Any
exact_real_uniform_fixed_point_preservation: bool
materialized_binary64_uniform_fixed_point_preservation: bool
exact_real_continuous_time_model_certified: bool
operational_binary64_rate_available: bool
runtime_integration_certified: bool
scope: str
@dataclass(frozen=True)
class EulerRelaxationWindowDiagnostic:
"""Modal explicit-Euler relaxation for one frozen symmetric network.
``modal_steps`` counts integration steps, whereas ``policy_window`` counts
operator positions. Both are reported for comparison and are not treated
as interchangeable semantics.
"""
dt: float
target_fraction: float
spectral_relative_tolerance: float
spectral_zero_threshold: float
decay_rates: Any
modal_multipliers: Any
slowest_decay_rate: float
fastest_decay_rate: float
euler_stability_limit: float
maximum_modal_factor: float
modal_steps: int | None
policy_window: int
is_euler_stable: bool
scope: str
def _connected_symmetric_transport(
G: Any,
nodes: list | None = None,
) -> tuple[ConductanceSnapshot, Any, Any]:
"""Return one validated positive, connected symmetric transport snapshot."""
conductance = read_conductance(G, nodes, symmetric=True)
if len(conductance.nodes) < 2:
raise ValueError("Stability certification requires at least two nodes")
strength = conductance.strength
if np.any(strength <= 0.0):
raise ValueError("Stability certification requires positive row strength")
adjacency = conductance.dense()
reached = {0}
frontier = [0]
while frontier:
source = frontier.pop()
for target in np.flatnonzero(adjacency[source] > 0.0):
target = int(target)
if target != source and target not in reached:
reached.add(target)
frontier.append(target)
if len(reached) != len(conductance.nodes):
raise ValueError(
"Stability certification requires connected positive conductance"
)
return conductance, adjacency, strength
def _exact_projected_quadratic(
matrix: tuple[tuple[Fraction, ...], ...],
metric_weights: tuple[Fraction, ...],
) -> tuple[tuple[Fraction, ...], ...]:
"""Restrict a quadratic form to ``h.T y=0`` in a rational basis."""
dimension = len(metric_weights)
pivot = max(range(dimension), key=metric_weights.__getitem__)
free = tuple(index for index in range(dimension) if index != pivot)
# Column a is e_free[a] - (h_free[a]/h_pivot)e_pivot. Choosing the
# largest metric entry keeps every ratio at most one. Expanding this
# two-sparse basis avoids an O(n^4) generic matrix multiplication.
ratios = tuple(metric_weights[index] / metric_weights[pivot] for index in free)
return tuple(
tuple(
(
matrix[free[left]][free[right]]
- ratios[left] * matrix[pivot][free[right]]
- ratios[right] * matrix[free[left]][pivot]
+ ratios[left] * ratios[right] * matrix[pivot][pivot]
)
for right in range(dimension - 1)
)
for left in range(dimension - 1)
)
def _exact_inverse_norm_gap_lower_bound(
dissipation: tuple[tuple[Fraction, ...], ...],
metric: tuple[tuple[Fraction, ...], ...],
) -> Fraction:
r"""Bound ``min z'Kz/z'Gz`` with exact rational arithmetic.
For symmetric positive-definite ``K`` and ``G``,
``lambda_min(K,G) >= 1/(||K^-1||_inf ||G||_inf)``.
That norm bound initializes a safe lower endpoint. Coordinate Rayleigh
quotients give an upper endpoint, and exact semidefiniteness tests then
refine the lower endpoint by rational bisection. Every returned value is
a proved lower bound; floating-point eigensolver output is never used.
"""
if not _exact_symmetric_semidefinite(dissipation, strict=True):
return Fraction(0)
inverse = _exact_matrix_inverse(dissipation)
inverse_norm = max(
sum((abs(value) for value in row), Fraction(0)) for row in inverse
)
metric_norm = max(sum((abs(value) for value in row), Fraction(0)) for row in metric)
if inverse_norm <= 0 or metric_norm <= 0:
return Fraction(0)
lower = Fraction(1) / (inverse_norm * metric_norm)
upper = min(
dissipation[index][index] / metric[index][index]
for index in range(len(dissipation))
)
if upper <= lower:
return lower
def candidate_is_valid(candidate: Fraction) -> bool:
shifted = tuple(
tuple(
dissipation[i][j] - candidate * metric[i][j]
for j in range(len(dissipation))
)
for i in range(len(dissipation))
)
return _exact_symmetric_semidefinite(shifted, strict=False)
if candidate_is_valid(upper):
return upper
if len(dissipation) > _EXACT_QUOTIENT_REFINEMENT_MAX_DIMENSION:
return lower
for _ in range(_EXACT_QUOTIENT_BISECTION_STEPS):
midpoint = (lower + upper) / 2
if candidate_is_valid(midpoint):
lower = midpoint
else:
upper = midpoint
return lower
@lru_cache(maxsize=128)
def _exact_real_laplacian_gap_lower_bound(
laplacian: tuple[tuple[Fraction, ...], ...],
) -> tuple[Fraction, bool]:
"""Certify the Euclidean disagreement gap of one rational Laplacian.
The caller forms degrees exactly from the rational interpretations of the
stored binary64 conductances. Symmetry and annihilation of the uniform
field are still checked before the quotient proof, so a malformed internal
matrix can only cause safe abstention.
"""
dimension = len(laplacian)
is_symmetric = all(
laplacian[i][j] == laplacian[j][i] for i in range(dimension) for j in range(i)
)
preserves_uniform_fixed_point = is_symmetric and all(
sum(row, Fraction(0)) == 0 for row in laplacian
)
if not preserves_uniform_fixed_point:
return Fraction(0), False
unit_weights = (Fraction(1),) * dimension
identity = tuple(
tuple(Fraction(i == j) for j in range(dimension)) for i in range(dimension)
)
restricted_laplacian = _exact_projected_quadratic(laplacian, unit_weights)
restricted_identity = _exact_projected_quadratic(identity, unit_weights)
gap = _exact_inverse_norm_gap_lower_bound(restricted_laplacian, restricted_identity)
return gap, True
def _exact_real_dirichlet_energy(
laplacian: tuple[tuple[Fraction, ...], ...],
field_values: Any,
) -> Fraction:
"""Evaluate ``x.T @ B @ x / 2`` for a rational model exactly."""
exact_field = tuple(
Fraction.from_float(float(value))
for value in np.asarray(field_values, dtype=float)
)
return (
sum(
(
exact_field[i] * laplacian[i][j] * exact_field[j]
for i in range(len(exact_field))
for j in range(len(exact_field))
),
Fraction(0),
)
/ 2
)
@lru_cache(maxsize=128)
def _exact_flow_gap_from_rationals(
laplacian: tuple[tuple[Fraction, ...], ...],
mobility: tuple[Fraction, ...],
metric: tuple[Fraction, ...],
) -> tuple[Fraction, bool, bool, bool]:
"""Cached exact quotient proof for one represented frozen generator."""
generator = tuple(
tuple(mobility[i] * laplacian[i][j] for j in range(len(metric)))
for i in range(len(metric))
)
mapped_consensus = tuple(sum(row, Fraction(0)) for row in generator)
preserves_consensus_subspace = all(
value == mapped_consensus[0] for value in mapped_consensus[1:]
)
preserves_uniform_fixed_points = all(value == 0 for value in mapped_consensus)
weighted_generator_row = tuple(
sum(
(metric[i] * generator[i][j] for i in range(len(metric))),
Fraction(0),
)
for j in range(len(metric))
)
preserves_weighted_mean = all(value == 0 for value in weighted_generator_row)
if not preserves_consensus_subspace:
return (
Fraction(0),
preserves_weighted_mean,
False,
preserves_uniform_fixed_points,
)
weighted_mobility = tuple(
metric[index] * mobility[index] for index in range(len(metric))
)
symmetric_dissipation = tuple(
tuple(
(
weighted_mobility[i] * laplacian[i][j]
+ laplacian[i][j] * weighted_mobility[j]
)
/ 2
for j in range(len(metric))
)
for i in range(len(metric))
)
metric_matrix = tuple(
tuple(metric[i] if i == j else Fraction(0) for j in range(len(metric)))
for i in range(len(metric))
)
restricted_dissipation = _exact_projected_quadratic(symmetric_dissipation, metric)
restricted_metric = _exact_projected_quadratic(metric_matrix, metric)
gap = _exact_inverse_norm_gap_lower_bound(restricted_dissipation, restricted_metric)
return (
gap,
preserves_weighted_mean,
True,