Device.py
2.45 KB
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
80
81
82
from pyrosapp.models import *
import socket
class DeviceObj():
'''
Generic object for the communication with all the devices (need inheritance)
'''
def __init__(self):
'''
The messages and their parameters will be defined in the objects that inherits from this class
'''
set_msgs = []
get_msgs = []
do_msgs = []
self.ip = None
self.port = None
def init_socket(self):
if self.ip is None or self.port is None:
raise ValueError("IP and/or PORT not initialized")
self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.sock.connect((self.ip, self.port))
def set(self, cmd, *args):
'''
Send a SET message to the device
'''
msg = [msg for msg in self.set_msgs if msg[0] == cmd]
if len(msg) == 0:
raise ValueError("Invalid message argument %s" % (cmd,))
msg = msg[0]
if len(args) not in msg[1]:
raise ValueError("Bad number of arguments")
for arg in args:
if type(arg) != msg[2]:
raise TypeError("Bad type of argument: expected %s" % (msg[2],))
self.init_socket()
message = "SET " + cmd + " " + "".join([str(arg) for arg in args])
self.sock.send(bytes(message, "UTF-8"))
print("set ", cmd)
def get(self, cmd):
'''
Send a GET message to the device
'''
if cmd not in self.get_msgs:
raise ValueError("Invalid message argument %s" % (cmd,))
self.init_socket()
message = "GET " + cmd
self.sock.send(bytes(message, "UTF-8"))
print("get ", cmd)
ret = self.sock.recv(1024).decode()
print("return : ", ret)
return ret
def do(self, cmd, *args):
'''
Send a DO message to the device
'''
msg = [msg for msg in self.do_msgs if msg[0] == cmd]
if len(msg) == 0:
raise ValueError("Invalid message argument %s" % (cmd,))
msg = msg[0]
if len(args) not in msg[1]:
raise ValueError("Bad number of arguments")
for arg in args:
if type(arg) != msg[2]:
raise TypeError("Bad type of argument: expected %s" % (msg[2],))
self.init_socket()
message = "DO " + cmd + " " + "".join([str(arg) for arg in args])
self.sock.send(bytes(message, "UTF-8"))
print("do ", cmd)