import os
import subprocess as sp
import datetime
from calendar import monthrange
from shutil import copyfile
[docs]class WittyPi():
''' Central instance to run functions on the wittyPy.
:param path:
Path to your wittyPi folder (Where "wittyPi.sh", "utilities.sh", "daemon.sh" can be found).
:type path: ``str``
:param next_start:
WittyTime object, if a startup is set, else None.
:type next_start: ``WittyTime``
:param next_shutdown:
WittyTime object, if a shutdown is set, else None.
:type next_shutdown: ``WittyTime``
:param available_schedules:
List of available schedules.
:type available_schedules: ``list``
:param timedelta:
Difference between system time and Realtime Clock time. If positive, system time is ahead of the RTC and vice versa.
:type timedelta: ``float``
'''
required_files = ["wittyPi.sh", "utilities.sh", "daemon.sh"]
def __init__(self, path):
self.path = path
self.next_start = None
self.next_shutdown = None
self.timedelta = None
self.available_schedules = []
self._initial_check()
def __repr__(self):
return "WittyPi2(State: {}, {}, {}, Available Schedules: {})".format(self.path, self.next_start, self.next_shutdown, self.available_schedules)
def __run_util_function(self, function, *args):
connection = sp.run(["sudo", "bash","-c", "source "+ self.path +"utilities.sh && " + function + " "+" ".join([str(arg) for arg in args])], stdout=sp.PIPE)
return (connection.returncode, connection.stdout.decode("UTF-8"))
def _initial_check(self):
for f in self.required_files:
if not os.path.isfile(self.path+f):
raise Exception("Requirement Error: WittyPi Software '{}' not in '{}'.".format(f, self.path))
if self.__run_util_function("is_rtc_connected")[0] == 1:
raise Exception("Requirement Error: WittyPi2 Hardware not correctly connected.")
self._get_startup()
self._get_shutdown()
self._list_schedules()
self._compare_times()
def _split_date_str(self, date_str):
return date_str.replace('\n', '').replace(":", " ").split(" ")
def _get_startup(self):
t = self.__run_util_function("get_startup_time")
local_t = self.__run_util_function("get_local_date_time", "'{}'".format(t[1].replace("\n", "")))
if len(local_t[1]) < 5:
self.next_start = WittyTime(name = "boot")
else:
self.next_start = WittyTime(*self._split_date_str(local_t[1]), name="boot")
def _get_shutdown(self):
t = self.__run_util_function("get_shutdown_time")
local_t = self.__run_util_function("get_local_date_time", "'{}'".format(t[1].replace("\n", "")))
if len(local_t[1]) < 5:
self.next_shutdown = WittyTime(name = "shutdown")
else:
self.next_shutdown = WittyTime(*self._split_date_str(local_t[1]), name="shutdown")
def _list_schedules(self):
tmpls = []
for f in os.listdir(self.path+"schedules"):
if f.endswith(".wpi"):
tmpls.append(f)
self.available_schedules = sorted(tmpls)
[docs] def activate_schedule(self, index):
'''Activates a WittyPi schedule, chosen by their index in self.available_schedules.'''
if index -1 > len(self.available_schedules):
raise Exception("Index Error: No schedule at index {} of self.available_schedules.".format(index))
copyfile(self.path + "schedules/" + self.available_schedules[index], self.path + "/schedule.wpi")
sp.run(["sudo", "bash","-c", "source "+ self.path +"runScript.sh"], stdout=sp.PIPE)
self._get_startup()
self._get_shutdown()
[docs] def set_startup(self, *args):
'''Set startup for the WittyPi. Takes up to four \*args.
:param \*args:
1. Day (``int`` or ``str``): Either integer between 1-31 or "??" for wildcard.
2. Hour (``int`` or ``str``): Either integer between 0-23 or "??" for wildcard.
3. Minute (``int``): Integer between 0-59.
4. Second (``int``): Integer between 0-59.
'''
new_time = WittyTime(*args)
when = self.__run_util_function("get_utc_date_time", new_time.day, new_time.hour, new_time.minute, new_time.second)
self.__run_util_function("set_startup_time", *self._split_date_str(when[1]))
self._get_startup()
[docs] def set_shutdown(self, *args):
'''Set shutdown for the WittyPi. Takes up to three \*args.
:param \*args:
1. Day (``int`` or ``str``): Either integer between 1-31 or "??" for wildcard.
2. Hour (``int`` or ``str``): Either integer between 0-23 or "??" for wildcard.
3. Minute (``int``): Integer between 0-59.
'''
new_time = WittyTime(*args)
when = self.__run_util_function("get_utc_date_time", new_time.day, new_time.hour, new_time.minute, new_time.second)
self.__run_util_function("set_shutdown_time", *self._split_date_str(when[1]))
self._get_shutdown()
[docs] def reset(self, option):
'''Resets the WittyPi according to options
:param option:
Options: startup, shutdown, script, all.
:type options: ``str``
'''
options = ["startup", "shutdown", "script", "all"]
if option not in options:
raise Exception('Error: Chose one of these options: "startup", "shutdown", "script", "all"')
if option != "shutdown":
self.__run_util_function("clear_startup_time")
if option != "startup":
self.__run_util_function("clear_shutdown_time")
if option in ["script", "all"]:
try:
os.remove(self.path + "/schedule.wpi")
except:
pass
self._get_startup()
self._get_shutdown()
def _compare_times(self):
'''Compare RTC time with system time. Positive self.timdelta means your system time is ahead of the RTC and vice versa.'''
rtc = self.__run_util_function("get_rtc_time")[1].replace("\n", "")
sys_time = datetime.datetime.now()
self.timedelta = (sys_time - datetime.datetime.strptime(rtc, "%a %d %b %Y %H:%M:%S %Z")).total_seconds()
[docs]class WittyTime():
'''WittyTime object to validate and store times for the WittyPi.
:param name:
Type of time. Shutdown or Boot.
:type name: ``str``
:param as_date:
WittyTime as datetime object.
:type as_date: ``datetime``
:param day:
:type day: ``int`` or ``str``
:param hour:
:type hour: ``int`` or ``str``
:param minute:
:type minute: ``int`` or ``str``
:param second:
:type second: ``int`` or ``str``
'''
def __init__(self, *args, name=None):
self.name = name
self.day = "unset"
self.hour = "unset"
self.minute = "unset"
self.second = "unset"
self.as_date = "unset"
self.__validate(*args)
def __validate(self, *args):
self.__check_format(args)
self.__to_datetime()
def __check_format(self, args):
args = [arg for arg in args]
if len(args) == 0:
return
for var in args:
try:
args[args.index(var)] = int(var)
except:
pass
if args[0] == 0:
return
elif args[0] not in range(1, 32) and args[0] != "??":
print(args[0])
print(type(args[0]))
raise Exception("Format Error: Day must be an integer between 1-31 or '??' for wildcard.")
else:
self.day = args[0]
if len(args) == 1:
self.hour = 00
self.minute = 00
self.second = 00
return
elif args[1] in range(24):
self.hour = args[1]
elif args[1] == "??":
if not self.day == "??":
raise Exception("Format Error: Wildcards must be used from left to right.")
else:
self.hour = args[1]
else:
raise Exception("Format Error: Hour must be an integer between 0-24 or '??' for wildcard.")
if len(args) == 2:
self.minute = 00
self.second = 00
return
elif args[2] in range(60):
self.minute = args[2]
elif args[2] == "??":
raise Exception("Format Error: Wildcards should only be used for Days and Hours. Not for Minutes")
else:
raise Exception("Format Error: Minute must be integer between 0-59.")
if len(args) == 3:
self.second = 00
elif args[3] in range(60):
self.second = args[3]
elif args[3] == "??":
raise Exception("Format Error: Wildcards should only be used for Days and Hours. Not for Seconds")
else:
raise Exception("Format Error: Second must be integer between 0-59.")
return
def __to_datetime(self):
if self.day == 'unset':
return
now = datetime.datetime.now()
next = {'year':now.year, 'month': now.month, 'minute': self.minute, 'second':self.second}
extramonth = False
if self.hour == '??':
if self.minute < now.minute:
next['hour'] = now.hour + 1
if next['hour'] > 23:
next['hour'] = 0
else:
next['hour'] = now.hour
else:
next['hour'] = self.hour
if self.day == '??':
if next['hour'] < now.hour or next['hour'] == now.hour and next['minute'] < now.minute:
next['day'] = now.day + 1
if next['day'] > monthrange(now.year, now.month)[1]:
next['day'] = 1
extramonth = True
else:
next['day'] = now.day
else:
next['day'] = self.day
if extramonth:
next['month'] = now.month +1
if next['month'] == 13:
next['month'] = 1
next['year'] = now.year +1
next = '{}-{}-{} {}:{}:{}'.format(next['year'], next['month'], next['day'], next['hour'], next['minute'], next['second'])
self.as_date = datetime.datetime.strptime(next, "%Y-%m-%d %H:%M:%S")
[docs] def t_left(self):
'''Returns time in seconds between WittyTime and system time.'''
if self.as_date == 'unset':
return "{} is unset.".format(self.name)
return self.as_date - datetime.datetime.now()
def __str__(self):
if self.name in ['boot', 'shutdown']:
return "Next {} at {}".format(self.name, self.as_date)
else:
return "WittyTime: {}".format(self.as_date)
def __repr__(self):
return "WittyTime(type={}, day={}, hour={}, minute={}, second={}, as_date={})".format(self.name, self.day, self.hour, self.minute, self.second, self.as_date)
