mirror of
https://github.com/agessaman/meshcore-bot.git
synced 2026-11-02 02:52:22 +00:00
agessaman
49e24ef441
Three self-contained blocks of the 386-line method become `_rollout_classify_repeaters` (fills the repeater list and status counts in place), `_rollout_path_traffic_share` (returns the multi-byte, single-byte and total path observations and the percentage) and `_rollout_daily_trend`. Their statements are AST-identical to the originals and run in the same order inside the same connection and try; inputs and outputs were derived from the AST, and nothing else crosses the boundaries.
722 lines
33 KiB
Python
722 lines
33 KiB
Python
"""Multi-byte path rollout statistics for the web viewer: hop chunks, path encoding badges and relay metrics."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import threading
|
|
from typing import Any
|
|
|
|
|
|
class MultibyteRolloutMixin:
|
|
"""Mixed into BotDataViewer."""
|
|
|
|
_contacts_badge_cache: Any
|
|
_contacts_badge_cache_lock: Any
|
|
_db_connection: Any
|
|
logger: Any
|
|
|
|
@staticmethod
|
|
def _chunks_from_multibyte_path_hex(path_hex: str, bytes_per_hop: int) -> list[str]:
|
|
"""Split path hex into per-hop segments for 2- or 3-byte hop encoding."""
|
|
if not path_hex or bytes_per_hop not in (2, 3):
|
|
return []
|
|
step = bytes_per_hop * 2
|
|
out: list[str] = []
|
|
for i in range(0, len(path_hex), step):
|
|
seg = path_hex[i : i + step]
|
|
if len(seg) == step:
|
|
out.append(seg.lower())
|
|
return out
|
|
|
|
def _collect_multibyte_hop_chunks(
|
|
self, cursor, recent_days: int | None = None
|
|
) -> set[str]:
|
|
"""Hop prefixes from multibyte paths in observed_paths (for repeater/room pubkey matching).
|
|
|
|
If ``recent_days`` is set (e.g. 7), only paths whose ``last_seen`` falls within that
|
|
window are used. Default (None) keeps full history — used by the contacts API badge.
|
|
Dashboard 7d stats pass ``recent_days=7`` so percentages match the chart title.
|
|
"""
|
|
chunks: set[str] = set()
|
|
try:
|
|
extra = ""
|
|
if recent_days is not None:
|
|
d = max(1, min(int(recent_days), 366))
|
|
extra = f" AND date(last_seen) >= date('now', 'localtime', '-{d} days')"
|
|
# DISTINCT collapses the many duplicate (path_hex, bytes_per_hop) advert rows in SQL,
|
|
# so Python only splits each distinct path once instead of every observation.
|
|
cursor.execute(
|
|
f"""
|
|
SELECT DISTINCT path_hex, bytes_per_hop FROM observed_paths
|
|
WHERE bytes_per_hop >= 2 AND bytes_per_hop <= 3
|
|
AND path_hex IS NOT NULL AND length(path_hex) > 0
|
|
{extra}
|
|
"""
|
|
)
|
|
for row in cursor.fetchall():
|
|
ph = row["path_hex"]
|
|
bph = row["bytes_per_hop"]
|
|
try:
|
|
bph_i = int(bph) if bph is not None else 0
|
|
except (TypeError, ValueError):
|
|
bph_i = 0
|
|
for c in self._chunks_from_multibyte_path_hex(ph, bph_i):
|
|
if len(c) in (4, 6):
|
|
chunks.add(c)
|
|
except Exception as e:
|
|
self.logger.debug(f"Could not load multibyte hop chunks: {e}")
|
|
return chunks
|
|
|
|
def _get_cached_contact_multibyte_hop_chunks(
|
|
self, cursor, recent_days: int | None = None
|
|
) -> set[str]:
|
|
"""Return contact badge evidence without rebuilding it per page request.
|
|
|
|
Memoized per ``recent_days`` window: the contacts list wants all-time
|
|
evidence (None) while the dashboard wants the 7-day set, and both are
|
|
rebuilt often enough that sharing one slot would thrash the cache.
|
|
"""
|
|
try:
|
|
# last_seen changes when an existing path is observed again; count changes on
|
|
# inserts and retention deletes. Together they cheaply invalidate the cache while
|
|
# using the multibyte covering index rather than the wide observed_paths table.
|
|
cursor.execute(
|
|
"""
|
|
SELECT MAX(last_seen), COUNT(*) FROM observed_paths
|
|
WHERE bytes_per_hop >= 2 AND bytes_per_hop <= 3
|
|
"""
|
|
)
|
|
signature_row = cursor.fetchone()
|
|
signature = tuple(signature_row) if signature_row else (None, 0)
|
|
if recent_days is not None:
|
|
# A windowed set also changes when the window itself slides, which
|
|
# leaves the row fingerprint untouched — pin it to the local date.
|
|
cursor.execute("SELECT date('now', 'localtime')")
|
|
signature = (*signature, cursor.fetchone()[0])
|
|
except Exception as e:
|
|
self.logger.debug(f"Could not fingerprint multibyte hop chunks: {e}")
|
|
return self._collect_multibyte_hop_chunks(cursor, recent_days=recent_days)
|
|
|
|
lock = getattr(self, '_contacts_badge_cache_lock', None)
|
|
if lock is None:
|
|
lock = self._contacts_badge_cache_lock = threading.Lock()
|
|
with lock:
|
|
cache = getattr(self, '_contacts_badge_cache', None)
|
|
if cache is None:
|
|
cache = self._contacts_badge_cache = {}
|
|
cached = cache.get(recent_days)
|
|
if cached is not None and cached[0] == signature:
|
|
return cached[1]
|
|
chunks = self._collect_multibyte_hop_chunks(cursor, recent_days=recent_days)
|
|
cache[recent_days] = (signature, chunks)
|
|
return chunks
|
|
|
|
@staticmethod
|
|
def _bucket_hop_chunks(multibyte_hop_chunks: set[str]) -> dict[int, set[str]]:
|
|
"""Bucket hop-prefix chunks by length (4 or 6) for O(1) prefix matching.
|
|
|
|
Build this once per request and reuse across the per-contact badge loop instead of
|
|
scanning the whole chunk set for every contact.
|
|
"""
|
|
return {
|
|
4: {c for c in multibyte_hop_chunks if len(c) == 4},
|
|
6: {c for c in multibyte_hop_chunks if len(c) == 6},
|
|
}
|
|
|
|
def _compute_path_encoding_badge(
|
|
self,
|
|
row: Any,
|
|
all_paths: list[dict[str, Any]],
|
|
chunk_buckets: dict[int, set[str]],
|
|
) -> str | None:
|
|
"""Return 'multibyte', 'one_byte', or None for contacts path-encoding badge.
|
|
|
|
``chunk_buckets`` is the length-bucketed form of the multibyte hop chunks
|
|
(see _bucket_hop_chunks).
|
|
"""
|
|
pk = row["public_key"] or ""
|
|
role = (row["role"] or "").lower()
|
|
obph_raw = row["out_bytes_per_hop"]
|
|
obph: int | None
|
|
try:
|
|
obph = int(obph_raw) if obph_raw is not None else None
|
|
except (TypeError, ValueError):
|
|
obph = None
|
|
if obph is not None and obph not in (1, 2, 3):
|
|
obph = None
|
|
|
|
out_path_len = row["out_path_len"]
|
|
if out_path_len is None:
|
|
out_path_len = -1
|
|
try:
|
|
out_path_len = int(out_path_len)
|
|
except (TypeError, ValueError):
|
|
out_path_len = -1
|
|
|
|
advert_count = row["advert_count"] or 0
|
|
|
|
def norm_bph(b: Any) -> int:
|
|
if b is None:
|
|
return 1
|
|
try:
|
|
i = int(b)
|
|
return i if i in (1, 2, 3) else 1
|
|
except (TypeError, ValueError):
|
|
return 1
|
|
|
|
# Multibyte evidence
|
|
if obph in (2, 3):
|
|
return "multibyte"
|
|
for p in all_paths:
|
|
if norm_bph(p.get("bytes_per_hop")) in (2, 3):
|
|
return "multibyte"
|
|
if role in ("repeater", "roomserver") and pk:
|
|
pk_low = pk.lower()
|
|
# Chunks are fixed-length (4 or 6), so startswith reduces to prefix-equality lookups.
|
|
if pk_low[:4] in chunk_buckets[4] or pk_low[:6] in chunk_buckets[6]:
|
|
return "multibyte"
|
|
|
|
# One-byte: positive signal and no multibyte observation
|
|
has_signal = bool(
|
|
advert_count > 0 or len(all_paths) > 0 or out_path_len >= 0
|
|
)
|
|
if not has_signal:
|
|
return None
|
|
|
|
if obph is not None and obph != 1:
|
|
return None
|
|
for p in all_paths:
|
|
if norm_bph(p.get("bytes_per_hop")) != 1:
|
|
return None
|
|
|
|
return "one_byte"
|
|
|
|
def _contact_has_multibyte_path_evidence(
|
|
self,
|
|
public_key: str,
|
|
role: str | None,
|
|
out_bytes_per_hop: Any,
|
|
multibyte_advert_public_keys: set[str],
|
|
chunk_buckets: dict[int, set[str]],
|
|
) -> bool:
|
|
"""Multibyte detection for dashboard 7d stats (observed_paths scoped by date in SQL).
|
|
|
|
``chunk_buckets`` is the length-bucketed form of the multibyte hop chunks
|
|
(see _bucket_hop_chunks) — build it once per request, not per contact.
|
|
"""
|
|
pk = public_key or ""
|
|
role_l = (role or "").lower()
|
|
obph: int | None
|
|
try:
|
|
obph = int(out_bytes_per_hop) if out_bytes_per_hop is not None else None
|
|
except (TypeError, ValueError):
|
|
obph = None
|
|
if obph is not None and obph not in (1, 2, 3):
|
|
obph = None
|
|
|
|
if obph in (2, 3):
|
|
return True
|
|
if pk and pk in multibyte_advert_public_keys:
|
|
return True
|
|
if role_l in ("repeater", "roomserver") and pk:
|
|
# Chunks are fixed-length (4 or 6), so startswith reduces to prefix-equality
|
|
# lookups — O(1) per contact instead of a scan of the whole chunk set.
|
|
pk_low = pk.lower()
|
|
if pk_low[:4] in chunk_buckets[4] or pk_low[:6] in chunk_buckets[6]:
|
|
return True
|
|
return False
|
|
|
|
def _compute_single_byte_relay_metrics(
|
|
self, cursor, path_time_cond: str
|
|
) -> dict[str, dict]:
|
|
"""Per-1-byte-prefix relay metrics derived from single-byte advert paths.
|
|
|
|
Senders who also appear in multibyte advert paths in the same window are excluded —
|
|
their single-byte paths are parallel/legacy routes that do not need this relay upgraded.
|
|
|
|
Returns: {hex_prefix_2chars: {relay_score, unique_sources, unique_dests}}
|
|
"""
|
|
try:
|
|
cursor.execute(
|
|
f"""
|
|
SELECT DISTINCT public_key FROM observed_paths
|
|
WHERE bytes_per_hop IN (2,3) AND packet_type = 'advert'
|
|
AND public_key IS NOT NULL {path_time_cond}
|
|
"""
|
|
)
|
|
mb_senders: set[str] = {r['public_key'] for r in cursor.fetchall()}
|
|
|
|
cursor.execute(
|
|
f"""
|
|
SELECT path_hex, public_key, to_prefix, observation_count
|
|
FROM observed_paths
|
|
WHERE bytes_per_hop = 1 AND packet_type = 'advert'
|
|
AND public_key IS NOT NULL {path_time_cond}
|
|
"""
|
|
)
|
|
|
|
metrics: dict[str, dict] = {}
|
|
for row in cursor.fetchall():
|
|
if row['public_key'] in mb_senders:
|
|
continue
|
|
ph = (row['path_hex'] or '').lower()
|
|
if not ph:
|
|
continue
|
|
obs = row['observation_count'] or 1
|
|
for i in range(0, len(ph) - 1, 2):
|
|
chunk = ph[i:i + 2]
|
|
if len(chunk) != 2:
|
|
continue
|
|
if chunk not in metrics:
|
|
metrics[chunk] = {'relay_score': 0, 'sources': set(), 'to_prefixes': set()}
|
|
metrics[chunk]['relay_score'] += obs
|
|
metrics[chunk]['sources'].add(row['public_key'])
|
|
if row['to_prefix']:
|
|
metrics[chunk]['to_prefixes'].add(row['to_prefix'])
|
|
|
|
return {
|
|
pfx: {
|
|
'relay_score': v['relay_score'],
|
|
'unique_sources': len(v['sources']),
|
|
'unique_dests': len(v['to_prefixes']),
|
|
}
|
|
for pfx, v in metrics.items()
|
|
}
|
|
except Exception as e:
|
|
self.logger.debug(f"Could not compute single-byte relay metrics: {e}")
|
|
return {}
|
|
|
|
def _compute_legacy_degree(
|
|
self, cursor, unupgraded_pks: set[str], mc_since_cond: str
|
|
) -> dict[str, int]:
|
|
"""Count mesh_connections edges where both endpoints are unupgraded (single_byte) nodes.
|
|
|
|
Uses from_public_key / to_public_key for precise matching — no 1-byte collision ambiguity.
|
|
Returns: {public_key: edge_count}
|
|
"""
|
|
try:
|
|
cursor.execute(
|
|
f"""
|
|
SELECT from_public_key, to_public_key
|
|
FROM mesh_connections
|
|
WHERE from_public_key IS NOT NULL AND to_public_key IS NOT NULL
|
|
{mc_since_cond}
|
|
"""
|
|
)
|
|
degree: dict[str, int] = {}
|
|
for row in cursor.fetchall():
|
|
fk, tk = row['from_public_key'], row['to_public_key']
|
|
if fk in unupgraded_pks and tk in unupgraded_pks:
|
|
degree[fk] = degree.get(fk, 0) + 1
|
|
degree[tk] = degree.get(tk, 0) + 1
|
|
return degree
|
|
except Exception as e:
|
|
self.logger.debug(f"Could not compute legacy degree: {e}")
|
|
return {}
|
|
|
|
def _get_multibyte_rollout_data(self, since: str = '30d', node_type: str = 'all') -> dict:
|
|
"""Return multibyte hash rollout analytics for repeaters and/or room servers.
|
|
|
|
since: 24h | 7d | 30d | 90d | all — window applied to last_heard in complete_contact_tracking.
|
|
node_type: all | repeater | roomserver — filters which relay node types are included.
|
|
The daily_trend array always covers the last 30 days regardless of ``since``.
|
|
"""
|
|
try:
|
|
with self._db_connection() as conn:
|
|
cursor = conn.cursor()
|
|
|
|
# Role filter based on node_type
|
|
if node_type == 'repeater':
|
|
role_filter = "c.role = 'repeater'"
|
|
elif node_type == 'roomserver':
|
|
role_filter = "c.role = 'roomserver'"
|
|
else:
|
|
role_filter = "c.role IN ('repeater', 'roomserver')"
|
|
|
|
# last_heard is stored as a Python datetime serialised to ISO text by sqlite3
|
|
# (e.g. '2026-05-17 10:30:00.123456'). Use SQLite's datetime() so the comparison
|
|
# is ISO-text vs ISO-text, which sorts correctly lexicographically.
|
|
datetime_offsets = {
|
|
'24h': "'-24 hours'",
|
|
'7d': "'-7 days'",
|
|
'30d': "'-30 days'",
|
|
'90d': "'-90 days'",
|
|
}
|
|
if since in datetime_offsets:
|
|
where_clause = (
|
|
f"WHERE {role_filter}"
|
|
f" AND c.last_heard >= datetime('now', 'localtime', {datetime_offsets[since]})"
|
|
)
|
|
else:
|
|
where_clause = f"WHERE {role_filter}"
|
|
|
|
# Check for observed_paths table
|
|
cursor.execute(
|
|
"SELECT name FROM sqlite_master WHERE type='table' AND name='observed_paths'"
|
|
)
|
|
has_observed_paths = cursor.fetchone() is not None
|
|
|
|
# Collect multibyte hop chunks for prefix matching, scoped to the same window as the
|
|
# contact filter so the rollout numbers are consistent with the dashboard's 7d stats.
|
|
_since_to_days = {'24h': 1, '7d': 7, '30d': 30, '90d': 90}
|
|
_chunk_recent_days = _since_to_days.get(since) # None → all-time for 'all'
|
|
multibyte_hop_chunks: set[str] = set()
|
|
if has_observed_paths:
|
|
multibyte_hop_chunks = self._collect_multibyte_hop_chunks(
|
|
cursor, recent_days=_chunk_recent_days
|
|
)
|
|
chunk_buckets = self._bucket_hop_chunks(multibyte_hop_chunks)
|
|
|
|
# Per-repeater path traffic totals (advert paths only)
|
|
path_traffic: dict[str, dict] = {}
|
|
if has_observed_paths:
|
|
cursor.execute(
|
|
"""
|
|
SELECT public_key,
|
|
SUM(observation_count) as total_traffic,
|
|
SUM(CASE WHEN bytes_per_hop IN (2,3) THEN observation_count ELSE 0 END) as mb_traffic,
|
|
MAX(last_seen) as latest_path_seen
|
|
FROM observed_paths
|
|
WHERE public_key IS NOT NULL AND packet_type = 'advert'
|
|
GROUP BY public_key
|
|
"""
|
|
)
|
|
for pt_row in cursor.fetchall():
|
|
path_traffic[pt_row['public_key']] = {
|
|
'total_traffic': pt_row['total_traffic'] or 0,
|
|
'mb_traffic': pt_row['mb_traffic'] or 0,
|
|
'latest_path_seen': pt_row['latest_path_seen'],
|
|
}
|
|
|
|
# Scope the path history to the same window so badge evidence matches the since filter.
|
|
path_time_cond = ""
|
|
if since in datetime_offsets:
|
|
path_time_cond = f"AND last_seen >= datetime('now', 'localtime', {datetime_offsets[since]})"
|
|
|
|
# Fetch the relay contacts directly (no join/group-by); their recent advert paths are
|
|
# loaded separately below and assembled in Python. This mirrors _get_tracking_data and
|
|
# keeps the planner on the idx_observed_paths_advert_pk_seen covering index for the path
|
|
# scan instead of building a runtime automatic index for a LEFT JOIN + GROUP_CONCAT.
|
|
cursor.execute(
|
|
f"""
|
|
SELECT
|
|
c.public_key, c.name, c.role, c.device_type,
|
|
c.latitude, c.longitude, c.city, c.state, c.country,
|
|
c.first_heard, c.last_heard,
|
|
c.advert_count, c.out_bytes_per_hop, c.out_path_len
|
|
FROM complete_contact_tracking c
|
|
{where_clause}
|
|
ORDER BY c.last_heard DESC
|
|
"""
|
|
)
|
|
main_rows = cursor.fetchall()
|
|
|
|
# Recent advert paths per contact (most-recent 50), grouped in Python by public_key.
|
|
# With idx_observed_paths_advert_pk_seen this runs as an ordered covering index scan.
|
|
# We don't filter to the relay keys here: the loop only looks up paths for contacts in
|
|
# main_rows, and joining the key set pushes the planner off the covering index.
|
|
paths_by_key: dict[str, list[dict]] = {}
|
|
if has_observed_paths:
|
|
cursor.execute(
|
|
f"""
|
|
WITH recent_paths AS (
|
|
SELECT public_key, path_hex, bytes_per_hop, observation_count, last_seen,
|
|
ROW_NUMBER() OVER (PARTITION BY public_key ORDER BY last_seen DESC) as rn
|
|
FROM observed_paths
|
|
WHERE packet_type = 'advert' AND public_key IS NOT NULL
|
|
{path_time_cond}
|
|
)
|
|
SELECT public_key, path_hex, bytes_per_hop, observation_count, last_seen
|
|
FROM recent_paths WHERE rn <= 50
|
|
ORDER BY public_key, last_seen DESC
|
|
"""
|
|
)
|
|
for prow in cursor.fetchall():
|
|
ph = prow['path_hex']
|
|
if not ph:
|
|
continue
|
|
bph = 1
|
|
if prow['bytes_per_hop'] is not None:
|
|
try:
|
|
bph = int(prow['bytes_per_hop'])
|
|
if bph not in (1, 2, 3):
|
|
bph = 1
|
|
except (TypeError, ValueError):
|
|
bph = 1
|
|
paths_by_key.setdefault(prow['public_key'], []).append({
|
|
'path_hex': ph,
|
|
'bytes_per_hop': bph,
|
|
'observation_count': int(prow['observation_count']) if prow['observation_count'] is not None else 1,
|
|
'last_seen': prow['last_seen'] if prow['last_seen'] is not None else None,
|
|
})
|
|
|
|
# Classify each repeater
|
|
status_counts: dict[str, int] = {
|
|
'multibyte_direct': 0,
|
|
'multibyte_relayed': 0,
|
|
'single_byte': 0,
|
|
'unknown': 0,
|
|
}
|
|
all_repeaters: list[dict] = []
|
|
|
|
self._rollout_classify_repeaters(
|
|
all_repeaters, chunk_buckets, main_rows, path_traffic, paths_by_key, status_counts
|
|
)
|
|
|
|
# ── Priority scoring ──────────────────────────────────────────────────
|
|
unupgraded_pks: set[str] = set()
|
|
# Build prefix → all nodes (any status) for complete prefix_peers reporting.
|
|
# Confirmed-multibyte nodes sharing a 1-byte prefix may be the actual relay
|
|
# in those paths, so surfacing them in the tooltip is important.
|
|
all_by_prefix: dict[str, list[dict]] = {}
|
|
for r in all_repeaters:
|
|
if r['public_key']:
|
|
pfx = r['public_key'][:2].lower()
|
|
all_by_prefix.setdefault(pfx, []).append(r)
|
|
if r['status'] == 'single_byte' and r['public_key']:
|
|
unupgraded_pks.add(r['public_key'])
|
|
|
|
sb_metrics: dict[str, dict] = {}
|
|
if has_observed_paths and unupgraded_pks:
|
|
sb_metrics = self._compute_single_byte_relay_metrics(cursor, path_time_cond)
|
|
|
|
legacy_degree: dict[str, int] = {}
|
|
try:
|
|
cursor.execute(
|
|
"SELECT name FROM sqlite_master WHERE type='table' AND name='mesh_connections'"
|
|
)
|
|
if cursor.fetchone() is not None and unupgraded_pks:
|
|
mc_since_cond = ""
|
|
if since in datetime_offsets:
|
|
mc_since_cond = (
|
|
f"AND last_seen >= datetime('now', 'localtime', {datetime_offsets[since]})"
|
|
)
|
|
legacy_degree = self._compute_legacy_degree(
|
|
cursor, unupgraded_pks, mc_since_cond
|
|
)
|
|
except Exception as e:
|
|
self.logger.debug(f"Legacy degree skipped: {e}")
|
|
|
|
for r in all_repeaters:
|
|
if r['status'] == 'single_byte' and r['public_key']:
|
|
pfx = r['public_key'][:2].lower()
|
|
m = sb_metrics.get(pfx, {})
|
|
r['relay_score'] = m.get('relay_score', 0)
|
|
r['unique_sources'] = m.get('unique_sources', 0)
|
|
r['unique_dests'] = m.get('unique_dests', 0)
|
|
r['legacy_degree'] = legacy_degree.get(r['public_key'], 0)
|
|
# prefix_peers: ALL other nodes (any status) sharing this 1-byte prefix.
|
|
# Confirmed-multibyte peers explain inflated scores on low-traffic nodes.
|
|
r['prefix_peers'] = [
|
|
{'public_key': r2['public_key'], 'name': r2['name'], 'status': r2['status']}
|
|
for r2 in all_by_prefix.get(pfx, [])
|
|
if r2['public_key'] != r['public_key']
|
|
]
|
|
else:
|
|
r['relay_score'] = 0
|
|
r['unique_sources'] = 0
|
|
r['unique_dests'] = 0
|
|
r['legacy_degree'] = 0
|
|
r['prefix_peers'] = []
|
|
|
|
# ── Name-based suppression ────────────────────────────────────────────
|
|
# A single_byte record that shares a name with a confirmed-multibyte
|
|
# record is almost certainly the same physical device — the single_byte
|
|
# entry reflects stale traffic from before the firmware upgrade.
|
|
# Suppress it so it doesn't appear as an upgrade target or inflate counts.
|
|
multibyte_names: set[str] = {
|
|
r['name'].strip().lower()
|
|
for r in all_repeaters
|
|
if r['status'] in ('multibyte_direct', 'multibyte_relayed') and r['name']
|
|
}
|
|
if multibyte_names:
|
|
kept: list[dict] = []
|
|
for r in all_repeaters:
|
|
if (r['status'] == 'single_byte'
|
|
and r['name']
|
|
and r['name'].strip().lower() in multibyte_names):
|
|
status_counts['single_byte'] -= 1
|
|
else:
|
|
kept.append(r)
|
|
all_repeaters = kept
|
|
|
|
# Priority list: single_byte nodes ranked by relay_score desc
|
|
priority_list = sorted(
|
|
[r for r in all_repeaters if r['status'] == 'single_byte'],
|
|
key=lambda r: (-(r['relay_score'] or 0), -(r['unique_sources'] or 0)),
|
|
)
|
|
|
|
# Summary
|
|
total = len(all_repeaters)
|
|
mb_total = status_counts['multibyte_direct'] + status_counts['multibyte_relayed']
|
|
adoption_pct = round(mb_total / total * 100, 1) if total > 0 else 0.0
|
|
|
|
multibyte_path_obs, single_byte_path_obs, total_path_obs, traffic_mb_pct = (
|
|
self._rollout_path_traffic_share(cursor, datetime_offsets, has_observed_paths, node_type, since)
|
|
)
|
|
|
|
daily_trend = self._rollout_daily_trend(cursor, has_observed_paths, node_type)
|
|
|
|
return {
|
|
'since': since,
|
|
'node_type': node_type,
|
|
'summary': {
|
|
'total_repeaters': total,
|
|
'multibyte_direct': status_counts['multibyte_direct'],
|
|
'multibyte_relayed': status_counts['multibyte_relayed'],
|
|
'single_byte': status_counts['single_byte'],
|
|
'unknown': status_counts['unknown'],
|
|
'adoption_pct': adoption_pct,
|
|
'total_path_obs': total_path_obs,
|
|
'multibyte_path_obs': multibyte_path_obs,
|
|
'single_byte_path_obs': single_byte_path_obs,
|
|
'traffic_multibyte_pct': traffic_mb_pct,
|
|
},
|
|
'priority_list': priority_list,
|
|
'all_repeaters': all_repeaters,
|
|
'daily_trend': daily_trend,
|
|
}
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Error getting multibyte rollout data: {e}")
|
|
return {
|
|
'since': since,
|
|
'error': str(e),
|
|
'summary': {
|
|
'total_repeaters': 0, 'multibyte_direct': 0,
|
|
'multibyte_relayed': 0, 'single_byte': 0, 'unknown': 0,
|
|
'adoption_pct': 0.0, 'total_path_obs': 0,
|
|
'multibyte_path_obs': 0, 'single_byte_path_obs': 0,
|
|
'traffic_multibyte_pct': 0.0,
|
|
},
|
|
'priority_list': [],
|
|
'all_repeaters': [],
|
|
'daily_trend': [],
|
|
}
|
|
|
|
def _rollout_classify_repeaters(
|
|
self, all_repeaters, chunk_buckets, main_rows, path_traffic, paths_by_key, status_counts
|
|
):
|
|
"""Classify each repeater row into *all_repeaters* and tally *status_counts*."""
|
|
for row in main_rows:
|
|
pk = row['public_key']
|
|
|
|
# Recent advert paths for this contact (grouped from the path query above)
|
|
all_paths = paths_by_key.get(pk, [])
|
|
|
|
badge = self._compute_path_encoding_badge(row, all_paths, chunk_buckets)
|
|
|
|
obph_raw = row['out_bytes_per_hop']
|
|
try:
|
|
obph: int | None = int(obph_raw) if obph_raw is not None else None
|
|
except (TypeError, ValueError):
|
|
obph = None
|
|
|
|
if badge == 'multibyte':
|
|
status = 'multibyte_direct' if obph in (2, 3) else 'multibyte_relayed'
|
|
elif badge == 'one_byte':
|
|
status = 'single_byte'
|
|
else:
|
|
status = 'unknown'
|
|
|
|
status_counts[status] += 1
|
|
|
|
loc_parts = [p for p in [row['city'], row['state'], row['country']] if p]
|
|
|
|
pt = path_traffic.get(pk or '', {})
|
|
all_repeaters.append({
|
|
'public_key': pk,
|
|
'name': row['name'],
|
|
'role': row['role'],
|
|
'device_type': row['device_type'],
|
|
'status': status,
|
|
'out_bytes_per_hop': obph,
|
|
'advert_count': row['advert_count'] or 0,
|
|
'total_traffic': pt.get('total_traffic', 0),
|
|
'last_seen': row['last_heard'],
|
|
'first_heard': row['first_heard'],
|
|
'location': ', '.join(loc_parts),
|
|
'latitude': row['latitude'],
|
|
'longitude': row['longitude'],
|
|
})
|
|
|
|
def _rollout_path_traffic_share(self, cursor, datetime_offsets, has_observed_paths, node_type, since):
|
|
"""Multi-byte share of observed path traffic: multi-byte, single-byte and total observations, and the percentage."""
|
|
# Overall path observation traffic breakdown (advert paths, within the since window,
|
|
# for nodes matching the node_type filter)
|
|
total_path_obs = multibyte_path_obs = single_byte_path_obs = 0
|
|
traffic_mb_pct = 0.0
|
|
if has_observed_paths:
|
|
# observed_paths.last_seen is an ISO timestamp → use datetime() comparison
|
|
if node_type == 'repeater':
|
|
obs_role_filter = "c.role = 'repeater'"
|
|
elif node_type == 'roomserver':
|
|
obs_role_filter = "c.role = 'roomserver'"
|
|
else:
|
|
obs_role_filter = "c.role IN ('repeater', 'roomserver')"
|
|
|
|
obs_time_cond = ""
|
|
if since in datetime_offsets:
|
|
obs_time_cond = f" AND op.last_seen >= datetime('now', 'localtime', {datetime_offsets[since]})"
|
|
|
|
cursor.execute(
|
|
f"""
|
|
SELECT op.bytes_per_hop, SUM(op.observation_count) as n
|
|
FROM observed_paths op
|
|
JOIN complete_contact_tracking c ON c.public_key = op.public_key
|
|
WHERE op.packet_type = 'advert' AND {obs_role_filter}{obs_time_cond}
|
|
GROUP BY op.bytes_per_hop
|
|
"""
|
|
)
|
|
for tr in cursor.fetchall():
|
|
n = tr['n'] or 0
|
|
total_path_obs += n
|
|
bph = tr['bytes_per_hop'] or 1
|
|
try:
|
|
bph = int(bph)
|
|
except (TypeError, ValueError):
|
|
bph = 1
|
|
if bph in (2, 3):
|
|
multibyte_path_obs += n
|
|
else:
|
|
single_byte_path_obs += n
|
|
|
|
if total_path_obs > 0:
|
|
traffic_mb_pct = round(multibyte_path_obs / total_path_obs * 100, 1)
|
|
return multibyte_path_obs, single_byte_path_obs, total_path_obs, traffic_mb_pct
|
|
|
|
def _rollout_daily_trend(self, cursor, has_observed_paths, node_type):
|
|
"""Daily multi-byte and single-byte path observations over the last 30 days, for the trend chart."""
|
|
# Daily trend: last 30 days, filtered by node_type
|
|
daily_trend: list[dict] = []
|
|
if has_observed_paths:
|
|
if node_type == 'repeater':
|
|
trend_role_filter = "c.role = 'repeater'"
|
|
elif node_type == 'roomserver':
|
|
trend_role_filter = "c.role = 'roomserver'"
|
|
else:
|
|
trend_role_filter = "c.role IN ('repeater', 'roomserver')"
|
|
|
|
cursor.execute(
|
|
f"""
|
|
SELECT date(op.last_seen) as obs_date,
|
|
SUM(CASE WHEN op.bytes_per_hop IN (2,3)
|
|
THEN op.observation_count ELSE 0 END) as mb_obs,
|
|
SUM(CASE WHEN op.bytes_per_hop = 1 OR op.bytes_per_hop IS NULL
|
|
THEN op.observation_count ELSE 0 END) as sb_obs
|
|
FROM observed_paths op
|
|
JOIN complete_contact_tracking c ON c.public_key = op.public_key
|
|
WHERE {trend_role_filter}
|
|
AND op.last_seen >= datetime('now', 'localtime', '-30 days')
|
|
AND op.packet_type = 'advert'
|
|
GROUP BY date(op.last_seen)
|
|
ORDER BY obs_date
|
|
"""
|
|
)
|
|
for tr in cursor.fetchall():
|
|
daily_trend.append({
|
|
'date': tr['obs_date'],
|
|
'multibyte': tr['mb_obs'] or 0,
|
|
'single_byte': tr['sb_obs'] or 0,
|
|
})
|
|
return daily_trend
|