Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- #!/usr/bin/env python
- # coding: utf-8
- import time
- import random
- from selenium import webdriver
- import threading
- import sys
- from PyQt5 import QtCore, QtWidgets
- from gui import Ui_MainWindow
- class HTaskLib:
- def __init__(self) -> None:
- self._driver = ""
- self._options = ""
- @HTask
- def browser_start(self):
- self._options = webdriver.ChromeOptions()
- self._driver = webdriver.Chrome(
- options=self._options)
- self._driver.set_window_size(1000, 660)
- @HTask
- def open(self):
- self._driver.get("https://www.google.fr")
- @HTask
- def print_timer(self):
- for i in range(10):
- self.wait_flag()
- print(time.time())
- # self.pause()
- time.sleep(0.5)
- self.done()
- class HTask(threading.Thread):
- def __init__(self, func, *args, **kwargs):
- super(HTask, self).__init__(*args, **kwargs)
- self.__flag = threading.Event() # The flag used to pause the thread
- self.__flag.set() # Set to True
- self.__running = threading.Event() # Used to stop the thread identification
- self.__running.set() # Set running to True
- self._func = func
- self.__done = threading.Event()
- def run(self):
- while self.__running.isSet():
- # return immediately when it is True, block until the internal flag is True when it is False
- self.__flag.wait()
- self._func(self)
- def pause(self):
- print('[+] Paused')
- self.clear_flag() # Set to False to block the thread
- def resume(self):
- print('[+] Resumed')
- self.set_flag() # Set to True, let the thread stop blocking
- def stop(self):
- print('[+] Stoped')
- self.set_flag() # Resume the thread from the suspended state, if it is already suspended
- self.clear_running() # Set to False
- def done(self):
- print('[+] Task done')
- self.__flag.set()
- self.__done.set()
- self.__running.clear()
- def set_flag(self):
- self.__flag.set()
- def clear_flag(self):
- self.__flag.clear()
- def wait_flag(self):
- self.__flag.wait()
- def is_flagged(self):
- return self.__flag.is_set()
- def clear_running(self):
- self.__running.clear()
- def wait_running(self):
- self.__running.wait()
- def is_running(self):
- return self.__running.is_set()
- def is_done(self):
- return self.__done.is_set()
- class ApplicationWindow(QtWidgets.QMainWindow):
- def __init__(self) -> None:
- super(ApplicationWindow, self).__init__()
- self.ui = Ui_MainWindow()
- self.ui.setupUi(self)
- def main():
- _translate = QtCore.QCoreApplication.translate
- app = QtWidgets.QApplication(sys.argv)
- application = ApplicationWindow()
- application.show()
- def play_pause_task(task: HTask):
- if task.is_flagged():
- application.ui.play_pause_button.setText(
- _translate("MainWindow", "Play"))
- task.pause()
- else:
- application.ui.play_pause_button.setText(
- _translate("MainWindow", "Pause"))
- task.resume()
- def stop_task(task: HTask, app: QtWidgets.QApplication):
- app.close()
- task.stop()
- # TODO : If there is still time, implement shorcut keyboard for better UX
- htl = HTaskList()
- task_todo = []
- task_done = []
- task_todo.append(htl.browser_start)
- # task_todo.append(htl.print_hello)
- task_todo.append(htl.open)
- for id, task in enumerate(task_todo):
- print('Task ID' + f"{id+1:03}")
- application.ui.play_pause_button.clicked.connect(
- lambda: play_pause_task(task))
- application.ui.stop_button.clicked.connect(
- lambda: stop_task(task, application))
- task.start()
- if task.is_done():
- task_done.append(task_todo[task])
- sys.exit(app.exec_())
- if __name__ == '__main__':
- main()
Add Comment
Please, Sign In to add comment