From cfbbcd8e03ddbf586e22cd9db8cc6f6938be085e Mon Sep 17 00:00:00 2001 From: AlienMajik <118037572+AlienMajik@users.noreply.github.com> Date: Sun, 16 Mar 2025 16:52:34 -0700 Subject: [PATCH] Update age.py New Enhancements in v2.0.3: Documented Training Logic: In on_epoch, a comment explains why train_epochs increments every 10 epochs: # Increment train_epochs every 10 epochs to simulate slower training progress. Enhanced File I/O Safety: In on_handshake, handshake logging is wrapped in a try-except block to handle file writing errors gracefully. Refined Decay Mechanics: In apply_decay, decay calculation uses floating-point division for smoother, more proportional point reduction. Increased Logging: Added debug and info logs for better transparency: on_epoch: Logs epoch number and points (logging.debug). check_achievements: Logs new titles (logging.info). apply_decay: Logs points lost due to decay (logging.info). on_handshake: Logs captured handshake details (logging.info). Thread Safety: Imported threading and added a data_lock in __init__. Used in save_data to ensure thread-safe file writing. Accurate Achievement Notifications: Tracks previous titles and stars to ensure new achievements are detected and announced correctly. Robust Handshake Handling: Adds type checking and error logging to prevent crashes from unexpected data, making the plugin more stable. Seamless New Installations: Fully initializes all attributes when starting fresh, improving reliability for new users. Persistent Progress: Saves achievement states to the data file, maintaining continuity across sessions. --- age.py | 161 ++++++++++++++++++++++++++++++++++++++------------------- 1 file changed, 109 insertions(+), 52 deletions(-) diff --git a/age.py b/age.py index c72904a..354d549 100644 --- a/age.py +++ b/age.py @@ -3,6 +3,7 @@ import json import logging import time import random +import threading import pwnagotchi import pwnagotchi.plugins as plugins @@ -13,7 +14,7 @@ from pwnagotchi.ui.view import BLACK class Age(plugins.Plugin): __author__ = 'AlienMajik' - __version__ = '2.0.2' + __version__ = '2.0.3' __license__ = 'MIT' __description__ = ('Enhanced plugin with achievement tiers, configurable titles, decay mechanics, ' 'progress tracking, and dynamic status messages.') @@ -21,7 +22,7 @@ class Age(plugins.Plugin): DEFAULT_AGE_TITLES = { 1000: "Neon Spawn", 2000: "Script Kiddie", - 5000: "WiFi Outlaw", + 5000: "WiFi Outlaw", 10000: "Data Raider", 25000: "Prophet", 33333: "Off the Grid" @@ -36,7 +37,7 @@ class Age(plugins.Plugin): } def __init__(self): - # Default positions (x, y) + # Default UI positions (x, y) self.default_positions = { 'age': (10, 40), 'strength': (80, 40), @@ -44,6 +45,7 @@ class Age(plugins.Plugin): 'stars': (10, 80), } + # Initialize core metrics self.epochs = 0 self.train_epochs = 0 self.network_points = 0 @@ -60,20 +62,44 @@ class Age(plugins.Plugin): self.decay_amount = 10 self.age_titles = self.DEFAULT_AGE_TITLES self.strength_titles = self.DEFAULT_STRENGTH_TITLES + + # Achievement tracking attributes + self.prev_age_title = "Unborn" + self.prev_strength_title = "Untrained" + self.prev_stars = 0 + + # Additional configurations + self.points_map = { + 'wpa3': 10, + 'wpa2': 5, + 'wep': 2, + 'wpa': 2 + } + self.motivational_quotes = [ + "Keep going, you're crushing it!", + "You're a WiFi wizard in the making!", + "More handshakes, more power!", + "Don't stop now, you're almost there!", + "Keep evolving, don't let decay catch you!" + ] + self.data_lock = threading.Lock() def on_loaded(self): - # Load configuration with fallbacks + # Load configuration options with fallbacks self.max_stars = self.options.get('max_stars', 5) self.star_interval = self.options.get('star_interval', 1000) self.decay_interval = self.options.get('decay_interval', 50) self.decay_amount = self.options.get('decay_amount', 10) self.age_titles = self.options.get('age_titles', self.DEFAULT_AGE_TITLES) self.strength_titles = self.options.get('strength_titles', self.DEFAULT_STRENGTH_TITLES) + self.points_map = self.options.get('points_map', self.points_map) + self.motivational_quotes = self.options.get('motivational_quotes', self.motivational_quotes) self.load_data() self.initialize_handshakes() def initialize_handshakes(self): + """Initialize handshake count based on existing .pcap files.""" if self.handshake_count == 0 and os.path.isdir(self.handshake_dir): existing = [f for f in os.listdir(self.handshake_dir) if f.endswith('.pcap')] if existing: @@ -82,6 +108,7 @@ class Age(plugins.Plugin): self.save_data() def get_age_title(self): + """Determine age title based on epochs.""" thresholds = sorted(self.age_titles.keys(), reverse=True) for t in thresholds: if self.epochs >= t: @@ -89,6 +116,7 @@ class Age(plugins.Plugin): return "Unborn" def get_strength_title(self): + """Determine strength title based on train_epochs.""" thresholds = sorted(self.strength_titles.keys(), reverse=True) for t in thresholds: if self.train_epochs >= t: @@ -96,16 +124,11 @@ class Age(plugins.Plugin): return "Untrained" def random_motivational_quote(self): - quotes = [ - "Keep going, you're crushing it!", - "You're a WiFi wizard in the making!", - "More handshakes, more power!", - "Don't stop now, you're almost there!", - "Keep evolving, don't let decay catch you!" - ] - return random.choice(quotes) + """Return a random motivational quote from the configurable list.""" + return random.choice(self.motivational_quotes) def random_inactivity_message(self, points_lost): + """Return a random inactivity message with points lost.""" messages = [ "Time to wake up, you're rusting!", "Decayed by {points_lost}, keep it active!", @@ -116,33 +139,39 @@ class Age(plugins.Plugin): return random.choice(messages).format(points_lost=points_lost) def check_achievements(self, agent): + """Check and announce new age or strength achievements.""" current_age = self.get_age_title() current_strength = self.get_strength_title() if current_age != self.prev_age_title: agent.view().set('face', faces.HAPPY) agent.view().set('status', f"🎉 {current_age} Achieved! {self.random_motivational_quote()}") + logging.info(f"[Age] New age title: {current_age}") self.prev_age_title = current_age if current_strength != self.prev_strength_title: agent.view().set('face', faces.MOTIVATED) agent.view().set('status', f"💪 Evolved to {current_strength}!") + logging.info(f"[Age] New strength title: {current_strength}") self.prev_strength_title = current_strength def apply_decay(self, agent): + """Apply decay to network points based on inactivity with refined mechanics.""" inactive_epochs = self.epochs - self.last_active_epoch if inactive_epochs >= self.decay_interval: - decay_cycles = inactive_epochs // self.decay_interval - points_lost = decay_cycles * self.decay_amount + decay_factor = inactive_epochs / self.decay_interval + points_lost = int(decay_factor * self.decay_amount) self.network_points = max(0, self.network_points - points_lost) if points_lost > 0: agent.view().set('face', faces.SAD) agent.view().set('status', self.random_inactivity_message(points_lost)) + logging.info(f"[Age] Applied decay: lost {points_lost} points due to inactivity") self.last_active_epoch = self.epochs self.save_data() def on_ui_setup(self, ui): + """Set up UI elements with configurable positions.""" def get_position(element): x = self.options.get( f"{element}_x", @@ -184,14 +213,18 @@ class Age(plugins.Plugin): position=positions['stars'], label_font=fonts.Bold, text_font=fonts.Medium)) def on_ui_update(self, ui): + """Update UI elements with current values.""" ui.set('Age', self.get_age_title()) ui.set('Strength', self.get_strength_title()) ui.set('Points', self.abrev_number(self.network_points)) ui.set('ReP', self.get_star_string()) def on_epoch(self, agent, epoch, epoch_data): + """Handle epoch events: increment counters, apply decay, and check achievements.""" self.epochs += 1 + # Increment train_epochs every 10 epochs to simulate slower training progress self.train_epochs += 1 if self.epochs % 10 == 0 else 0 + logging.debug(f"[Age] Epoch {self.epochs}, Points: {self.network_points}") self.apply_decay(agent) self.check_achievements(agent) @@ -202,33 +235,49 @@ class Age(plugins.Plugin): self.save_data() def age_checkpoint(self, agent): - # Status update at every epoch milestone (e.g., every 100 epochs) + """Display milestone message every 100 epochs.""" view = agent.view() view.set('face', faces.HAPPY) view.set('status', f"Epoch milestone: {self.epochs} epochs!") view.update(force=True) def on_handshake(self, agent, *args): - self.last_active_epoch = self.epochs - enc = args[2].get('encryption', '').lower() - - points = { - 'wpa3': 10, 'wpa2': 5, - 'wep': 2, 'wpa': 2 - }.get(enc, 1) - - self.network_points += points - self.handshake_count += 1 - - # Log details - with open(self.log_path, 'a') as f: - essid = args[2].get('essid', 'unknown') - f.write(f"{time.time()},{essid},{enc},{points}\n") - - self.new_star_checkpoint(agent) - self.save_data() + """Handle handshake events with enhanced error handling and logging.""" + try: + if len(args) < 3: + logging.warning("[Age] Insufficient arguments in on_handshake") + return + + ap = args[2] + if isinstance(ap, dict): + enc = ap.get('encryption', '').lower() + essid = ap.get('essid', 'unknown') + else: + logging.warning(f"[Age] AP is a string, not a dictionary: {ap}. Skipping handshake processing.") + return + + points = self.points_map.get(enc, 1) + + self.network_points += points + self.handshake_count += 1 + self.last_active_epoch = self.epochs + + # Log handshake details with enhanced file I/O safety + try: + with open(self.log_path, 'a') as f: + f.write(f"{time.time()},{essid},{enc},{points}\n") + except Exception as e: + logging.error(f"[Age] Failed to log handshake: {str(e)}") + + logging.info(f"[Age] Captured handshake: {essid}, encryption: {enc}, points gained: {points}") + + self.new_star_checkpoint(agent) + self.save_data() + except Exception as e: + logging.error(f"[Age] Error in handshake processing: {str(e)}") def new_star_checkpoint(self, agent): + """Check and announce new star tier achievements.""" stars = self.get_stars_count() if stars > self.prev_stars: symbol = self.get_symbol_for_handshakes() @@ -237,32 +286,34 @@ class Age(plugins.Plugin): self.prev_stars = stars def load_data(self): + """Load saved data from JSON file with defaults for new installations.""" try: if os.path.exists(self.data_path): with open(self.data_path, 'r') as f: data = json.load(f) - - # Handle old format compatibility - self.epochs = data.get('epochs', data.get('epochs_lived', 0)) - self.train_epochs = data.get('train_epochs', data.get('epochs_trained', 0)) - self.network_points = data.get('points', data.get('network_points', 0)) - self.handshake_count = data.get('handshakes', data.get('handshake_count', 0)) - - # New fields with defaults + self.epochs = data.get('epochs', 0) + self.train_epochs = data.get('train_epochs', 0) + self.network_points = data.get('points', 0) + self.handshake_count = data.get('handshakes', 0) self.last_active_epoch = data.get('last_active', 0) self.prev_age_title = data.get('prev_age', self.get_age_title()) self.prev_strength_title = data.get('prev_strength', self.get_strength_title()) self.prev_stars = data.get('prev_stars', self.get_stars_count()) - - # Migrate old format to new format - if 'epochs_lived' in data: - self.save_data() # Resave in new format - logging.info("[Age] Migrated old data format to new format") - + else: + # Set defaults for a new installation + self.epochs = 0 + self.train_epochs = 0 + self.network_points = 0 + self.handshake_count = 0 + self.last_active_epoch = 0 + self.prev_age_title = self.get_age_title() + self.prev_strength_title = self.get_strength_title() + self.prev_stars = self.get_stars_count() except Exception as e: logging.error(f"[Age] Load error: {str(e)}") def save_data(self): + """Save current data to JSON file with thread safety.""" data = { 'epochs': self.epochs, 'train_epochs': self.train_epochs, @@ -273,22 +324,27 @@ class Age(plugins.Plugin): 'prev_strength': self.get_strength_title(), 'prev_stars': self.get_stars_count(), } - try: - with open(self.data_path, 'w') as f: - json.dump(data, f, indent=2) - except Exception as e: - logging.error(f"[Age] Save error: {str(e)}") + with self.data_lock: + try: + with open(self.data_path, 'w') as f: + json.dump(data, f, indent=2) + except Exception as e: + logging.error(f"[Age] Save error: {str(e)}") def get_stars_count(self): + """Calculate current number of stars.""" return min(self.handshake_count // self.star_interval, self.max_stars) def get_symbol_for_handshakes(self): + """Return symbol based on handshake count.""" return '♣' if self.handshake_count >= 10000 else '♦' if self.handshake_count >= 5000 else '★' def get_star_string(self): + """Return string of star symbols.""" return self.get_symbol_for_handshakes() * self.get_stars_count() def abrev_number(self, num): + """Abbreviate large numbers (e.g., 1000 -> 1K).""" for unit in ['','K','M','B']: if abs(num) < 1000: return f"{num:.1f}{unit}".rstrip('.0') @@ -298,3 +354,4 @@ class Age(plugins.Plugin): +