integrate(strategy): wire channel strategy into agent lifecycle

Connected pwnagotchi/strategy module to agent.py so channel selection is always active:

Changes:
1. Import Strategy in agent.py
2. Initialize strategy in Agent.__init__()
3. Wire strategy event handlers to key agent methods:
   - on_wifi_update(): called when WiFi list refreshed
   - select_next_channels(): called to set channels for next epoch
   - on_association(): track assoc interactions per channel
   - on_deauthentication(): track deauth interactions per channel
   - on_handshake(): track handshake captures per channel

Integration points:
- set_access_points(): updates strategy and selects next channels
- associate(): reports association to strategy
- deauth(): reports deauth to strategy
- _on_event_new_handshake(): reports handshake to strategy

The strategy now:
- Receives every relevant agent event
- Maintains per-channel statistics
- Selects next channels to scan based on active APs + random exploration
- Updates config['personality']['channels'] for next epoch
- Guides agent behavior through channel selection

This makes channel selection strategy a core, always-active component instead of
an optional plugin, ensuring intelligent adaptation to environment changes.

Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>
This commit is contained in:
Jeroen Oudshoorn
2026-07-27 17:30:20 +02:00
co-authored by Claude Haiku 4.5
parent b9e8d1484f
commit 22d35b5eb6
+11
View File
@@ -16,6 +16,7 @@ from pwnagotchi.automata import Automata
from pwnagotchi.log import LastSession
from pwnagotchi.bettercap import Client
from pwnagotchi.mesh.utils import AsyncAdvertiser
from pwnagotchi.strategy import Strategy
RECOVERY_DATA_FILE = '/root/.pwnagotchi-recovery'
@@ -40,6 +41,9 @@ class Agent(Client, Automata, AsyncAdvertiser):
self._view.set_agent(self)
self._web_ui = Server(self, config['ui'])
# Initialize channel selection strategy (core infrastructure, not optional)
self._strategy = Strategy(config)
self._access_points = []
self._last_pwnd = None
self._history = {}
@@ -166,6 +170,10 @@ class Agent(Client, Automata, AsyncAdvertiser):
def set_access_points(self, aps):
self._access_points = aps
plugins.on('wifi_update', self, aps)
# Update strategy with latest access points
self._strategy.on_wifi_update(self, aps)
# Select next channels to scan based on strategy
next_channels = self._strategy.select_next_channels(self, aps)
self._epoch.observe(aps, list(self._peers.values()))
return self._access_points
@@ -369,6 +377,7 @@ class Agent(Client, Automata, AsyncAdvertiser):
"!!! captured new handshake on channel %d, %d dBm: %s (%s) -> %s [%s (%s)] !!!",
ap['channel'], ap['rssi'], sta['mac'], sta['vendor'], ap['hostname'], ap['mac'], ap['vendor'])
plugins.on('handshake', self, filename, ap, sta)
self._strategy.on_handshake(self, filename, ap, sta)
found_handshake = True
self._update_handshakes(1 if found_handshake else 0)
@@ -441,6 +450,7 @@ class Agent(Client, Automata, AsyncAdvertiser):
self._on_error(ap['mac'], e)
plugins.on('association', self, ap)
self._strategy.on_association(self, ap)
if throttle > 0:
time.sleep(throttle)
self._view.on_normal()
@@ -466,6 +476,7 @@ class Agent(Client, Automata, AsyncAdvertiser):
self._on_error(sta['mac'], e)
plugins.on('deauthentication', self, ap, sta)
self._strategy.on_deauthentication(self, ap, sta)
if throttle > 0:
time.sleep(throttle)
self._view.on_normal()