107 lines
3.5 KiB
Python
107 lines
3.5 KiB
Python
import time
|
|
import config
|
|
import logging
|
|
import smtplib
|
|
from nut2 import PyNUTClient
|
|
from email.mime.text import MIMEText
|
|
from logging.handlers import TimedRotatingFileHandler
|
|
|
|
ups_id = config.UPS_ID
|
|
gmail_user = config.GMAIL_USER
|
|
gmail_password = config.GMAIL_PASS
|
|
recipient_email = config.GMAIL_ADDR
|
|
|
|
def setup_logging():
|
|
logger = logging.getLogger('powerlog')
|
|
logger.setLevel(logging.INFO)
|
|
handler = TimedRotatingFileHandler(
|
|
'power.log',
|
|
when='midnight',
|
|
interval=1,
|
|
backupCount=7,
|
|
delay=True
|
|
)
|
|
formatter = logging.Formatter(
|
|
'%(asctime)s [%(levelname)s] %(message)s',
|
|
datefmt='%Y-%m-%d %H:%M:%S'
|
|
)
|
|
handler.setFormatter(formatter)
|
|
logger.addHandler(handler)
|
|
return logger
|
|
|
|
def monitor_ups(logger):
|
|
client = PyNUTClient()
|
|
try:
|
|
ups_status = client.list_vars('ups')
|
|
return {
|
|
'status': ups_status.get('ups.status'),
|
|
'battery_charge': float(ups_status.get('battery.charge'))
|
|
}
|
|
except Exception as e:
|
|
logger.error(f"Error communicating with NUT: {e}")
|
|
return None
|
|
|
|
def send_email(subject, body, recipient_email, gmail_user, gmail_password, logger):
|
|
msg = MIMEText(body)
|
|
msg['Subject'] = subject
|
|
msg['From'] = gmail_user
|
|
msg['To'] = recipient_email
|
|
try:
|
|
with smtplib.SMTP('smtp.gmail.com', 587) as server:
|
|
server.starttls()
|
|
server.login(gmail_user, gmail_password)
|
|
server.send_message(msg)
|
|
logger.info(f"Email message sent: {subject}")
|
|
except Exception as e:
|
|
logger.info(f"Failed to send email: {e}")
|
|
|
|
def main():
|
|
logger = setup_logging()
|
|
logger.info("Starting UPS monitoring service.")
|
|
prev_status = None
|
|
while True:
|
|
ups_data = monitor_ups(logger)
|
|
if ups_data:
|
|
status = ups_data['status']
|
|
battery = ups_data['battery_charge']
|
|
if status and status != prev_status:
|
|
if status == "OB DISCHRG":
|
|
logger.warning(f"Outage detected! Power On Battery. {battery}% charge remaining.")
|
|
send_email(
|
|
f"{ups_id}: Power Outage Detected",
|
|
f"The {ups_id} UPS is running on battery power! {battery}% charge remaining.",
|
|
recipient_email,
|
|
gmail_user,
|
|
gmail_password,
|
|
logger
|
|
)
|
|
elif status == "OL":
|
|
logger.info(f"Power On Line. {battery}% charge remaining.")
|
|
send_email(
|
|
f"{ups_id}: Power On Line",
|
|
f"{ups_id} UPS power has been restored! {battery}% charge remaining.",
|
|
recipient_email,
|
|
gmail_user,
|
|
gmail_password,
|
|
logger
|
|
)
|
|
else:
|
|
logger.info(f"UPS status changed to: {status}")
|
|
prev_status = status
|
|
if status == "OB DISCHRG" and battery < 50:
|
|
logger.warning(f"Battery level low: {battery}% charge remaining.")
|
|
send_email(
|
|
f"{ups_id} battery low.",
|
|
f"{ups_id} battery level low: {battery}% charge remaining.",
|
|
recipient_email,
|
|
gmail_user,
|
|
gmail_password,
|
|
logger
|
|
)
|
|
time.sleep(30)
|
|
time.sleep(5)
|
|
|
|
if __name__ == "__main__":
|
|
main()
|
|
|