This repository has been archived by the owner on Jun 26, 2021. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathoutput.py
124 lines (90 loc) · 2.97 KB
/
output.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
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
import time
import json
import random
import threading
import re
import os.path
#m- motion
#s- status
#o- output
class OUTPUT_test():
def __init__(self, USB, FILE):
self.cmd_list = ["L", "I"]
self.USB = USB
self.FILE = FILE
def __del__(self):
pass
def load_test(self):
cmd = random.choice(self.cmd_list)
self.random_test('test_out', cmd, 1)
self.tests_list = self.data["CMDS"]
def random_test(self, group, cmd, y):
while 1==1:
x = random.randint(1,y)
name = ('tests/{}_{}_{}.json'.format(group, cmd, x))
if os.path.exists(name) == 1:
break
with open(name) as json_file:
self.data = json.load(json_file)
# print(name)
def start(self):
while 1==1:
# print(threading.currentThread())
time.sleep(1)
self.USB.buf_o.clear()
is_passed = 1
self.load_test()
for self.test in self.tests_list:
if self.test["DIR"]=='R':
# print("odbieram")
if self.recive()==0:
is_passed = 0
# print("niezdany")
break
# print("zdany")
elif self.test["DIR"]=='T':
# print("wysylam")
if self.transmite()==0:
is_passed = 0
break
else:
FILE.write_f("Bledna komenda okreslajaca kierunek transferu danych")
is_passed = 0
break
time.sleep(3)
if is_passed==0:
self.failed()
else:
self.passed()
# self.USB.buf_o.clear()
def failed(self):
self.FILE.write_f("{}==> FAILED\n".format(self.data['DESC']))
def passed(self):
self.FILE.write_p("{}==> PASSED\n".format(self.data['DESC']))
def recive(self):
t = time.time()
while time.time() < t+(self.test["TIMEOUT"]/100):
time.sleep(0.05)
if len(self.USB.buf_o) != 0:
self.answ = self.USB.buf_o.pop(0).decode('utf-8')
if self.is_correct_answ()==1:
return 1
else:
return 0
# print("timeout")
return 0
def is_correct_answ(self):
print(self.test["CMD"])
print(self.test["VAL"])
print(self.answ)
regex = r"^\b{}\b\s\b{}\b".format(self.test["CMD"], self.test["VAL"])
if re.search(regex ,self.answ):
return 1
else:
return 0
def transmite(self):
cmd_to_send = self.test["CMD_TO_SEND"]
cmd_to_send += "\r"
cmd_to_send = cmd_to_send.encode('utf-8')
self.USB.send(cmd_to_send)
return 1