mirror of
https://git.roussel.pro/telecom-paris/pact.git
synced 2026-02-09 02:20:17 +01:00
96 lines
3.6 KiB
Python
96 lines
3.6 KiB
Python
from hand_detector import HandDetector
|
|
from audio_detector import record, analyze, test
|
|
from network import ApiClient, WebsocketServer
|
|
import time
|
|
|
|
#Classe qui coordonne les différents modules et qui s'occupe de construire l'avis au fur et a mesure
|
|
class Manager():
|
|
def __init__(self):
|
|
self.state = 0
|
|
self.defualtAvis = {
|
|
"note": None,
|
|
"commentaire": None,
|
|
"notes_autres": {}
|
|
}
|
|
|
|
self.TIMEOUT_CAMERA = 5
|
|
|
|
self.avis = self.defualtAvis
|
|
self.server = WebsocketServer(None)
|
|
self.server.start()
|
|
self.handDetector = HandDetector()
|
|
self.api = ApiClient()
|
|
self.timeLastChange = time.time()
|
|
self.isLastHandPacketEmpty = False
|
|
print("Backend ready")
|
|
|
|
#Boucle principale
|
|
def loop(self):
|
|
while(True):
|
|
if(self.state == 0):
|
|
self.sleep()
|
|
if(self.state == 1):
|
|
self.camera()
|
|
if(self.state == 2):
|
|
self.audio()
|
|
if(self.state == 3):
|
|
self.thankYou()
|
|
time.sleep(0.01)
|
|
|
|
#Fonction qui est executée pendant que la borne est en veille, reveille la borne si une main est detectée
|
|
def sleep(self):
|
|
res = self.handDetector.detect()
|
|
if(res != False):
|
|
self.state = 1
|
|
self.timeLastChange = time.time()
|
|
self.server.sendMessage({"type": "state", "state": 1})
|
|
|
|
#Envoie la position de la main a l'écran et passe a l'étape suivante si une main est detectée pendant assez longtemps
|
|
def camera(self):
|
|
if(time.time() - self.timeLastChange > self.TIMEOUT_CAMERA):
|
|
self.server.sendMessage({"type":"reset"})
|
|
self.reset()
|
|
return
|
|
|
|
res = self.handDetector.detect()
|
|
if(res != False):
|
|
state, coords, size, finalDecision, progress = res
|
|
self.server.sendMessage({"type": "effects", "effects": [{"type": "loading", "x":coords[0], "y": coords[1], "width": size, "height": size, "progress": progress},{"type": state, "x":coords[0], "y": coords[1], "width": size, "height": size}]})
|
|
self.isLastHandPacketEmpty = False
|
|
self.timeLastChange = time.time()
|
|
if(finalDecision != False):
|
|
self.avis["note"] = 10 if finalDecision == "thumbs_up" else 0
|
|
self.state = 2
|
|
self.server.sendMessage({"type": "state", "state": 2})
|
|
elif self.isLastHandPacketEmpty == False:
|
|
self.server.sendMessage({"type":"effects","effects":[]})
|
|
self.isLastHandPacketEmpty = True
|
|
|
|
def audio(self):
|
|
audio_query,sr = record()
|
|
self.server.sendMessage({"type":"recording_done"})
|
|
result = analyze(audio_query,sr)
|
|
# result = test()
|
|
if(result != False):
|
|
print("mot detecté : " + result["display"] + " avec une note de " + str(result["grade"]))
|
|
self.server.sendMessage({"type":"new_grade","word":result["display"]})
|
|
self.avis["notes_autres"]["test"] = result["grade"]
|
|
time.sleep(3)
|
|
self.state = 3
|
|
self.server.sendMessage({"type": "state", "state": 3})
|
|
|
|
def thankYou(self):
|
|
time.sleep(10)
|
|
print("Reseting...")
|
|
self.timeLastChange = time.time()
|
|
self.server.sendMessage({"type": "state", "state": 0})
|
|
res = self.api.send(self.avis["note"],self.avis["notes_autres"]["test"])
|
|
print(res.text)
|
|
self.reset()
|
|
|
|
def reset(self):
|
|
self.state = 0
|
|
self.avis = self.defualtAvis
|
|
self.handDetector.reset()
|
|
|