1
0
mirror of https://github.com/MarkParker5/STARK.git synced 2025-07-12 22:50:22 +02:00

move singleton to parent class Control

add MyTeleBot
add libs.txt
This commit is contained in:
MarkParker5
2021-10-17 23:52:57 +03:00
parent 04f38b1ed3
commit 22a440379d
8 changed files with 65 additions and 25 deletions

View File

@ -1,6 +1,12 @@
from abc import ABC, abstractmethod from abc import ABC, abstractmethod
class Control(ABC): class Control(ABC):
# Singleton
def __new__(cls):
if not hasattr(cls, 'instance'):
cls.instance = super().__new__(cls)
return cls.instance
@abstractmethod @abstractmethod
def start(self): def start(self):

View File

@ -2,5 +2,5 @@ from ..Control import Control
class Django(Control): class Django(Control):
def start(self): def start(self):
# run manage.py #TODO: run manage.py
pass pass

View File

@ -10,12 +10,6 @@ class RemoteControl(Control):
url = 'http://t97316v1.beget.tech/read' url = 'http://t97316v1.beget.tech/read'
session = requests.Session() session = requests.Session()
# Singleton
def __new__(cls):
if not hasattr(cls, 'instance'):
cls.instance = super(RemoteControl, cls).__new__(cls)
return cls.instance
def start(self): def start(self):
while True: while True:
time.sleep(1) time.sleep(1)

View File

@ -0,0 +1,46 @@
from telebot import TeleBot
# added callback for __threaded_polling
class MyTeleBot(TeleBot):
def polling(self, none_stop=False, interval=0, timeout=20, callback = lambda *args: None, args = () ):
if self.threaded:
self.__threaded_polling(none_stop, interval, timeout, callback, args)
else:
self.__non_threaded_polling(none_stop, interval, timeout)
def __threaded_polling(self, none_stop=False, interval=0, timeout=20, callback = lambda *args: None, args = () ):
self.__stop_polling.clear()
error_interval = 0.25
polling_thread = util.WorkerThread(name="PollingThread")
or_event = util.OrEvent(
polling_thread.done_event,
polling_thread.exception_event,
self.worker_pool.exception_event
)
while not self.__stop_polling.wait(interval):
callback(*args) # added by Parker
or_event.clear()
try:
polling_thread.put(self.__retrieve_updates, timeout)
or_event.wait() # wait for polling thread finish, polling thread error or thread pool error
polling_thread.raise_exceptions()
self.worker_pool.raise_exceptions()
error_interval = 0.25
except apihelper.ApiException as e:
if not none_stop:
self.__stop_polling.set()
else:
polling_thread.clear_exceptions()
self.worker_pool.clear_exceptions()
time.sleep(error_interval)
error_interval *= 2
except KeyboardInterrupt:
self.__stop_polling.set()
break
polling_thread.stop()

View File

@ -2,11 +2,11 @@
import time import time
import os import os
import telebot
import config import config
from Features import Command from Features import Command
from General import Text2Speech from General import Text2Speech
from Controls.Control import Control from Controls.Control import Control
from .MyTeleBot import MyTeleBot
class TelegramBot(Control): class TelegramBot(Control):
threads = [] threads = []
@ -14,13 +14,7 @@ class TelegramBot(Control):
voids = 0 voids = 0
memory = [] memory = []
voice = Text2Speech.Engine() voice = Text2Speech.Engine()
bot = telebot.TeleBot(config.telebot) bot = MyTeleBot(config.telebot)
# Singleton
def __new__(cls):
if not hasattr(cls, 'instance'):
cls.instance = super(TelegramBot, cls).__new__(cls)
return cls.instance
def reply(self, id, response): def reply(self, id, response):
if response.text: if response.text:

View File

@ -16,12 +16,6 @@ class VoiceAssistant(Control):
lastClapTime = 0 lastClapTime = 0
doubleClap = False doubleClap = False
# Singleton
def __new__(cls):
if not hasattr(cls, 'instance'):
cls.instance = super(VoiceAssistant, cls).__new__(cls)
return cls.instance
def start(self): def start(self):
self.listener.listen_noise() self.listener.listen_noise()
os.system('clear') os.system('clear')

10
libs.txt Normal file
View File

@ -0,0 +1,10 @@
pip install speech_recognition
pip install google-cloud-texttospeech
pip install pygame
pip install bs4
pip install wikipedia
pip install xlrd
pip install xlwt
pip install xlutils
pip install pyaudio # if instalation fails, try install from .whl (https://www.lfd.uci.edu/~gohlke/pythonlibs/#pyaudio)
pip install pip install PyTelegramBotApi

View File

@ -4,10 +4,6 @@
import multiprocessing import multiprocessing
import Controls import Controls
# TODO:
# start oll controls in own thread or subprocess:
# voice assistant, telegram bot, django(api and ui)
def main(): def main():
controls = [ controls = [
Controls.VoiceAssistant(), Controls.VoiceAssistant(),