271 lines
9.1 KiB
Python
Executable File
271 lines
9.1 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
Battery Monitor for Stormux Gaming Image
|
|
|
|
Monitors battery levels and provides warnings at 10% and 5%,
|
|
with automatic shutdown at 3% to prevent data loss.
|
|
|
|
Only activates if a real battery is detected to avoid false alarms.
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import time
|
|
import subprocess
|
|
import logging
|
|
import wave
|
|
import numpy as np
|
|
from pathlib import Path
|
|
|
|
try:
|
|
import speechd
|
|
import simpleaudio as sa
|
|
except ImportError as e:
|
|
print(f"Required module missing: {e}")
|
|
sys.exit(1)
|
|
|
|
# Set up logging - create log directory if it doesn't exist
|
|
try:
|
|
log_dir = Path.home() / '.config' / 'stormux'
|
|
log_dir.mkdir(parents=True, exist_ok=True)
|
|
log_file = log_dir / 'battery_monitor.log'
|
|
except:
|
|
# Fallback to /tmp if home directory not accessible
|
|
log_file = '/tmp/battery_monitor.log'
|
|
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format='%(asctime)s - %(levelname)s - %(message)s',
|
|
handlers=[
|
|
logging.FileHandler(log_file),
|
|
logging.StreamHandler()
|
|
]
|
|
)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
class BatteryMonitor:
|
|
def __init__(self):
|
|
# Settings
|
|
self.enabled = True
|
|
self.warning_10_percent = True
|
|
self.warning_5_percent = True
|
|
self.shutdown_3_percent = True
|
|
self.check_interval = 30
|
|
self.speech_enabled = True
|
|
|
|
# Warning state tracking
|
|
self.warned_10 = False
|
|
self.warned_5 = False
|
|
|
|
# Initialize speech client
|
|
self.speech_client = None
|
|
self.init_speech()
|
|
|
|
# Ensure config directory exists (handled in logging setup above)
|
|
|
|
def init_speech(self):
|
|
"""Initialize speech-dispatcher client"""
|
|
try:
|
|
self.speech_client = speechd.SSIPClient("battery_monitor")
|
|
self.speech_client.set_priority(speechd.Priority.IMPORTANT)
|
|
logger.info("Speech client initialized")
|
|
except Exception as e:
|
|
logger.error(f"Failed to initialize speech: {e}")
|
|
self.speech_enabled = False
|
|
|
|
def has_battery(self):
|
|
"""Check if system has a real battery"""
|
|
power_supply_dir = Path('/sys/class/power_supply')
|
|
if not power_supply_dir.exists():
|
|
return False
|
|
|
|
for item in power_supply_dir.iterdir():
|
|
type_file = item / 'type'
|
|
if type_file.exists():
|
|
try:
|
|
if type_file.read_text().strip() == 'Battery':
|
|
logger.info(f"Battery detected: {item.name}")
|
|
return True
|
|
except Exception:
|
|
continue
|
|
|
|
logger.info("No battery detected")
|
|
return False
|
|
|
|
def get_battery_level(self):
|
|
"""Get current battery percentage"""
|
|
power_supply_dir = Path('/sys/class/power_supply')
|
|
|
|
for item in power_supply_dir.iterdir():
|
|
type_file = item / 'type'
|
|
capacity_file = item / 'capacity'
|
|
|
|
if (type_file.exists() and capacity_file.exists()):
|
|
try:
|
|
if type_file.read_text().strip() == 'Battery':
|
|
capacity = int(capacity_file.read_text().strip())
|
|
return capacity
|
|
except Exception:
|
|
continue
|
|
|
|
return None
|
|
|
|
def is_on_ac_power(self):
|
|
"""Check if system is plugged into AC power"""
|
|
power_supply_dir = Path('/sys/class/power_supply')
|
|
|
|
for item in power_supply_dir.iterdir():
|
|
type_file = item / 'type'
|
|
online_file = item / 'online'
|
|
|
|
if (type_file.exists() and online_file.exists()):
|
|
try:
|
|
device_type = type_file.read_text().strip()
|
|
if device_type in ['ADP', 'Mains', 'AC']:
|
|
online = int(online_file.read_text().strip())
|
|
return online == 1
|
|
except Exception:
|
|
continue
|
|
|
|
return False
|
|
|
|
def speak(self, message):
|
|
"""Speak message using speech-dispatcher"""
|
|
if not self.speech_enabled or not self.speech_client:
|
|
logger.warning("Speech not available")
|
|
return
|
|
|
|
try:
|
|
self.speech_client.cancel()
|
|
self.speech_client.speak(message)
|
|
logger.info(f"Speaking: {message}")
|
|
except Exception as e:
|
|
logger.error(f"Speech error: {e}")
|
|
|
|
def generate_urgent_sound(self):
|
|
"""Generate urgent beep sound in memory"""
|
|
try:
|
|
# Generate urgent beeping sound
|
|
sample_rate = 44100
|
|
duration = 0.5
|
|
frequency = 800
|
|
|
|
# Create beep pattern: 3 short beeps
|
|
beeps = []
|
|
for _ in range(3):
|
|
t = np.linspace(0, duration, int(sample_rate * duration))
|
|
wave_data = np.sin(2 * np.pi * frequency * t)
|
|
# Add fade in/out
|
|
fade_samples = int(0.05 * sample_rate)
|
|
wave_data[:fade_samples] *= np.linspace(0, 1, fade_samples)
|
|
wave_data[-fade_samples:] *= np.linspace(1, 0, fade_samples)
|
|
beeps.extend(wave_data)
|
|
# Add silence between beeps
|
|
beeps.extend([0] * int(0.2 * sample_rate))
|
|
|
|
# Convert to 16-bit integers
|
|
audio_data = np.array(beeps) * 32767
|
|
audio_data = audio_data.astype(np.int16)
|
|
|
|
return audio_data, sample_rate
|
|
|
|
except Exception as e:
|
|
logger.error(f"Sound generation error: {e}")
|
|
return None, None
|
|
|
|
def play_urgent_sound(self):
|
|
"""Play urgent warning sound"""
|
|
try:
|
|
audio_data, sample_rate = self.generate_urgent_sound()
|
|
if audio_data is not None:
|
|
play_obj = sa.play_buffer(audio_data, 1, 2, sample_rate)
|
|
play_obj.wait_done()
|
|
logger.info("Urgent sound played")
|
|
except Exception as e:
|
|
logger.error(f"Sound playback error: {e}")
|
|
|
|
def handle_low_battery(self, level):
|
|
"""Handle low battery warnings and actions"""
|
|
# Don't give warnings if we're plugged into AC power
|
|
if self.is_on_ac_power():
|
|
return
|
|
|
|
if level <= 3:
|
|
if self.shutdown_3_percent:
|
|
logger.critical(f"Battery at {level}% - initiating shutdown")
|
|
self.speak("Critical battery level. System shutting down now.")
|
|
time.sleep(3) # Give speech time to complete
|
|
subprocess.run(['sudo', 'systemctl', 'poweroff'], check=True)
|
|
return
|
|
|
|
elif level <= 5 and not self.warned_5:
|
|
if self.warning_5_percent:
|
|
logger.warning(f"Battery at {level}% - urgent warning")
|
|
self.play_urgent_sound()
|
|
self.speak("Extremely low battery. Computer will shut down soon.")
|
|
self.warned_5 = True
|
|
|
|
elif level <= 10 and not self.warned_10:
|
|
if self.warning_10_percent:
|
|
logger.warning(f"Battery at {level}% - first warning")
|
|
self.speak("Low battery warning. Please connect power adapter.")
|
|
self.warned_10 = True
|
|
|
|
def reset_warnings_if_charging(self):
|
|
"""Reset warning flags if system is charging"""
|
|
if self.is_on_ac_power():
|
|
if self.warned_10 or self.warned_5:
|
|
logger.info("AC power connected - resetting warning flags")
|
|
self.warned_10 = False
|
|
self.warned_5 = False
|
|
|
|
def monitor(self):
|
|
"""Main monitoring loop"""
|
|
if not self.enabled:
|
|
logger.info("Battery monitoring disabled")
|
|
return
|
|
|
|
if not self.has_battery():
|
|
logger.info("No battery detected - monitoring disabled")
|
|
return
|
|
|
|
logger.info("Starting battery monitoring")
|
|
|
|
while True:
|
|
try:
|
|
level = self.get_battery_level()
|
|
if level is not None:
|
|
logger.debug(f"Battery level: {level}%")
|
|
self.reset_warnings_if_charging()
|
|
self.handle_low_battery(level)
|
|
else:
|
|
logger.warning("Could not read battery level")
|
|
|
|
time.sleep(self.check_interval)
|
|
|
|
except KeyboardInterrupt:
|
|
logger.info("Battery monitoring stopped by user")
|
|
break
|
|
except Exception as e:
|
|
logger.error(f"Monitoring error: {e}")
|
|
time.sleep(self.check_interval)
|
|
|
|
def main():
|
|
try:
|
|
monitor = BatteryMonitor()
|
|
monitor.monitor()
|
|
except KeyboardInterrupt:
|
|
logger.info("Battery monitor stopped")
|
|
except Exception as e:
|
|
logger.error(f"Fatal error: {e}")
|
|
sys.exit(1)
|
|
finally:
|
|
# Clean up speech client
|
|
try:
|
|
if 'monitor' in locals() and monitor.speech_client:
|
|
monitor.speech_client.close()
|
|
except:
|
|
pass
|
|
|
|
if __name__ == '__main__':
|
|
main() |