89 lines
2.6 KiB
Python
89 lines
2.6 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
|
||
|
|
|
||
|
|
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 ups_status.get('ups.status')
|
||
|
|
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:
|
||
|
|
status = monitor_ups(logger)
|
||
|
|
if status and status != prev_status:
|
||
|
|
if status == "OB DISCHRG": # "On Battery"
|
||
|
|
logger.warning("Outage detected! Power On Battery.")
|
||
|
|
send_email(
|
||
|
|
"Power Outage Detected",
|
||
|
|
"The UPS is running on battery power!",
|
||
|
|
recipient_email,
|
||
|
|
gmail_user,
|
||
|
|
gmail_password,
|
||
|
|
logger
|
||
|
|
)
|
||
|
|
elif status == "OL": # "On Line"
|
||
|
|
logger.info("Power On Line.")
|
||
|
|
send_email(
|
||
|
|
"Power On Line",
|
||
|
|
"UPS power has been restored!",
|
||
|
|
recipient_email,
|
||
|
|
gmail_user,
|
||
|
|
gmail_password,
|
||
|
|
logger
|
||
|
|
)
|
||
|
|
else:
|
||
|
|
logger.info(f"UPS status changed to: {status}")
|
||
|
|
prev_status = status
|
||
|
|
time.sleep(5)
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
main()
|
||
|
|
|