-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathdevicesData.py
79 lines (64 loc) · 2.52 KB
/
devicesData.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
import os
import json
from datetime import datetime, timedelta
class DevicesDataClass:
instance = None
data = {}
lastCommandDate = datetime.now()
path = os.path.dirname(os.path.abspath(__file__))
_observers = []
def __new__(cls):
if cls.instance is None:
cls.instance = super(DevicesDataClass, cls).__new__(cls)
cls.instance.load()
return cls.instance
def __getitem__(self, key):
return self.data[key]
def __setitem__(self, key, value):
self.data[key] = value
def setCommand(self, id, type, value):
self.data[id]['commands'][type] = value
self.notify_observers(id)
def updateLastCommandDate(self):
self.lastCommandDate = datetime.now()
def commandReceivedRecently(self):
# Calculer la différence entre les deux dates
difference = abs(datetime.now() - self.lastCommandDate)
# Vérifier si la différence est inférieure à 2 secondes
if difference < timedelta(seconds=2):
return True
else:
return False
def load(self):
with open(self.path + '/config/devices.json', 'r') as jsonfile:
data = json.load(jsonfile)
self.data = {}
for desc in data:
id = desc["id"]
self.data[id] = {}
self.data[id]["desc"] = desc
self.data[id]["status"] = {}
self.data[id]["commands"] = {}
switchCommandId = None
for key, value in desc["mapping"].items():
if "code" in value and (value["code"] == "switch" or value["code"] == "switch_1"):
switchCommandId = key
break
self.data[id]["desc"]["switchCommandId"]=switchCommandId
def save(self):
with open(self.path + '/config/snapshot.json', 'w') as jsonfile:
json.dump(DevicesData.data, jsonfile, indent=2)
def getIdFromName(self, name):
for id, device in self.data.items():
if device['desc']['name'] == name:
return id
return null
def add_observer(self, observer):
self._observers.append(observer)
def remove_observer(self, observer):
self._observers.remove(observer)
def notify_observers(self,id):
for observer in self._observers:
observer.onCommandReceived(id,self.data[id]['commands'])
# Exemple d'utilisation
DevicesData = DevicesDataClass()