Repository navigation
Expand file tree
/
Copy pathdnfr.py
More file actions
3444 lines (3067 loc) · 120 KB
/
Copy pathdnfr.py
File metadata and controls
3444 lines (3067 loc) · 120 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
"""ΔNFR structural reorganization pressure utilities and strategies.
This module provides helper functions to configure, cache and apply ΔNFR
components from phase, EPI, structural frequency and topology during
simulations. The neighbour accumulation helpers reuse cached edge indices
and NumPy workspaces whenever available so cosine, sine, EPI, νf and topology
means remain faithful to the canonical ΔNFR reorganisation without redundant
allocations.
"""
from __future__ import annotations
import inspect
import math
import sys
from collections.abc import Callable, Iterator, Mapping, MutableMapping, Sequence
from concurrent.futures import ProcessPoolExecutor
from fractions import Fraction
from time import perf_counter
from types import ModuleType
from typing import Any, cast
from .._exact_time import finite_represented_real
from ..alias import get_attr, get_theta_attr, set_dnfr
from ..constants import DEFAULTS, get_param
from ..constants.aliases import ALIAS_EPI, ALIAS_VF
from ..mathematics._neighbor_differences import (
_require_finite_pressure,
edge_mean_differences,
mean_neighbor_difference,
)
from ..mathematics._phase_midpoint import certified_two_neighbor_phase
from ..mathematics.unified_numerical import compute_phase_difference, np
from ..metrics.common import (
_materialize_weight_mapping,
merge_and_normalize_weights,
merge_graph_weights,
)
from ..metrics.trig_cache import compute_theta_trig
from ..types import (
DeltaNFRHook,
DnfrCacheVectors,
DnfrVectorMap,
NeighborStats,
NodeId,
TNFRGraph,
require_finite_real_scalar_epi,
)
from ..utils import (
DNFR_PREP_STATE_KEY,
CacheManager,
DnfrCache,
DnfrPrepState,
_graph_cache_manager,
angle_diff,
cached_node_list,
cached_nodes_and_A,
new_dnfr_cache,
normalize_weights,
resolve_chunk_size,
)
from .canonical import validate_structural_frequency
from .fused_dnfr import compute_fused_gradients_symmetric
_SPARSE_DENSITY_THRESHOLD = 0.25
_DNFR_APPROX_BYTES_PER_EDGE = 48
_DNFR_CHANNELS = ("phase", "epi", "vf", "topo")
def _should_vectorize(G: TNFRGraph, np_module: ModuleType | None) -> bool:
"""Return ``True`` when NumPy is available unless the graph disables it."""
if np_module is None:
return False
flag = G.graph.get("vectorized_dnfr")
if flag is None:
return True
return bool(flag)
_NUMPY_CACHE_ATTRS = (
"theta_np",
"epi_np",
"vf_np",
"cos_theta_np",
"sin_theta_np",
"deg_array",
"neighbor_x_np",
"neighbor_y_np",
"neighbor_epi_sum_np",
"neighbor_vf_sum_np",
"neighbor_count_np",
"neighbor_deg_sum_np",
"neighbor_inv_count_np",
"neighbor_cos_avg_np",
"neighbor_sin_avg_np",
"neighbor_mean_tmp_np",
"neighbor_mean_length_np",
"neighbor_accum_np",
"neighbor_edge_values_np",
"dense_components_np",
"dense_accum_np",
"dense_degree_np",
)
def _profile_start_stop(
profile: MutableMapping[str, Any] | None,
*,
keys: Sequence[str] = (),
) -> tuple[Callable[[], float], Callable[[str, float], None]]:
"""Return helpers to measure wall-clock durations for ``profile`` keys."""
if profile is not None:
for key in keys:
profile.setdefault(key, 0.0)
def _start() -> float:
return perf_counter()
def _stop(metric: str, start: float) -> None:
profile[metric] = float(profile.get(metric, 0.0)) + (perf_counter() - start)
else:
def _start() -> float:
return 0.0
def _stop(
metric: str, start: float
) -> None: # noqa: ARG001 - uniform signature
return None
return _start, _stop
def _iter_chunk_offsets(total: int, jobs: int) -> Iterator[tuple[int, int]]:
"""Yield ``(start, end)`` offsets splitting ``total`` items across ``jobs``."""
if total <= 0 or jobs <= 1:
return
jobs = max(1, min(int(jobs), total))
base, extra = divmod(total, jobs)
start = 0
for i in range(jobs):
size = base + (1 if i < extra else 0)
if size <= 0:
continue
end = start + size
yield start, end
start = end
def _neighbor_sums_worker(
start: int,
end: int,
neighbor_indices: Sequence[Sequence[int]],
cos_th: Sequence[float],
sin_th: Sequence[float],
epi: Sequence[float],
vf: Sequence[float],
x_base: Sequence[float],
y_base: Sequence[float],
epi_base: Sequence[float],
vf_base: Sequence[float],
count_base: Sequence[float],
deg_base: Sequence[float] | None,
deg_list: Sequence[float] | None,
degs_list: Sequence[float] | None,
) -> tuple[
int,
list[float],
list[float],
list[float],
list[float],
list[float],
list[float] | None,
]:
"""Return partial neighbour sums for the ``[start, end)`` range."""
chunk_x: list[float] = []
chunk_y: list[float] = []
chunk_epi: list[float] = []
chunk_vf: list[float] = []
chunk_count: list[float] = []
chunk_deg: list[float] | None = [] if deg_base is not None else None
for offset, idx in enumerate(range(start, end)):
neighbors = neighbor_indices[idx]
x_i = float(x_base[offset])
y_i = float(y_base[offset])
epi_i = float(epi_base[offset])
vf_i = float(vf_base[offset])
count_i = float(count_base[offset])
if deg_base is not None and chunk_deg is not None:
deg_i_acc = float(deg_base[offset])
else:
deg_i_acc = 0.0
deg_i = float(degs_list[idx]) if degs_list is not None else 0.0
for neighbor_idx in neighbors:
x_i += float(cos_th[neighbor_idx])
y_i += float(sin_th[neighbor_idx])
epi_i += float(epi[neighbor_idx])
vf_i += float(vf[neighbor_idx])
count_i += 1.0
if chunk_deg is not None:
if deg_list is not None:
deg_i_acc += float(deg_list[neighbor_idx])
else:
deg_i_acc += deg_i
chunk_x.append(x_i)
chunk_y.append(y_i)
chunk_epi.append(epi_i)
chunk_vf.append(vf_i)
chunk_count.append(count_i)
if chunk_deg is not None:
chunk_deg.append(deg_i_acc)
return (
start,
chunk_x,
chunk_y,
chunk_epi,
chunk_vf,
chunk_count,
chunk_deg,
)
def _dnfr_gradients_worker(
start: int,
end: int,
nodes: Sequence[NodeId],
theta: Sequence[float] | Any,
epi: Sequence[float] | Any,
vf: Sequence[float] | Any,
th_bar: Sequence[float] | Any,
epi_bar: Sequence[float] | Any,
vf_bar: Sequence[float] | Any,
deg_bar: Sequence[float] | Any | None,
degs: Mapping[Any, float] | Sequence[float] | None,
w_phase: float,
w_epi: float,
w_vf: float,
w_topo: float,
epi_gradient: Sequence[float] | None = None,
vf_gradient: Sequence[float] | None = None,
phase_overrides: Mapping[int, float] | None = None,
) -> tuple[int, list[float]]:
"""Return partial ΔNFR gradients for the ``[start, end)`` range."""
chunk: list[float] = []
for idx in range(start, end):
n = nodes[idx]
g_phase = (
phase_overrides[idx]
if phase_overrides and idx in phase_overrides
else -angle_diff(theta[idx], th_bar[idx]) / math.pi
)
g_epi = (
(epi_gradient[idx] if epi_gradient is not None else epi_bar[idx] - epi[idx])
if w_epi
else 0.0
)
g_vf = (
(vf_gradient[idx] if vf_gradient is not None else vf_bar[idx] - vf[idx])
if w_vf
else 0.0
)
if w_topo != 0.0 and deg_bar is not None and degs is not None:
if isinstance(degs, dict):
deg_i = float(degs.get(n, 0))
else:
deg_i = float(degs[idx])
g_topo = deg_bar[idx] - deg_i
else:
g_topo = 0.0
chunk.append(
w_phase * g_phase
+ (g_epi if epi_gradient is not None else w_epi * g_epi)
+ (g_vf if vf_gradient is not None else w_vf * g_vf)
+ w_topo * g_topo
)
return start, chunk
def _resolve_parallel_jobs(n_jobs: int | None, total: int) -> int | None:
"""Return an effective worker count for ``total`` items or ``None``."""
if n_jobs is None:
return None
try:
jobs = int(n_jobs)
except (TypeError, ValueError):
return None
if jobs <= 1 or total <= 1:
return None
return max(1, min(jobs, total))
def _is_numpy_like(obj) -> bool:
return (
getattr(obj, "dtype", None) is not None
and getattr(obj, "shape", None) is not None
)
def _has_cached_numpy_buffers(
data: MutableMapping[str, Any], cache: DnfrCache | None
) -> bool:
for attr in _NUMPY_CACHE_ATTRS:
arr = data.get(attr)
if _is_numpy_like(arr):
return True
if cache is not None:
for attr in _NUMPY_CACHE_ATTRS:
arr = getattr(cache, attr, None)
if _is_numpy_like(arr):
return True
A = data.get("A")
if _is_numpy_like(A):
return True
return False
__all__ = (
"default_compute_delta_nfr",
"set_delta_nfr_hook",
"dnfr_phase_only",
"dnfr_epi_vf_mixed",
"dnfr_laplacian",
"compute_delta_nfr_hamiltonian",
)
def _write_dnfr_metadata(
G, *, weights: Mapping[str, float], hook_name: str, note: str | None = None
) -> None:
"""Write a ``_DNFR_META`` block in ``G.graph`` with the mix and hook name.
``weights`` may include arbitrary components (phase/epi/vf/topo/etc.).
"""
weights_norm = normalize_weights(weights, weights.keys())
meta = {
"hook": hook_name,
"weights_raw": dict(weights),
"weights_effective": dict(weights),
"weights_norm": weights_norm,
"components": [k for k, v in weights.items() if v != 0.0],
"doc": "ΔNFR = Σ w_i·g_i",
}
if note:
meta["note"] = str(note)
G.graph["_DNFR_META"] = meta
G.graph["_dnfr_hook_name"] = hook_name # string friendly
def _resolve_dnfr_weights(G) -> Mapping[str, float]:
"""Read effective coefficients without mutating the graph.
A legacy explicitly supplied private mix remains an override until the
public configuration changes after its first pressure preparation. An
engine-owned cached mix must never hide a subsequent public configuration
edit. Detached observers use this same resolution rule.
"""
cached = G.graph.get("_dnfr_weights")
source = G.graph.get("_dnfr_weights_source")
configured = _materialize_weight_mapping(
merge_graph_weights(G, "DNFR_WEIGHTS", strict=True),
_DNFR_CHANNELS,
name="DNFR_WEIGHTS",
)
if cached is not None and (source is None or source == configured):
return _materialize_weight_mapping(
cached, _DNFR_CHANNELS, name="cached DNFR weights"
)
return merge_and_normalize_weights(G, "DNFR_WEIGHTS", _DNFR_CHANNELS, default=0.0)
def _configure_dnfr_weights(G) -> Mapping[str, float]:
"""Store the effective mix and a detached snapshot of its public source."""
weights = _resolve_dnfr_weights(G)
G.graph["_dnfr_weights"] = weights
G.graph["_dnfr_weights_source"] = _materialize_weight_mapping(
merge_graph_weights(G, "DNFR_WEIGHTS", strict=True),
_DNFR_CHANNELS,
name="DNFR_WEIGHTS",
)
return weights
def _read_scalar_epi(nd: Mapping[str, Any]) -> float:
"""Read the finite signed scalar chart required by this pressure model."""
return get_attr(
nd, ALIAS_EPI, 0.0, strict=True, conv=require_finite_real_scalar_epi
)
def _read_capacity(nd: Mapping[str, Any], default: float = 0.0) -> float:
"""Apply nodal capacity admission before any alias or float coercion."""
return get_attr(
nd, ALIAS_VF, default, strict=True, conv=validate_structural_frequency
)
def _read_phase(nd: Mapping[str, Any]) -> float:
"""Read the authoritative finite phase before building trigonometric data."""
return get_theta_attr(
nd,
0.0,
strict=True,
conv=lambda raw: finite_represented_real(raw, "pressure phase")[0],
)
def _pressure_phase_coordinate(value: float) -> float:
"""Read the principal phasor chart before a phase-pressure subtraction."""
return (
value if abs(value) <= math.pi else math.atan2(math.sin(value), math.cos(value))
)
def _init_dnfr_cache(
G: TNFRGraph,
nodes: Sequence[NodeId],
cache_or_manager: CacheManager | DnfrCache | None = None,
checksum: Any | None = None,
force_refresh: bool = False,
*,
manager: CacheManager | None = None,
) -> tuple[
DnfrCache,
dict[NodeId, int],
list[float],
list[float],
list[float],
list[float],
list[float],
bool,
]:
"""Initialise or reuse cached ΔNFR arrays.
``manager`` telemetry records cache hits, misses and timings. Legacy
callers can still pass a ``cache`` instance as the
third positional argument; this helper supports both signatures by seeding
the manager-backed state with the provided cache when necessary.
"""
if manager is None and isinstance(cache_or_manager, CacheManager):
manager = cache_or_manager
cache_or_manager = None
if manager is None:
manager = _graph_cache_manager(G.graph)
graph = G.graph
state = manager.get(DNFR_PREP_STATE_KEY)
if not isinstance(state, DnfrPrepState):
manager.clear(DNFR_PREP_STATE_KEY)
state = manager.get(DNFR_PREP_STATE_KEY)
if isinstance(cache_or_manager, DnfrCache):
state.cache = cache_or_manager
if checksum is None:
checksum = cache_or_manager.checksum
cache = state.cache
reuse = (
not force_refresh
and isinstance(cache, DnfrCache)
and cache.checksum == checksum
and len(cache.theta) == len(nodes)
)
if reuse:
manager.increment_hit(DNFR_PREP_STATE_KEY)
graph["_dnfr_prep_cache"] = cache
return (
cache,
cache.idx,
cache.theta,
cache.epi,
cache.vf,
cache.cos_theta,
cache.sin_theta,
False,
)
def _rebuild(current: DnfrPrepState | Any) -> DnfrPrepState:
if not isinstance(current, DnfrPrepState):
raise RuntimeError("ΔNFR prep state unavailable during rebuild")
prev_cache = current.cache if isinstance(current.cache, DnfrCache) else None
idx_local = {n: i for i, n in enumerate(nodes)}
size = len(nodes)
zeros = [0.0] * size
cache_new = prev_cache if prev_cache is not None else new_dnfr_cache()
cache_new.idx = idx_local
cache_new.theta = zeros.copy()
cache_new.epi = zeros.copy()
cache_new.vf = zeros.copy()
cache_new.cos_theta = [1.0] * size
cache_new.sin_theta = [0.0] * size
cache_new.neighbor_x = zeros.copy()
cache_new.neighbor_y = zeros.copy()
cache_new.neighbor_epi_sum = zeros.copy()
cache_new.neighbor_vf_sum = zeros.copy()
cache_new.neighbor_count = zeros.copy()
cache_new.neighbor_deg_sum = zeros.copy() if size else []
cache_new.degs = None
cache_new.edge_src = None
cache_new.edge_dst = None
cache_new.checksum = checksum
# Reset any numpy mirrors or aggregated buffers to avoid leaking
# state across refresh cycles (e.g. switching between vectorised
# and Python paths or reusing legacy caches).
if prev_cache is not None:
for attr in _NUMPY_CACHE_ATTRS:
setattr(cache_new, attr, None)
for attr in (
"th_bar_np",
"epi_bar_np",
"vf_bar_np",
"deg_bar_np",
"grad_phase_np",
"grad_epi_np",
"grad_vf_np",
"grad_topo_np",
"grad_total_np",
):
setattr(cache_new, attr, None)
cache_new.edge_src = None
cache_new.edge_dst = None
cache_new.edge_signature = None
cache_new.neighbor_accum_signature = None
cache_new.degs = prev_cache.degs if prev_cache else None
cache_new.checksum = checksum
current.cache = cache_new
graph["_dnfr_prep_cache"] = cache_new
return current
with manager.timer(DNFR_PREP_STATE_KEY):
state = manager.update(DNFR_PREP_STATE_KEY, _rebuild)
manager.increment_miss(DNFR_PREP_STATE_KEY)
cache = state.cache
if not isinstance(cache, DnfrCache): # pragma: no cover - defensive guard
raise RuntimeError("ΔNFR cache initialisation failed")
return (
cache,
cache.idx,
cache.theta,
cache.epi,
cache.vf,
cache.cos_theta,
cache.sin_theta,
True,
)
def _ensure_numpy_vectors(cache: DnfrCache) -> DnfrCacheVectors:
"""Ensure NumPy copies of cached vectors are initialised and up to date."""
if cache is None:
return (None, None, None, None, None)
arrays: list[Any | None] = []
size = len(cache.theta)
for attr_np, source_attr in (
("theta_np", "theta"),
("epi_np", "epi"),
("vf_np", "vf"),
("cos_theta_np", "cos_theta"),
("sin_theta_np", "sin_theta"),
):
arr = getattr(cache, attr_np)
if arr is not None and getattr(arr, "shape", None) == (size,):
arrays.append(arr)
continue
src = getattr(cache, source_attr)
if src is None:
setattr(cache, attr_np, None)
arrays.append(None)
continue
arr = np.asarray(src, dtype=float)
if getattr(arr, "shape", None) != (size,):
arr = np.array(src, dtype=float)
setattr(cache, attr_np, arr)
arrays.append(arr)
return tuple(arrays)
def _ensure_numpy_degrees(
cache: DnfrCache,
deg_list: Sequence[float] | None,
) -> np.ndarray | None:
"""Initialise/update NumPy array mirroring ``deg_list``.
Deg_array reuse pattern:
-------------------------
The degree array (deg_array) is a cached NumPy buffer that stores node
degrees for topology-based ΔNFR computations. The reuse pattern follows:
1. **Allocation**: Created once when topology weight (w_topo) > 0 or when
caching is enabled, sized to match the node count.
2. **Reuse across steps**: When the graph topology is stable (no edge
additions/removals), the same deg_array buffer is reused across
multiple ΔNFR computation steps by updating in-place via np.copyto.
3. **Count buffer optimization**: For undirected graphs where node degree
equals neighbor count, deg_array can serve double duty as the count
buffer (see _accumulate_neighbors_numpy lines 2185-2194), eliminating
the need for an extra accumulator row.
4. **Invalidation**: Cache is cleared when graph.edges changes or when
_dnfr_prep_dirty flag is set, ensuring fresh allocation on next use.
This pattern maintains ΔNFR computational accuracy (Invariant #1:
Nodal Equation Integrity) while
minimizing allocations for stable topologies.
"""
if deg_list is None:
if cache is not None:
cache.deg_array = None
return None
if cache is None:
return np.array(deg_list, dtype=float)
arr = cache.deg_array
if arr is None or len(arr) != len(deg_list):
arr = np.array(deg_list, dtype=float)
else:
np.copyto(arr, deg_list, casting="unsafe")
cache.deg_array = arr
return arr
def _resolve_numpy_degree_array(
data: MutableMapping[str, Any],
count: np.ndarray | None,
*,
cache: DnfrCache | None,
) -> np.ndarray | None:
"""Return the vector of node degrees required for topology gradients."""
if data["w_topo"] == 0.0:
return None
deg_array = data.get("deg_array")
if deg_array is not None:
return deg_array
deg_list = data.get("deg_list")
if deg_list is not None:
deg_array = np.array(deg_list, dtype=float)
data["deg_array"] = deg_array
if cache is not None:
cache.deg_array = deg_array
return deg_array
return count
def _ensure_cached_array(
cache: DnfrCache | None,
attr: str,
shape: tuple[int, ...],
) -> np.ndarray:
"""Return a cached NumPy buffer with ``shape`` creating/reusing it."""
if np is None:
raise RuntimeError("NumPy is required to build cached arrays")
arr = getattr(cache, attr) if cache is not None else None
if arr is None or getattr(arr, "shape", None) != shape:
arr = np.empty(shape, dtype=float)
if cache is not None:
setattr(cache, attr, arr)
return arr
def _ensure_numpy_state_vectors(data: MutableMapping[str, Any]) -> DnfrVectorMap:
"""Synchronise list-based state vectors with their NumPy counterparts."""
nodes = data.get("nodes") or ()
size = len(nodes)
cache: DnfrCache | None = data.get("cache")
cache_arrays: DnfrCacheVectors = (None, None, None, None, None)
if cache is not None:
cache_arrays = _ensure_numpy_vectors(cache)
result: dict[str, Any | None] = {}
for plain_key, np_key, cached_arr, result_key in (
("theta", "theta_np", cache_arrays[0], "theta"),
("epi", "epi_np", cache_arrays[1], "epi"),
("vf", "vf_np", cache_arrays[2], "vf"),
("cos_theta", "cos_theta_np", cache_arrays[3], "cos"),
("sin_theta", "sin_theta_np", cache_arrays[4], "sin"),
):
arr = data.get(np_key)
if arr is None:
arr = cached_arr
if arr is None or getattr(arr, "shape", None) != (size,):
src = data.get(plain_key)
if src is None and cache is not None:
src = getattr(cache, plain_key)
if src is None:
arr = None
else:
arr = np.asarray(src, dtype=float)
if getattr(arr, "shape", None) != (size,):
arr = np.array(src, dtype=float)
if arr is not None:
data[np_key] = arr
data[plain_key] = arr
if cache is not None:
setattr(cache, np_key, arr)
else:
data[np_key] = None
result[result_key] = arr
return result
def _build_edge_index_arrays(
G: TNFRGraph,
nodes: Sequence[NodeId],
idx: Mapping[NodeId, int],
) -> tuple[Any, Any]:
"""Create (src, dst) index arrays for ``G`` respecting ``nodes`` order."""
from ..mathematics.unified_numerical import np
if np is None:
return None, None
if not nodes:
empty = np.empty(0, dtype=np.intp)
return empty, empty
src = []
dst = []
append_src = src.append
append_dst = dst.append
for node in nodes:
i = idx.get(node)
if i is None:
continue
for neighbor in G.neighbors(node):
j = idx.get(neighbor)
if j is None:
continue
append_src(i)
append_dst(j)
if not src:
empty = np.empty(0, dtype=np.intp)
return empty, empty
edge_src = np.asarray(src, dtype=np.intp)
edge_dst = np.asarray(dst, dtype=np.intp)
return edge_src, edge_dst
def _build_edge_weight_array(
G: TNFRGraph,
nodes: Sequence[NodeId],
edge_src: Any,
edge_dst: Any,
) -> Any:
"""Per-edge weights aligned to ``(edge_src, edge_dst)``, read live from G.
The EPI channel realizes the random-walk Laplacian L_rw = I − D⁻¹W, so it
must use the current edge weights. Reading them fresh each call keeps the
weighted diffusion correct even when edge weights mutate without a topology
change. Returns ``None`` when every weight is unity — the fused kernel then
takes the exact unweighted path.
"""
if np is None or edge_src is None or edge_dst is None:
return None
n_edges = int(getattr(edge_src, "shape", (0,))[0])
if n_edges == 0:
return None
weights = None
is_multigraph = G.is_multigraph()
for k in range(n_edges):
u = nodes[int(edge_src[k])]
v = nodes[int(edge_dst[k])]
edge_data = G[u][v]
if is_multigraph:
# NetworkX adjacency sums parallel edges; the EPI channel must
# realize the same W while other channels count unique neighbors.
w = float(sum(attrs.get("weight", 1.0) for attrs in edge_data.values()))
else:
w = float(edge_data.get("weight", 1.0))
if w != 1.0:
if weights is None:
weights = np.ones(n_edges, dtype=float)
weights[k] = w
return weights
def _refresh_dnfr_vectors(
G: TNFRGraph, nodes: Sequence[NodeId], cache: DnfrCache
) -> None:
"""Update cached angle and state vectors for ΔNFR."""
np_module = np
trig = compute_theta_trig((n, _read_phase(G.nodes[n])) for n in nodes)
use_numpy = _should_vectorize(G, np_module)
node_count = len(nodes)
trig_theta = getattr(trig, "theta_values", None)
trig_cos = getattr(trig, "cos_values", None)
trig_sin = getattr(trig, "sin_values", None)
np_ready = (
use_numpy
and np_module is not None
and isinstance(trig_theta, getattr(np_module, "ndarray", tuple()))
and isinstance(trig_cos, getattr(np_module, "ndarray", tuple()))
and isinstance(trig_sin, getattr(np_module, "ndarray", tuple()))
and getattr(trig_theta, "shape", None) == getattr(trig_cos, "shape", None)
and getattr(trig_theta, "shape", None) == getattr(trig_sin, "shape", None)
and (trig_theta.shape[0] if getattr(trig_theta, "ndim", 0) else 0) == node_count
)
if np_ready:
if node_count:
epi_arr = np_module.fromiter(
(_read_scalar_epi(G.nodes[node]) for node in nodes),
dtype=float,
count=node_count,
)
vf_arr = np_module.fromiter(
(_read_capacity(G.nodes[node]) for node in nodes),
dtype=float,
count=node_count,
)
else:
epi_arr = np_module.empty(0, dtype=float)
vf_arr = np_module.empty(0, dtype=float)
theta_arr = np_module.asarray(trig_theta, dtype=float)
cos_arr = np_module.asarray(trig_cos, dtype=float)
sin_arr = np_module.asarray(trig_sin, dtype=float)
def _sync_numpy(attr: str, source: Any) -> Any:
dest = getattr(cache, attr)
if dest is None or getattr(dest, "shape", None) != source.shape:
dest = np_module.array(source, dtype=float)
else:
np_module.copyto(dest, source, casting="unsafe")
setattr(cache, attr, dest)
return dest
_sync_numpy("theta_np", theta_arr)
_sync_numpy("epi_np", epi_arr)
_sync_numpy("vf_np", vf_arr)
_sync_numpy("cos_theta_np", cos_arr)
_sync_numpy("sin_theta_np", sin_arr)
# Python mirrors remain untouched while the vectorised path is active.
# They will be rebuilt the next time the runtime falls back to lists.
if cache.theta is not None and len(cache.theta) != node_count:
cache.theta = [0.0] * node_count
if cache.epi is not None and len(cache.epi) != node_count:
cache.epi = [0.0] * node_count
if cache.vf is not None and len(cache.vf) != node_count:
cache.vf = [0.0] * node_count
if cache.cos_theta is not None and len(cache.cos_theta) != node_count:
cache.cos_theta = [1.0] * node_count
if cache.sin_theta is not None and len(cache.sin_theta) != node_count:
cache.sin_theta = [0.0] * node_count
else:
for index, node in enumerate(nodes):
i: int = int(index)
node_id: NodeId = node
nd = G.nodes[node_id]
cache.theta[i] = trig.theta[node_id]
cache.epi[i] = _read_scalar_epi(nd)
cache.vf[i] = _read_capacity(nd)
cache.cos_theta[i] = trig.cos[node_id]
cache.sin_theta[i] = trig.sin[node_id]
if use_numpy and np_module is not None:
_ensure_numpy_vectors(cache)
else:
cache.theta_np = None
cache.epi_np = None
cache.vf_np = None
cache.cos_theta_np = None
cache.sin_theta_np = None
def _prepare_dnfr_data(
G: TNFRGraph,
*,
cache_size: int | None = 128,
profile: MutableMapping[str, Any] | None = None,
) -> dict[str, Any]:
"""Precompute common data for ΔNFR strategies.
The helper decides between edge-wise and dense adjacency accumulation
heuristically. Graphs whose edge density exceeds
``_SPARSE_DENSITY_THRESHOLD`` receive a cached adjacency matrix so the
dense path can be exercised; callers may also force the dense mode by
setting ``G.graph['dnfr_force_dense']`` to a truthy value.
Parameters
----------
profile : MutableMapping[str, float] or None, optional
Mutable mapping that accumulates wall-clock timings for ΔNFR
preparation. When provided the helper increases the
``"dnfr_cache_rebuild"`` bucket with the time spent refreshing cached
node vectors and associated NumPy workspaces.
"""
start_timer, stop_timer = _profile_start_stop(
profile,
keys=("dnfr_cache_rebuild",),
)
graph = G.graph
weights = _configure_dnfr_weights(G)
result: dict[str, Any] = {
"weights": weights,
"cache_size": cache_size,
}
np_module = np
use_numpy = _should_vectorize(G, np_module)
nodes = cast(tuple[NodeId, ...], cached_node_list(G))
edge_count = G.number_of_edges()
# Centralized decision logic for sparse vs dense accumulation path.
# This decision affects which accumulation strategy will be used:
# - "sparse": edge-based accumulation (_accumulate_neighbors_broadcasted)
# - "dense": matrix multiplication with adjacency matrix (_accumulate_neighbors_dense)
# The decision is stored in dnfr_path_decision for telemetry and debugging.
prefer_sparse = False
dense_override = bool(G.graph.get("dnfr_force_dense"))
dnfr_path_decision = "fallback" # Default when numpy unavailable
if use_numpy:
# Heuristic: use sparse path when density <= _SPARSE_DENSITY_THRESHOLD (0.25)
prefer_sparse = _prefer_sparse_accumulation(len(nodes), edge_count)
if dense_override:
# User explicitly requested dense mode
prefer_sparse = False
dnfr_path_decision = "dense_forced"
elif not prefer_sparse:
# Heuristic chose dense path (high density graph)
dnfr_path_decision = "dense_auto"
else:
# Heuristic chose sparse path (low density graph)
dnfr_path_decision = "sparse"
nodes_cached, A_untyped = cached_nodes_and_A(
G,
cache_size=cache_size,
require_numpy=False,
prefer_sparse=prefer_sparse,
nodes=nodes,
)
nodes = cast(tuple[NodeId, ...], nodes_cached)
A: np.ndarray | None = A_untyped
result["nodes"] = nodes
result["A"] = A
manager = _graph_cache_manager(G.graph)
checksum = G.graph.get("_dnfr_nodes_checksum")
dirty_flag = bool(G.graph.pop("_dnfr_prep_dirty", False))
existing_cache = cast(DnfrCache | None, graph.get("_dnfr_prep_cache"))
cache_timer = start_timer()
cache, idx, theta, epi, vf, cos_theta, sin_theta, refreshed = _init_dnfr_cache(
G,
nodes,
existing_cache,
checksum,
force_refresh=dirty_flag,
manager=manager,
)
stop_timer("dnfr_cache_rebuild", cache_timer)
dirty = dirty_flag or refreshed
caching_enabled = cache is not None and (cache_size is None or cache_size > 0)
result["cache"] = cache
result["idx"] = idx
result["theta"] = theta
result["epi"] = epi
result["vf"] = vf
result["cos_theta"] = cos_theta
result["sin_theta"] = sin_theta
if cache is not None:
_refresh_dnfr_vectors(G, nodes, cache)
if np_module is None and not caching_enabled:
for attr in (
"neighbor_x_np",
"neighbor_y_np",
"neighbor_epi_sum_np",
"neighbor_vf_sum_np",
"neighbor_count_np",
"neighbor_deg_sum_np",
"neighbor_inv_count_np",
"neighbor_cos_avg_np",
"neighbor_sin_avg_np",
"neighbor_mean_tmp_np",
"neighbor_mean_length_np",