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.
This commit is contained in:
AlienMajik
2025-03-16 16:52:34 -07:00
committed by GitHub
parent c070f87bcf
commit cfbbcd8e03

161
age.py
View File

@ -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):