257 lines
9.3 KiB
Python
257 lines
9.3 KiB
Python
import sys
|
|
import time
|
|
import weakref
|
|
from datetime import datetime
|
|
|
|
from lib.helpers import timing
|
|
from PyQt5.QtCore import Qt, QTimer, pyqtSignal
|
|
from PyQt5.QtGui import QColor, QPalette, QPixmap
|
|
from ui import Dialog
|
|
from ui.test_admin_permission import Test_Admin_Permission
|
|
from ui.widget import Widget
|
|
|
|
|
|
class Test_Test(Widget):
|
|
ok = pyqtSignal(object)
|
|
ko = pyqtSignal(object)
|
|
|
|
def __init__(self, components=None, recipe=None, step=None, pieces=None, run_once=False, reset_on_start=True, enable_override=True):
|
|
super().__init__()
|
|
self.components = components
|
|
self.recipe = recipe
|
|
self.step = step
|
|
self.pieces = pieces
|
|
self.run_once = run_once
|
|
self.reset_on_start = reset_on_start
|
|
self.enable_override = enable_override
|
|
# setup variables
|
|
self.ok_counter = 0
|
|
self.ok_counter_limit = 1
|
|
# setup controls
|
|
self.done = False
|
|
self.done_ok = None
|
|
self.ok_timer = QTimer()
|
|
self.ok_timer.setSingleShot(True)
|
|
self.ok_timer.setInterval(1000)
|
|
self.ok_timer.timeout.connect(self.emit_ok)
|
|
self.ko_timer = QTimer()
|
|
self.ko_timer.setSingleShot(True)
|
|
self.ko_timer.setInterval(1000)
|
|
self.ko_timer.timeout.connect(self.emit_ko)
|
|
# setup save frame button
|
|
self.last = None
|
|
if hasattr(self, "save_b"):
|
|
self.save_b.setEnabled(False)
|
|
self.save_b.clicked.connect(self.save_last)
|
|
# setup override
|
|
self.admin_challenged = False
|
|
if self.enable_override:
|
|
self.override_b.setVisible(True)
|
|
self.override_b.setEnabled(True)
|
|
elif hasattr(self, "override_b"):
|
|
self.override_b.setVisible(False)
|
|
self.override_b.setEnabled(False)
|
|
if hasattr(self, "override_b"):
|
|
self.override_b.clicked.connect(self.override)
|
|
# setup vision staus gui
|
|
self.parent_assembly_widget = None
|
|
self.status_imgs_full = {
|
|
True: QPixmap("src/ui/imgs/success.png"),
|
|
"": QPixmap("src/ui/imgs/neo.ico"),
|
|
"warning": QPixmap("src/ui/imgs/warning.png"),
|
|
"wait": QPixmap("src/ui/imgs/wait.png"),
|
|
False: QPixmap("src/ui/imgs/fail.png"),
|
|
None: QPixmap("src/ui/imgs/wait.png"),
|
|
}
|
|
self.status_imgs_small = {k: i.scaled(32, 32, Qt.KeepAspectRatio, Qt.SmoothTransformation) for k, i in self.status_imgs_full.items()}
|
|
self.status_palettes = {
|
|
True: QPalette(),
|
|
"": QPalette(),
|
|
"warning": QPalette(),
|
|
False: QPalette(),
|
|
None: QPalette(),
|
|
}
|
|
self.status_palettes[True].setColor(QPalette.Base, Qt.green)
|
|
self.status_palettes[False].setColor(QPalette.Base, Qt.red)
|
|
self.status_palettes["warning"].setColor(QPalette.Base, QColor(255, 165, 0))
|
|
self.status_palettes[""].setColor(QPalette.Base, QColor(255, 255, 0))
|
|
self.reset()
|
|
self.visualize()
|
|
|
|
def set_parent_assembly_widget(self, parent_assembly_widget=None):
|
|
if parent_assembly_widget is None:
|
|
self.parent_assembly_widget = None
|
|
elif type(parent_assembly_widget) is weakref.ReferenceType:
|
|
self.parent_assembly_widget = parent_assembly_widget
|
|
else:
|
|
self.parent_assembly_widget = weakref.ref(parent_assembly_widget)
|
|
|
|
def start(self, recipe=None, step=None, pieces=None):
|
|
self.admin_challenged = False
|
|
self.recipe = recipe
|
|
self.step = step
|
|
self.pieces = pieces
|
|
if self.run_once is True and self.done is True and self.done_ok is True:
|
|
return False
|
|
if self.reset_on_start:
|
|
self.reset()
|
|
if self.enable_override:
|
|
self.override_b.setVisible(True)
|
|
self.override_b.setEnabled(True)
|
|
elif hasattr(self, "override_b"):
|
|
self.override_b.setVisible(False)
|
|
self.override_b.setEnabled(False)
|
|
self.visualize()
|
|
# start test
|
|
self.start_time = timing()
|
|
# TESTING
|
|
if "--test" in sys.argv:
|
|
if self.enable_override and hasattr(self, "_test_overridden_test_"):
|
|
del self._test_overridden_test_
|
|
# /TESTING
|
|
return True
|
|
|
|
def stop(self):
|
|
pass
|
|
|
|
def reset(self):
|
|
self.ok_counter = 0
|
|
self.done = False
|
|
self.done_ok = None
|
|
self.last = None
|
|
|
|
def get(self, data=None, override=False, fail=False, preserve_counter=False, skip_delay=False):
|
|
cur_timing = timing()
|
|
if self.done: # avoid proccessing if completed
|
|
return
|
|
if data is None:
|
|
data = self.last if self.last is not None else {}
|
|
else:
|
|
data_usable = [d is not None for d in reversed(data)]
|
|
if not any(data_usable):
|
|
data = self.last if self.last is not None else {}
|
|
else:
|
|
data = data[-data_usable.index(True) - 1]
|
|
data["time"] = datetime.now().strftime("%H:%M:%S")
|
|
if fail:
|
|
result_ok = False
|
|
elif override:
|
|
result_ok = True
|
|
else:
|
|
result_ok = data.get("results", {}).get("ok", None)
|
|
duration=cur_timing - self.start_time
|
|
data.pop("step",None)
|
|
self.last = {
|
|
**data,
|
|
"overridden": override,
|
|
"duration": float(f"{duration:.2f}"),
|
|
"ok": result_ok,
|
|
}
|
|
if fail:
|
|
self.ok_counter = 0
|
|
elif override:
|
|
self.ok_counter = self.ok_counter_limit
|
|
elif result_ok is True:
|
|
self.ok_counter += 1
|
|
elif not preserve_counter:
|
|
self.ok_counter = 0
|
|
# check if completed
|
|
if fail:
|
|
self.done = True
|
|
self.done_ok = False
|
|
if hasattr(self, "override_b"):
|
|
self.override_b.setEnabled(False)
|
|
self.stop()
|
|
if skip_delay:
|
|
self.emit_ko()
|
|
else:
|
|
self.ko_timer.start()
|
|
elif self.ok_counter >= self.ok_counter_limit:
|
|
self.done = True
|
|
self.done_ok = True
|
|
if hasattr(self, "override_b"):
|
|
self.override_b.setEnabled(False)
|
|
self.stop()
|
|
if skip_delay:
|
|
self.emit_ok()
|
|
else:
|
|
self.ok_timer.start()
|
|
self.visualize(self.last)
|
|
# TESTING
|
|
if "--test" in sys.argv:
|
|
if self.enable_override and not hasattr(self, "_test_overridden_test_"):
|
|
self._test_overridden_test_ = True
|
|
self.override_b.click()
|
|
# /TESTING
|
|
|
|
def visualize(self, data=None, img=None):
|
|
if data is None:
|
|
data = {}
|
|
overridden = data.get("overridden", False)
|
|
results = data.get("results", None)
|
|
if hasattr(self, "save_b"):
|
|
self.save_b.setEnabled(self.last is not None)
|
|
if hasattr(self, "state_l"):
|
|
if overridden:
|
|
self.state_l.setPixmap(self.status_imgs_small["warning"])
|
|
elif results is None:
|
|
self.state_l.setPixmap(self.status_imgs_small[None])
|
|
elif results.get("ok", None) is True:
|
|
self.state_l.setPixmap(self.status_imgs_small[True])
|
|
else:
|
|
self.state_l.setPixmap(self.status_imgs_small[False])
|
|
if hasattr(self, "ok_counter_pb"):
|
|
self.ok_counter_pb.setMaximum(self.ok_counter_limit)
|
|
self.ok_counter_pb.setValue(min(self.ok_counter, self.ok_counter_limit))
|
|
if self.ok_counter >= self.ok_counter_limit:
|
|
self.ok_counter_pb.setPalette(self.status_palettes[True])
|
|
else:
|
|
self.ok_counter_pb.setPalette(self.status_palettes[False])
|
|
if hasattr(self, "img_l"):
|
|
if img is None:
|
|
if overridden:
|
|
self.img = self.status_imgs_full["warning"]
|
|
elif results is None or results.get("ok", None) is None:
|
|
self.img = self.status_imgs_full[None]
|
|
elif results.get("ok", None) is False:
|
|
self.img = self.status_imgs_full[False]
|
|
else:
|
|
self.img = self.status_imgs_full[None]
|
|
else:
|
|
self.img = img
|
|
self.resizeEvent()
|
|
|
|
def save_last(self):
|
|
if self.last is None:
|
|
return
|
|
|
|
def resizeEvent(self, event=None):
|
|
if hasattr(self, "img_l"):
|
|
self.img_l.setPixmap(self.img.scaled(self.img_l.width(), self.img_l.height(), Qt.KeepAspectRatio, Qt.SmoothTransformation))
|
|
|
|
def challenge_admin(self, info):
|
|
if not self.admin_challenged:
|
|
d = Dialog(parent=self)
|
|
d.setCentralWidget(Test_Admin_Permission(info))
|
|
d.setModal(True)
|
|
d.show()
|
|
d.centralWidget.password_le.setFocus()
|
|
r = d.exec()
|
|
if r == d.Accepted:
|
|
self.admin_challenged = True
|
|
elif r == d.Rejected:
|
|
self.admin_challenged ^= False
|
|
else:
|
|
raise AssertionError("Bad admin challenge dialog return code")
|
|
return self.admin_challenged
|
|
|
|
def override(self):
|
|
if self.challenge_admin("Si sta tentando di bypassare il test"):
|
|
self.get(override=True)
|
|
|
|
def emit_ok(self):
|
|
self.ok.emit(self.last)
|
|
|
|
def emit_ko(self):
|
|
self.ko.emit(self.last)
|