Spaces:
Sleeping
Sleeping
File size: 2,318 Bytes
9580dda | 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 | from __future__ import annotations
from collections import deque
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Any
import networkx as nx
if TYPE_CHECKING:
Graph = nx.DiGraph[str, dict[str, Any], dict[str, Any]]
@dataclass
class NodeState:
impact: float
eta: float
hops: int
path: list[str] = field(default_factory=list)
def substitution_factor(graph: Graph, node_id: str, state: dict[str, NodeState]) -> float:
"""Reduce impact when a node still has healthy substitute nodes."""
substitutes: list[str] = graph.nodes[node_id].get("substitutes", [])
if not substitutes:
return 1.0
unaffected = sum(1 for substitute in substitutes if substitute not in state)
return 1.0 - 0.5 * (unaffected / len(substitutes))
def propagate(
graph: Graph,
epicentres: dict[str, float],
*,
max_hops: int = 4,
decay: float = 0.75,
min_impact: float = 0.02,
) -> dict[str, NodeState]:
"""Propagate impact with a deterministic, cycle-safe weighted BFS."""
state: dict[str, NodeState] = {
node_id: NodeState(impact=impact, eta=0.0, hops=0, path=[node_id])
for node_id, impact in epicentres.items()
}
frontier: deque[str] = deque(epicentres)
while frontier:
source = frontier.popleft()
source_state = state[source]
if source_state.hops >= max_hops:
continue
for target in graph.successors(source):
edge = graph.edges[source, target]
node = graph.nodes[target]
contribution = source_state.impact * edge["weight"] * decay ** (source_state.hops + 1)
contribution *= 1.0 - node.get("resilience", 0.0)
contribution *= substitution_factor(graph, target, state)
if contribution < min_impact:
continue
candidate = NodeState(
impact=min(1.0, contribution),
eta=source_state.eta + edge.get("lead_time_days", 0),
hops=source_state.hops + 1,
path=[*source_state.path, target],
)
previous = state.get(target)
if previous is None or candidate.impact > previous.impact:
state[target] = candidate
frontier.append(target)
return state
|