' + return ret + + def showHistogram(self): + ret = "" + histo = self._histogram + nloops = int(histo["loops"]) + if nloops > 0: + ret += "
APs per Channel over %s epochs
" % nloops + ret += "| Channel | " + totals = "||||
|---|---|---|---|---|
| APs seen | " + vals = "||||
| Avg APs/epoch | " + + for (ch, count) in sorted(histo.items(), key=lambda x: x[1], reverse=True): + if ch == "loops": + pass + else: + weight = float(count) / nloops + # ret +="||||
| %d | %0.2f | " % (ch, count) + chans += "%s | " % ch + totals += "%d | " % count + vals += "%0.1f | " % weight + chans += "
No channel data collected yet
" + + return ret + + def showInteractions(self): + ret = "" + numHidden = 0 + numVisible = 0 + if self._agent: + now = time.time() + ret += "Interactions per endpoint
" + ret += "Encounters is how many different times this AP has been seen, then not seen, then seen again. Interactions should be the sum of assoc and deauth attacks. All are per session stats. Age is seconds since AP was last seen by the plugin.
" + ret += "| Hostname | MAC | Channel | Age | RSSI | Encounters | Associates | Deauths | Handshakes | Interactions |
|---|---|---|---|---|---|---|---|---|---|
| %s | " % html.escape(ap['hostname']) + else: + ret += "|||||||||
| %s | " % html.escape( + ap['hostname']) # italicise hosts not currently visible + ret += "%s | %s | " % (ap['mac'], ap['channel']) + ret += "%d | " % int(now - ap['AT_lastseen']) # time since last interaction + ret += "%s | " % ap['rssi'] + for t in ['seen', 'assoc', 'deauth', 'handshake']: + tag = 'AT_' + t + if tag in ap: + ret += "%s | " % ap[tag] + else: + ret += "" + if lmac in self._agent._history: + ret += " | %s | " % self._agent._history[lmac] + else: + ret += "no attacks yet | " + ret += "|
| %s | %s | %s |
" % (numVisible, numHidden)
+
+ return ret
+
+ def update_parameter(self, cfg, parameter, vtype, val, ret):
+ changed = False
+ if parameter in cfg:
+ old_val = cfg[parameter]
+
+ if val == old_val:
+ pass
+ elif vtype == "int":
+ cfg[parameter] = int(val)
+ changed = True
+ elif vtype == "float":
+ cfg[parameter] = float(val)
+ ret += "Updated float %s: %s -> %s "
+ ret += self.showEditForm(request)
+
+ ret += self.showHistogram()
+ ret += self.showChistos()
+ if 'show_interactions' in self.options and self.options['show_interactions']:
+ ret += self.showInteractions()
+ ret += ""
+ return render_template_string(ret)
+ # other paths here
+ elif request.method == "POST":
+ ret = ' " % repr(path)
+ ret += " " % repr(request.values)
+ ret += ""
+ return render_template_string(ret)
+ except Exception as e:
+ ret = "
\n" % (parameter, old_val, val)
+ changed = True
+ elif vtype == "bool":
+ cfg[parameter] = bool(val == "True")
+ ret += "Updated boolean %s: %s -> %s
\n" % (parameter, old_val, val)
+ changed = True
+ elif vtype == "str":
+ cfg[parameter] = val
+ ret += "Updated string %s: %s -> %s
\n" % (parameter, old_val, val)
+ changed = True
+ else:
+ ret += "No update %s (%s): %s -> %s
\n" % (parameter, type, old_val, val)
+
+ return changed
+
+ # called when http://AUTO Tune not ready
"
+ return render_template_string(ret)
+
+ try:
+ if request.method == "GET":
+ if path == "/" or not path:
+ logging.debug("webhook called")
+ ret = 'AUTO Tune
AUTO Tune Update
"
+
+ # Handle preset operations
+ if 'save_preset' in request.values and 'preset_name' in request.values:
+ preset_name = request.values['preset_name'].strip()
+ if preset_name:
+ try:
+ self._save_preset(preset_name)
+ ret += "Processing changes
"
+ changed = False
+ for (key, val) in request.values.items():
+ if key != "":
+ # ret += "%s -> %s
"
+ if changed:
+ save_config(self._agent._config, "/etc/pwnagotchi/config.toml")
+ ret += self.showEditForm(request)
+ ret += self.showHistogram()
+ ret += self.showChistos()
+ ret += ""
+ else:
+ ret += "
\n" % (key,val)
+ try:
+ if key.startswith('newval,'):
+ (tag, value, parameter, vtype) = key.split(",", 4)
+ if value == val:
+ logging.debug("Skip unchanged value")
+ continue
+
+ if parameter in self._agent._config['personality']:
+ logging.debug("Personality update")
+ chg = self.update_parameter(self._agent._config['personality'], parameter,
+ vtype, val, ret)
+ elif parameter in self.options:
+ logging.debug("plugin settings update")
+ chg = self.update_parameter(self.options, parameter, vtype, val, ret)
+ else:
+ ret += "
\n" % (key, val)
+ except Exception as e:
+ ret += "Error
%s
" % repr(e)
+ logging.exception(e)
+ ret += "Unknown request
"
+ ret += '' % int(time.time())
+ ret += "
Path
%sRequest
%s%s
" % repr(e)
+ logging.exception("AUTO Tune error: %s" % repr(e))
+ return render_template_string(ret)
+
+ # called when the plugin is loaded
+ def on_loaded(self):
+ try:
+ defaults = {'show_hidden': False,
+ 'reset_history': True,
+ 'extra_channels': 15,
+ 'show_interactions': False,
+ }
+
+ for d in defaults:
+ if d not in self.options:
+ self.options[d] = defaults[d]
+ except Exception as e:
+ logging.exception(e)
+
+ def on_ready(self, agent):
+ self._agent = agent
+ if self.options['reset_history']:
+ self._agent._history = {} # clear "max_interactions" data
+ self._agent.run("wifi.recon clear")
+ self._agent.run("wifi.clear")
+
+ # called when the agent refreshed its access points list
+ def on_wifi_update(self, agent, access_points):
+ # check aps and update active channels
+ try:
+ active_channels = []
+ self._histogram["loops"] = self._histogram["loops"] + 1
+ for ap in access_points:
+ self.markAPSeen(ap, 'wifi_update')
+ ch = ap['channel']
+ logging.debug("%s %d" % (ap['hostname'], ch))
+ if ch not in active_channels:
+ active_channels.append(ch)
+ if ch in self._unscanned_channels:
+ self._unscanned_channels.remove(ch)
+ self._histogram[ch] = 1 if ch not in self._histogram else self._histogram[ch] + 1
+
+ self._active_channels = active_channels
+ logging.info("Histo: %s" % repr(self._histogram))
+ except Exception as e:
+ logging.exception(e)
+
+ # called when the agent refreshed an unfiltered access point list
+ # this list contains all access points that were detected BEFORE filtering
+ # def on_unfiltered_ap_list(self, agent, access_points):
+ # pass
+
+ # called when an epoch is over (where an epoch is a single loop of the main algorithm)
+ def on_epoch(self, agent, epoch, epoch_data):
+ # pick set of channels for next time
+ try:
+ next_channels = self._active_channels.copy()
+ n = 3 if "extra_channels" not in self.options else self.options["extra_channels"]
+ if len(self._unscanned_channels) == 0:
+ if "restrict_channels" in self.options:
+ logging.info("Repopulating from restricted list")
+ self._unscanned_channels = self.options["restrict_channels"].copy()
+ elif hasattr(agent, "_allowed_channels"):
+ logging.info("Repopulating from allowed list: %s" % agent._allowed_channels)
+ self._unscanned_channels = agent._allowed_channels.copy()
+ elif hasattr(agent, "_supported_channels"):
+ logging.info("Repopulating from supported list")
+ self._unscanned_channels = agent._supported_channels.copy()
+ else:
+ logging.info("Repopulating unscanned list")
+ self._unscanned_channels = pwnagotchi.utils.iface_channels(agent._config['main']['iface'])
+
+ for i in range(n):
+ if len(self._unscanned_channels):
+ ch = random.choice(list(self._unscanned_channels))
+ self._unscanned_channels.remove(ch)
+ next_channels.append(ch)
+ # update live config
+ agent._config['personality']['channels'] = next_channels
+ logging.info("Active: %s, Next scan: %s, yet unscanned: %d %s" % (
+ self._active_channels, next_channels, len(self._unscanned_channels), self._unscanned_channels))
+ except Exception as e:
+ logging.exception(e)
+
+ def markAPSeen(self, access_point, context=None):
+ try:
+ apname = self.normalize(access_point['hostname'])
+ apmac = self.normalize(access_point['mac'])
+ apID = apname + '-' + apmac
+ channel = access_point['channel']
+
+ contextlabel = " on " + context if context else ""
+ tag = 'AT_' + context if context else 'AT_seen'
+
+ if apID not in self._known_aps:
+ # first time seen this AP
+ self._known_aps[apID] = access_point.copy()
+ self._known_aps[apID]['AT_seen'] = 1
+ self._known_aps[apID][tag] = 1
+ self._known_aps[apID]['AT_visible'] = True
+
+ self.incrementChisto('Unique APs', channel)
+ self.incrementChisto('Current APs', channel)
+
+ logging.info("New AP%s: %s" % (contextlabel, apID))
+ else:
+ # seen before, merge info
+ for p in access_point:
+ self._known_aps[apID][p] = access_point[p]
+
+ # if wasn't visible, increment current count
+ if not self._known_aps[apID]['AT_visible']:
+ self._known_aps[apID]['AT_visible'] = True
+ self._known_aps[apID]['AT_seen'] += 1
+ self.incrementChisto('Current APs', channel)
+
+ # increment context count in the AP data
+ self._known_aps[apID][tag] = 1 if tag not in self._known_aps[apID] else self._known_aps[apID][tag] + 1
+ if not context:
+ logging.info("Returning AP: %s" % apID)
+
+ self._known_aps[apID]['AT_lastseen'] = time.time()
+ return True
+ except Exception as e:
+ logging.exception(e)
+ return False
+
+ # called when the agent is sending an association frame
+ def on_association(self, agent, access_point):
+ try:
+ self.incrementChisto('Associations', access_point['channel'])
+ self.markAPSeen(access_point, "assoc")
+
+ except Exception as e:
+ logging.exception(e)
+
+ # called when the agent is deauthenticating a client station from an AP
+ def on_deauthentication(self, agent, access_point, client_station):
+ try:
+ self.incrementChisto('Deauths', access_point['channel'])
+ self.markAPSeen(access_point, "deauth")
+
+ except Exception as e:
+ logging.exception(e)
+
+ # callend when the agent is tuning on a specific channel
+ def on_channel_hop(self, agent, channel):
+ pass
+
+ # called when a new handshake is captured, access_point and client_station are json objects
+ # if the agent could match the BSSIDs to the current list, otherwise they are just the strings of the BSSIDs
+ def on_handshake(self, agent, filename, access_point, client_station):
+ try:
+ self.incrementChisto('Handshakes', access_point['channel'])
+ self.markAPSeen(access_point, "handshake")
+ except Exception as e:
+ logging.exception(e)
+
+ def on_bcap_wifi_ap_new(self, agent, event):
+ try:
+ ap = event['data']
+ apname = self.normalize(ap['hostname'])
+ apmac = self.normalize(ap['mac'])
+ apID = apname + '-' + apmac
+ channel = ap['channel']
+
+ self.markAPSeen(ap)
+
+ except Exception as e:
+ logging.exception(repr(e))
+
+ def on_bcap_wifi_ap_lost(self, agent, event):
+ try:
+ ap = event['data']
+ apname = self.normalize(ap['hostname'])
+ apmac = self.normalize(ap['mac'])
+ apID = apname + '-' + apmac
+ channel = ap['channel']
+
+ if apID not in self._known_aps:
+ self.incrementChisto('Missed joins', channel)
+ logging.warn("Unknown AP '%s' seen leaving" % apID)
+ else:
+ if not self._known_aps[apID]['AT_visible']:
+ self.incrementChisto('Missed rejoins', channel)
+ logging.warn("AP '%s' already gone", apID)
+ else:
+ self._known_aps[apID]['AT_visible'] = False
+ self.incrementChisto('Current APs', channel, -1)
+
+ except Exception as e:
+ logging.exception(repr(e))
+
+ def on_bcap_wifi_client_new(self, agent, event):
+ try:
+ ap = event['data']['AP']
+ cl = event['data']['Client']
+ apmac = self.normalize(ap['mac'])
+ clmac = self.normalize(cl['mac'])
+ clID = clmac + '-' + apmac
+ channel = ap['channel']
+ except Exception as e:
+ logging.exception(repr(e))
+
+ def on_bcap_wifi_client_lost(self, agent, event):
+ try:
+ ap = event['data']['AP']
+ cl = event['data']['Client']
+ except Exception as e:
+ logging.exception(repr(e))
\ No newline at end of file
diff --git a/pwnagotchi/strategy/__init__.py b/pwnagotchi/strategy/__init__.py
new file mode 100644
index 00000000..1ce63ce6
--- /dev/null
+++ b/pwnagotchi/strategy/__init__.py
@@ -0,0 +1,71 @@
+"""Pwnagotchi channel selection strategy - core to agent behavior.
+
+This module replaces the removed RL-based AI with intelligent heuristics:
+- Track active channels (have APs) and prioritize them
+- Explore unscanned channels randomly
+- Maintain statistics per channel
+- Guide agent behavior through channel selection
+
+Originally extracted from auto-tune plugin to make it core functionality.
+"""
+
+import logging
+from .channels import ChannelStrategy
+from .statistics import ChannelStatistics
+
+
+class Strategy:
+ """
+ Main strategy facade.
+
+ Integrates channel selection with agent lifecycle and event handling.
+ """
+
+ def __init__(self, config, logger=None):
+ self.logger = logger or logging.getLogger(__name__)
+ self.channels = ChannelStrategy(config, logger=self.logger)
+ self.config = config
+
+ def start(self):
+ """Initialize strategy when agent starts."""
+ self.logger.info("Channel selection strategy initialized")
+
+ def select_next_channels(self, agent, access_points):
+ """
+ Select channels for next epoch.
+
+ Called after WiFi scan completes. Returns list of channels to scan.
+ """
+ return self.channels.select_channels(agent, access_points)
+
+ # Event handlers - connect to agent event system
+ def on_wifi_update(self, agent, access_points):
+ """WiFi list updated."""
+ self.channels.on_wifi_update(agent, access_points)
+
+ def on_association(self, agent, access_point):
+ """Association sent."""
+ self.channels.on_association(agent, access_point)
+
+ def on_deauthentication(self, agent, access_point, client_station):
+ """Deauthentication sent."""
+ self.channels.on_deauthentication(agent, access_point, client_station)
+
+ def on_handshake(self, agent, filename, access_point, client_station):
+ """Handshake captured."""
+ self.channels.on_handshake(agent, filename, access_point, client_station)
+
+ def on_bcap_wifi_ap_new(self, agent, event):
+ """Bettercap: new AP detected."""
+ self.channels.on_bcap_wifi_ap_new(agent, event)
+
+ def on_bcap_wifi_ap_lost(self, agent, event):
+ """Bettercap: AP lost."""
+ self.channels.on_bcap_wifi_ap_lost(agent, event)
+
+ def get_stats(self):
+ """Get current strategy statistics."""
+ return self.channels.get_stats()
+
+
+__all__ = ["Strategy", "ChannelStrategy", "ChannelStatistics"]
diff --git a/pwnagotchi/strategy/channels.py b/pwnagotchi/strategy/channels.py
new file mode 100644
index 00000000..2a30fce1
--- /dev/null
+++ b/pwnagotchi/strategy/channels.py
@@ -0,0 +1,137 @@
+"""Channel selection strategy - core to agent behavior."""
+
+import random
+import logging
+import pwnagotchi.utils
+from .statistics import ChannelStatistics
+
+
+class ChannelStrategy:
+ """
+ Intelligent channel selection strategy.
+
+ Strategy:
+ - Always scan channels with active APs (high probability of captures)
+ - Add random unscanned channels each epoch (explore new areas)
+ - Track statistics per channel to guide future decisions
+ """
+
+ def __init__(self, config, logger=None):
+ """
+ Initialize channel strategy.
+
+ Args:
+ config: pwnagotchi configuration dict
+ logger: optional logger instance
+ """
+ self.logger = logger or logging.getLogger(__name__)
+ self.config = config
+ self.stats = ChannelStatistics()
+
+ # Configuration options
+ self.extra_channels = config.get("main", {}).get("extra_channels", 15)
+ self.restrict_channels = config.get("main", {}).get("restrict_channels", None)
+ self.reset_history = config.get("main", {}).get("reset_history", True)
+
+ def select_channels(self, agent, access_points):
+ """
+ Select next set of channels to scan based on current APs and unscanned channels.
+
+ Returns list of channels to scan in next epoch.
+ """
+ try:
+ # Update active channels from current scan
+ self.stats.update_active_channels(access_points)
+
+ # Build next channel list: active + extra unscanned
+ next_channels = self.stats.active_channels.copy()
+
+ # Add random unscanned channels for exploration
+ n_extra = self.extra_channels
+ if len(self.stats.unscanned_channels) == 0:
+ self._repopulate_unscanned_channels(agent)
+
+ for _ in range(n_extra):
+ if len(self.stats.unscanned_channels):
+ ch = random.choice(list(self.stats.unscanned_channels))
+ self.stats.unscanned_channels.remove(ch)
+ next_channels.append(ch)
+
+ # Update agent config
+ if hasattr(agent, "_config"):
+ agent._config["personality"]["channels"] = next_channels
+
+ self.logger.info(
+ f"Active: {self.stats.active_channels}, "
+ f"Next: {next_channels}, "
+ f"Unscanned: {len(self.stats.unscanned_channels)}"
+ )
+
+ return next_channels
+
+ except Exception as e:
+ self.logger.error(f"Error selecting channels: {e}")
+ return self.stats.active_channels
+
+ def _repopulate_unscanned_channels(self, agent):
+ """Repopulate unscanned channel list from config or agent."""
+ try:
+ # Try restrict_channels first
+ if self.restrict_channels:
+ self.logger.info("Repopulating from restrict_channels")
+ self.stats.unscanned_channels = self.restrict_channels.copy()
+ # Try agent's allowed channels
+ elif hasattr(agent, "_allowed_channels"):
+ self.logger.info(f"Repopulating from allowed: {agent._allowed_channels}")
+ self.stats.unscanned_channels = agent._allowed_channels.copy()
+ # Try agent's supported channels
+ elif hasattr(agent, "_supported_channels"):
+ self.logger.info("Repopulating from supported")
+ self.stats.unscanned_channels = agent._supported_channels.copy()
+ # Fall back to all channels for interface
+ else:
+ self.logger.info("Repopulating from interface channels")
+ iface = self.config.get("main", {}).get("iface", "wlan0")
+ self.stats.unscanned_channels = pwnagotchi.utils.iface_channels(iface)
+
+ except Exception as e:
+ self.logger.warning(f"Error repopulating unscanned channels: {e}")
+
+ def on_wifi_update(self, agent, access_points):
+ """Called when agent updates its AP list."""
+ self.stats.update_active_channels(access_points)
+
+ def on_association(self, agent, access_point):
+ """Called when sending association frame."""
+ self.stats.record_interaction("Associations", access_point.get("channel", -1))
+ self.stats.mark_ap_seen(access_point, "assoc")
+
+ def on_deauthentication(self, agent, access_point, client_station):
+ """Called when sending deauth."""
+ self.stats.record_interaction("Deauths", access_point.get("channel", -1))
+ self.stats.mark_ap_seen(access_point, "deauth")
+
+ def on_handshake(self, agent, filename, access_point, client_station):
+ """Called when handshake is captured."""
+ self.stats.record_interaction("Handshakes", access_point.get("channel", -1))
+ self.stats.mark_ap_seen(access_point, "handshake")
+
+ def on_bcap_wifi_ap_new(self, agent, event):
+ """Called when bettercap detects new AP."""
+ try:
+ ap = event.get("data", {})
+ self.stats.mark_ap_seen(ap)
+ except Exception as e:
+ self.logger.debug(f"Error on_bcap_wifi_ap_new: {e}")
+
+ def on_bcap_wifi_ap_lost(self, agent, event):
+ """Called when bettercap loses AP."""
+ try:
+ ap = event.get("data", {})
+ self.stats.record_ap_lost(ap)
+ except Exception as e:
+ self.logger.debug(f"Error on_bcap_wifi_ap_lost: {e}")
+
+ def get_stats(self):
+ """Get current strategy statistics."""
+ return self.stats.get_stats()
diff --git a/pwnagotchi/strategy/statistics.py b/pwnagotchi/strategy/statistics.py
new file mode 100644
index 00000000..f72d5ff9
--- /dev/null
+++ b/pwnagotchi/strategy/statistics.py
@@ -0,0 +1,120 @@
+"""Channel and interaction statistics tracking."""
+
+import time
+import logging
+
+
+class ChannelStatistics:
+ """Tracks AP counts, interactions, and handshakes per channel."""
+
+ def __init__(self):
+ self.histogram = {"loops": 0} # AP count per channel per epoch
+ self.chistos = {"_all_actions": {-1: 0}} # Interaction stats per channel
+ self.known_aps = {} # All APs seen (key: normalized_name-mac)
+ self.active_channels = [] # Channels with APs in current scan
+ self.unscanned_channels = [] # Channels to explore next
+
+ def mark_ap_seen(self, access_point, context=None):
+ """Track an AP sighting and update stats."""
+ try:
+ apname = self._normalize(access_point.get("hostname", ""))
+ apmac = self._normalize(access_point.get("mac", ""))
+ apid = f"{apname}-{apmac}"
+ channel = access_point.get("channel", -1)
+
+ tag = f"AT_{context}" if context else "AT_seen"
+
+ if apid not in self.known_aps:
+ # First time seeing this AP
+ self.known_aps[apid] = access_point.copy()
+ self.known_aps[apid]["AT_seen"] = 1
+ self.known_aps[apid][tag] = 1
+ self.known_aps[apid]["AT_visible"] = True
+
+ self.increment_chisto("Unique APs", channel)
+ self.increment_chisto("Current APs", channel)
+ logging.info(f"New AP: {apid} on channel {channel}")
+ else:
+ # Update existing AP
+ for key in access_point:
+ self.known_aps[apid][key] = access_point[key]
+
+ if not self.known_aps[apid]["AT_visible"]:
+ self.known_aps[apid]["AT_visible"] = True
+ self.known_aps[apid]["AT_seen"] += 1
+ self.increment_chisto("Current APs", channel)
+
+ tag_count = self.known_aps[apid].get(tag, 0)
+ self.known_aps[apid][tag] = tag_count + 1
+
+ self.known_aps[apid]["AT_lastseen"] = time.time()
+ return True
+
+ except Exception as e:
+ logging.debug(f"Error marking AP seen: {e}")
+ return False
+
+ def increment_chisto(self, stat_name, channel, count=1):
+ """Increment a statistic for a channel."""
+ if stat_name not in self.chistos:
+ self.chistos[stat_name] = {}
+ if channel not in self.chistos[stat_name]:
+ self.chistos[stat_name][channel] = 0
+ self.chistos[stat_name][channel] += count
+
+ def update_active_channels(self, access_points):
+ """Update list of channels with active APs."""
+ active_channels = []
+ self.histogram["loops"] = self.histogram.get("loops", 0) + 1
+
+ for ap in access_points:
+ self.mark_ap_seen(ap, "wifi_update")
+ channel = ap.get("channel")
+ if channel and channel not in active_channels:
+ active_channels.append(channel)
+ if channel in self.unscanned_channels:
+ self.unscanned_channels.remove(channel)
+
+ self.histogram[channel] = self.histogram.get(channel, 0) + 1
+
+ self.active_channels = active_channels
+ logging.debug(f"Active channels: {active_channels}, Histogram: {self.histogram}")
+
+ def record_interaction(self, interaction_type, channel):
+ """Record an interaction (association, deauth, handshake) on a channel."""
+ self.increment_chisto(interaction_type, channel)
+
+ def record_ap_lost(self, access_point):
+ """Record an AP going offline."""
+ try:
+ apname = self._normalize(access_point.get("hostname", ""))
+ apmac = self._normalize(access_point.get("mac", ""))
+ apid = f"{apname}-{apmac}"
+ channel = access_point.get("channel", -1)
+
+ if apid in self.known_aps:
+ if self.known_aps[apid]["AT_visible"]:
+ self.known_aps[apid]["AT_visible"] = False
+ self.increment_chisto("Current APs", channel, -1)
+ else:
+ self.increment_chisto("Missed joins", channel)
+
+ except Exception as e:
+ logging.debug(f"Error recording AP lost: {e}")
+
+ @staticmethod
+ def _normalize(text):
+ """Normalize text for comparison."""
+ if not text:
+ return ""
+ return text.lower().replace(" ", "_")
+
+ def get_stats(self):
+ """Get current statistics snapshot."""
+ return {
+ "active_channels": self.active_channels.copy(),
+ "unscanned_count": len(self.unscanned_channels),
+ "known_aps": len(self.known_aps),
+ "histogram": self.histogram.copy(),
+ "chistos": self.chistos.copy(),
+ }
diff --git a/pwnagotchi/ui/web/handler.py b/pwnagotchi/ui/web/handler.py
index 06894ecf..890a5d93 100644
--- a/pwnagotchi/ui/web/handler.py
+++ b/pwnagotchi/ui/web/handler.py
@@ -217,9 +217,7 @@ class Handler:
if name is None:
# Determine which plugins are from the default folder
default_plugins = set()
- default_path = os.path.join(
- os.path.dirname(os.path.realpath(plugins.__file__)), "default"
- )
+ default_path = os.path.join(os.path.dirname(os.path.realpath(plugins.__file__)), "default")
for plugin_name, plugin_path in plugins.database.items():
if plugin_path.startswith(default_path):
default_plugins.add(plugin_name)