This commit is contained in:
samerbam 2023-06-22 13:14:13 -04:00
parent f9d54838d8
commit c696b8c964
9 changed files with 154 additions and 24 deletions

32
README.md Normal file
View File

@ -0,0 +1,32 @@
# A Siri like AI Assistant
---
* Uses ChatGPT for general queries
* Uses Wolfram Alpha for anything math related
* Has built in NLP (using a NLI model) for determining if we can process query locally
* Frontend/Backend architecture for ability to deploy lightweight clients
## Skills
---
[ ] Alarms
[ ] Calendar
[ ] Gmail
[ ] ChatGPT
[ ] Reminders
[ ] Timers
[ ] Todos
[ ] Weather
[ ] Wolfram
[x] NLP
[x] Speech to Text (frontend for sure)
[ ] API
[ ] Authentication
[ ] General API
[ ] TTS
- generate audio on backend or frontend?
- Perks of backend is fast generation
- Cons of backend is large file transfers between devices, lots of internet usage
- Perks of frontend is less data transfer between devices requiring less internet usage
- Cons of frontend is slower generation

0
backend/__init__.py Normal file
View File

0
backend/api.py Normal file
View File

View File

@ -1 +1,8 @@
# TODO: Handle all authentication stuff for verifying client is who we think it is in here.
# TODO: Handle all authentication stuff for verifying client is who we think it is in here.
# Example: https://github.com/miguelgrinberg/REST-auth
# User/Pass for initial token and refresh token generation, this should
# OAuth2 style authentication
# Flask-RESTFul with
#

View File

@ -7,11 +7,24 @@ from skills.timers import Timers
from skills.todos import Todos
from skills.weather import Weather
from skills.wolfram import Wolfram
from NLP import NLP
import sys
print(sys.version)
skills = [GPT(), Alarms(), Cal(), Gmail(), Reminders(), Timers(), Todos(), Weather(), Wolfram()]
skill_names = [skill.trigger_phrase for skill in skills]
print("test")
if __name__ == "__main__":
print("Skill Trigger Phrases: ")
for skill in skills:
print(skill.trigger_phrase)
# print("Skill Trigger Phrases: ")
print(f"Active Skills: {skill_names}")
nlp = NLP()
# for skill in skills:
# print(skill.trigger_phrase)

4
backend/requirements.txt Normal file
View File

@ -0,0 +1,4 @@
transformers
spacy
schedule
ctparse

1
backend/skills/auth.py Normal file
View File

@ -0,0 +1 @@
# TODO: Handle all authentication stuff for verifying client is who we think it is in here.

View File

@ -1,44 +1,107 @@
from skills.config import ntfy_url
import requests
if __name__ == "__main__": # Handle
from config import ntfy_url
else:
from skills.config import ntfy_url
import threading
import schedule
import time
# def job_that_executes_once():
# Do some work that only needs to happen once...
# return schedule.CancelJob
def run_continuously(schedule, interval=1):
# Borrowed from schedule documentation, why reinvent the wheel when its been created.
"""Continuously run, while executing pending jobs at each
elapsed time interval.
@return cease_continuous_run: threading. Event which can
be set to cease continuous run. Please note that it is
*intended behavior that run_continuously() does not run
missed jobs*. For example, if you've registered a job that
should run every minute and you set a continuous run
interval of one hour then your job won't be run 60 times
at each interval but only once.
"""
cease_continuous_run = threading.Event()
class ScheduleThread(threading.Thread):
@classmethod
def run(cls):
while not cease_continuous_run.is_set():
schedule.run_pending()
time.sleep(interval)
continuous_thread = ScheduleThread()
continuous_thread.start()
return cease_continuous_run
# while True:
# schedule.run_pending()
# time.sleep(1)
class Timers:
def __init__(self):
self.trigger_phrase = "timer"
self.timers = {}
self.schedule = schedule.Scheduler()
def _add_timer(self, time, name):
self.timers[name] = time
# use https://schedule.readthedocs.io/en/stable/examples.html#run-a-job-once to trigger self._trigger_timer()
def _remove_timer(self, name):
del self.timers[name]
def _trigger_timer(self, name):
if name in self.timers:
r = requests.post(f"https://ntfy.sh/{ntfy_url}",
data=f"{name}",
def _notify(self, device_id, timer_name):
r = requests.post(f"https://ntfy.sh/{device_id}",
data=f"{timer_name}",
headers={
"Title": "Your timer is going off!",
"Priority": "default",
"Tags": "bell"
})
print(r.text)
#TODO: send ntfy.sh to device
#TODO: play timer done sound
return r
def _add_timer(self, time, name):
if len(self.timers) == 0:
self.stop_run_continuously = run_continuously()
self.timers[name] = time
# duration =
self.schedule.every().day.at(time.strftime("%H:%M:%S", time.gmtime(duration))).do(self._trigger_timer(name)).tag(name)
# use https://schedule.readthedocs.io/en/stable/examples.html#run-a-job-once to trigger self._trigger_timer()
def _remove_timer(self, name):
del self.timers[name]
if len(self.timers) == 0:
self.stop_run_continuously.set()
def _trigger_timer(self, name):
if name in self.timers:
res = self._notify(ntfy_url, name).text
print(res)
self._remove_timer(name)
# #TODO: play timer done sound
return schedule.CancelJob
def get_remaining_time(self, name=""):
pass
# if name == "":
def run(self, query):
def run(self, query="", time=0, name=""):
if "add" in query:
time = 0 #TODO: Natural Language parse time out of phrase
self._add_timer(time)
self._add_timer(time, name)
return True # Return true to indicate success
if "remove" in query:
time = 0 #TODO: Natural Language parse time out of phrase
self._remove_timer(time)
self._remove_timer(name)
return True
return False # Return false to indicate failure
def _disable_timer_check_thread(self):
self.stop_run_continuously.set()
if __name__ == "__main__":

10
backend/skills/utility.py Normal file
View File

@ -0,0 +1,10 @@
from ctparse import ctparse
from datetime import datetime
def parsetime(phrase):
ts = datetime.now()
return ctparse(phrase, ts=ts)
if __name__ == "__main__":
print(parsetime('May 5th 2:30 in the afternoon').ref_time)