"""Interactive Plotly (Python) map of the gossip traversal. Produces a native plotly.graph_objects.Figure. Call .show() to open it in the user's default browser (Python-driven, no pre-rendered HTML file). """ from __future__ import annotations from pathlib import Path from typing import Optional import matplotlib.cm as mpl_cm import matplotlib.colors as mcolors import pandas as pd import plotly.graph_objects as go def _hex_from_cmap(value: float, cmap_name: str = "plasma") -> str: cmap = mpl_cm.get_cmap(cmap_name) return mcolors.to_hex(cmap(value)) def plot_interactive_map( path: Optional[Path], nodes: pd.DataFrame, origin_ip: str, hop_of: dict[str, int], parent_of: dict[str, str], parent_km_of: dict[str, float] | None, *, link_km: float, origin_label: str = "", show: bool = True, show_unreachable_in_view: bool = True, ) -> go.Figure: """ Build a Plotly interactive scattergeo figure of the gossip traversal. `path` — optional: if given, pickle the figure object there (`.pkl`), so it can be reloaded later with pickle.load. No HTML is written. `show` — if True, call fig.show() which opens in default browser. Returns the plotly Figure so the caller can inspect / extend it. """ reach_ips = set(hop_of.keys()) ip_to_row = nodes.set_index("ip").to_dict("index") max_hop = max(hop_of.values()) if hop_of else 0 o_row = ip_to_row[origin_ip] o_lat = float(o_row["latitude"]) o_lon = float(o_row["longitude"]) reach_df = nodes[nodes["ip"].isin(reach_ips)] min_lat = float(min(reach_df["latitude"].min(), o_lat)) max_lat = float(max(reach_df["latitude"].max(), o_lat)) min_lon = float(min(reach_df["longitude"].min(), o_lon)) max_lon = float(max(reach_df["longitude"].max(), o_lon)) fig = go.Figure() # --- parent -> child tree edges (one trace, gap-separated by None) ------ edge_lon: list[float | None] = [] edge_lat: list[float | None] = [] for child, par in parent_of.items(): c = ip_to_row.get(child) p = ip_to_row.get(par) if not c or not p: continue edge_lon.extend([float(p["longitude"]), float(c["longitude"]), None]) edge_lat.extend([float(p["latitude"]), float(c["latitude"]), None]) fig.add_trace( go.Scattergeo( lon=edge_lon, lat=edge_lat, mode="lines", line=dict(width=0.6, color="rgba(120,120,120,0.55)"), hoverinfo="skip", name="tree edges", ) ) # --- unreachable listeners (padded view, hidden by default) ------------- if show_unreachable_in_view: pad = 6.0 view = nodes[ (nodes["longitude"].between(min_lon - pad, max_lon + pad)) & (nodes["latitude"].between(min_lat - pad, max_lat + pad)) & (~nodes["ip"].isin(reach_ips)) & (nodes["ip"] != origin_ip) ] if len(view): fig.add_trace( go.Scattergeo( lon=view["longitude"], lat=view["latitude"], mode="markers", marker=dict(size=4, color="#bdbdbd", opacity=0.55, line=dict(width=0)), name=f"unreachable ({len(view)} in view)", text=[ f"{r['ip']}
{r.get('city','')}, {r.get('country','')}" f"
unreachable" for _, r in view.iterrows() ], hovertemplate="%{text}", visible="legendonly", ) ) # --- one trace per hop (legend-click toggles the layer) ----------------- for h in range(1, max_hop + 1): ips = [ip for ip, v in hop_of.items() if v == h] if not ips: continue rows = [ip_to_row[i] for i in ips if i in ip_to_row] lons = [float(r["longitude"]) for r in rows] lats = [float(r["latitude"]) for r in rows] hover = [] for ip, r in zip(ips, rows): par = parent_of.get(ip, "") km = (parent_km_of or {}).get(ip) km_str = f"{km:.1f} km" if km is not None else "-" hover.append( f"{ip}
" f"{r.get('city','')}, {r.get('country','')}
" f"hop {h}
" f"parent: {par}
" f"distance to parent: {km_str}" ) color = _hex_from_cmap(h / max(max_hop, 1)) fig.add_trace( go.Scattergeo( lon=lons, lat=lats, mode="markers", marker=dict( size=7, color=color, line=dict(width=0.5, color="black"), opacity=0.9, ), name=f"hop {h} ({len(ips)})", text=hover, hovertemplate="%{text}", ) ) # --- origin (tx entry) -------------------------------------------------- fig.add_trace( go.Scattergeo( lon=[o_lon], lat=[o_lat], mode="markers+text", marker=dict( size=18, color="red", symbol="star", line=dict(width=1.5, color="black"), ), text=[origin_label or origin_ip], textposition="top center", textfont=dict(size=12, color="darkred"), name=f"tx entry {origin_ip}", hovertemplate=( f"{origin_ip} (tx entry)
" f"{origin_label}
" f"({o_lat:.4f}, {o_lon:.4f})
" f"reachable: {len(reach_ips)}
" f"max hops: {max_hop}
" f"link_km: {link_km:g}" ), ) ) fig.update_geos( projection_type="natural earth", showcountries=True, countrycolor="#999", countrywidth=0.4, showcoastlines=True, coastlinecolor="#666", coastlinewidth=0.5, showland=True, landcolor="#f5f5f2", showocean=True, oceancolor="#eaf3fa", lonaxis=dict(range=[max(-180, min_lon - 6), min(180, max_lon + 6)]), lataxis=dict(range=[max(-85, min_lat - 6), min(85, max_lat + 6)]), ) fig.update_layout( title=dict( text=( f"TRON tx gossip traversal
" f"origin {origin_ip} " f"({origin_label}) · link_km={link_km:g} · " f"reachable={len(reach_ips)} · hops={max_hop}" ), x=0.02, xanchor="left", ), legend=dict( title="layers (click to toggle)", bgcolor="rgba(255,255,255,0.9)", bordercolor="#888", borderwidth=0.5, x=0.01, y=0.01, xanchor="left", yanchor="bottom", ), margin=dict(l=10, r=10, t=70, b=10), height=780, ) if path is not None: import pickle with open(path, "wb") as fh: pickle.dump(fig, fh) if show: try: fig.show() except Exception: # headless / no-browser environments — skip silently; caller gets fig pass return fig