142 lines
5.0 KiB
Python
142 lines
5.0 KiB
Python
# unreal_api.py
|
|
|
|
from PySide6.QtCore import QObject, QUrl, Signal, QByteArray
|
|
from PySide6.QtWebSockets import QWebSocket
|
|
import json
|
|
from datetime import datetime, timedelta
|
|
from Core.UnrealApi.message_filter import MessageFilter, MessageFilterResultType
|
|
test = '''{
|
|
"Sender": "F2DCD99C267C4F84B50CF091AA4ED608",
|
|
"Recipients": [],
|
|
"MessageType": "/Script/AvalancheMediaEditor.AvaRundownPing",
|
|
"Expiration": ,
|
|
"Scope": "Network",
|
|
"Annotations": {},
|
|
"Message":
|
|
{
|
|
"bAuto": false
|
|
}
|
|
}'''
|
|
|
|
class AvaPing:
|
|
def __init__(self, default_expiration=5):
|
|
self.default_expiration = default_expiration
|
|
self.Sender = "F2DCD99C267C4F84B50CF091AA4ED608"
|
|
self.Recipients = []
|
|
self.MessageType = "/Script/AvalancheMediaEditor.AvaRundownPing"
|
|
self.Expiration = None
|
|
self.Message = {"bAuto": False}
|
|
self.Annotations = {}
|
|
self.Scope = "Network"
|
|
|
|
def calculate_expire(self):
|
|
expiration_time = datetime.now() + timedelta(minutes=self.default_expiration)
|
|
return int(expiration_time.timestamp())
|
|
|
|
def update_expire(self):
|
|
self.Expiration = self.calculate_expire()
|
|
|
|
def auto_ping(self) -> str:
|
|
self.update_expire()
|
|
me = self.__dict__
|
|
del me["default_expiration"]
|
|
me["Message"] = {"bAuto": True}
|
|
return self.to_json()
|
|
|
|
def to_json(self):
|
|
self.update_expire()
|
|
me = self.__dict__
|
|
del me["default_expiration"]
|
|
return json.dumps(me)
|
|
|
|
# def __dict__(self):
|
|
# self.Expiration = self.calculate_expire()
|
|
# return self.__dict__
|
|
|
|
|
|
class UnrealApi(QObject):
|
|
# Definitions des Signaux
|
|
message_sent = Signal(object)
|
|
message_received = Signal(object)
|
|
ws_connected = Signal(bool)
|
|
|
|
def __init__(self, uri, parent=None):
|
|
super().__init__(parent)
|
|
self.uri = uri
|
|
self.websocket = None
|
|
self.connected = False
|
|
self._filterOnlyAva = False
|
|
self._filter = MessageFilter(key_filter="MessageType", filter="Ava")
|
|
|
|
def ws_connect(self):
|
|
if not self.websocket:
|
|
self.websocket = QWebSocket()
|
|
self.websocket.open(QUrl(self.uri))
|
|
self.websocket.binaryMessageReceived.connect(self.on_message_received)
|
|
self.websocket.connected.connect(self.on_connected)
|
|
self.websocket.disconnected.connect(self.on_disconnected)
|
|
|
|
def on_connected(self):
|
|
print("Connecté au serveur WebSocket")
|
|
self.connected = True
|
|
self.ws_connected.emit(True) # Émettre le signal de connexion établie
|
|
|
|
def on_disconnected(self):
|
|
print("Déconnecté du serveur WebSocket")
|
|
self.connected = False
|
|
self.ws_connected.emit(False) # Émettre le signal de déconnexion
|
|
|
|
def send_message(self, message):
|
|
if self.connected:
|
|
data = json.loads(message)
|
|
current_time = datetime.now()
|
|
expiration_time = current_time + timedelta(minutes=5)
|
|
unix_expiration = int(expiration_time.timestamp())
|
|
data["Expiration"] = unix_expiration
|
|
message_with_expiration = json.dumps(data)
|
|
self.websocket.sendTextMessage(message_with_expiration)
|
|
print(f"message sent: {data}")
|
|
self.message_sent.emit(data)
|
|
|
|
def on_message_received(self, data: QByteArray):
|
|
try:
|
|
message = json.loads(data.data())
|
|
print(f"received message: {message}")
|
|
except json.JSONDecodeError as e:
|
|
print(f"Error decoding JSON: {e}")
|
|
except UnicodeDecodeError as e:
|
|
print(f"UnicodeDecodeError decoding from server: {e}")
|
|
except TypeError as e:
|
|
print(f"TypeError decoding data from server: {e}")
|
|
except AttributeError as e:
|
|
print(f"AttributeError decoding data from server: {e}")
|
|
else:
|
|
print(f"message recived: {message}")
|
|
# print(f"MessageType: {message['MessageType']}")
|
|
# more complicated filter needed
|
|
# print(f"filter state : {self._filterOnlyAva}")
|
|
|
|
if( not self._filterOnlyAva):
|
|
# print("filter not activated")
|
|
self.message_received.emit(message)
|
|
return
|
|
else:
|
|
# print("filter activated")
|
|
if(self._filter.is_filtered(message) != MessageFilterResultType.NotFound ):
|
|
self.message_received.emit(message)
|
|
return
|
|
# self.message_received.emit(message)
|
|
|
|
|
|
def set_filtered(self, filtered: bool):
|
|
print(f"filtered activation state changed: {filtered}")
|
|
self._filterOnlyAva = filtered
|
|
|
|
def close_websocket(self):
|
|
if self.websocket:
|
|
self.websocket.close()
|
|
|
|
def send_ping(self):
|
|
ping = AvaPing()
|
|
# print(f"ping: {ping.to_json()}")
|
|
self.send_message(ping.to_json()) |