#!/usr/bin/env python # -*- coding: utf-8 -*- from threading import Thread from time import time as date_in_seconds from time import sleep from inspect import getfullargspec as get_full_arguments_properties if "LibreTranslatePlus" not in globals(): class LibreTranslatePlus:pass class Anonymous(LibreTranslatePlus.Abstracts.Base): # LibreTranslatePlus.Threads def __init__(self, ltp, _input = None): super().__init__(ltp) self._print("info", "ltp_threads_building") self.__started = False self.__closed = False self.__threads = [] self.__set_common_variables() self._print("ok", "ltp_threads_built") def start(self, callback = None): self._print("info", "ltp_threads_starting") end = lambda status:callable(callback) and callback(status) if self.__started: self._print("warn", "ltp_threads_already_started") end(False) return False self.__started = True self.__set_common_variables() self._print("ok", "ltp_threads_started") end(True) def __set_common_variables(self): self.__show_add_error_message = self.settings(("threads_show_add_error_message", "show_error_message")) self.__show_add_ok_message = self.settings(("threads_show_add_ok_message", "show_ok_message")) self.__show_add_exception_message = self.settings(("threads_show_add_exception_message", "show_exception_message")) self.__show_remove_error_message = self.settings(("threads_show_remove_error_message", "show_error_message")) self.__show_remove_ok_message = self.settings(("threads_show_remove_ok_message", "show_ok_message")) self.__show_remove_exception_message = self.settings(("threads_show_remove_exception_message", "show_exception_message")) self.__show_removing_all_message = self.settings(("threads_show_removing_all_exception_message", "show_info_message")) self.__show_remove_all_exception_message = self.settings(("threads_show_remove_all_exception_message", "show_exception_message")) self.__show_stop_error_message = self.settings(("threads_show_stop_error_message", "show_error_message")) self.__show_stop_ok_message = self.settings(("threads_show_stop_ok_message", "show_ok_message")) self.__show_stop_exception_message = self.settings(("threads_show_stop_exception_message", "show_exception_message")) self.__show_play_error_message = self.settings(("threads_show_play_error_message", "show_error_message")) self.__show_play_ok_message = self.settings(("threads_show_play_ok_message", "show_ok_message")) self.__show_play_exception_message = self.settings(("threads_show_play_exception_message", "show_exception_message")) def close(self, callback = None): self._print("info", "ltp_threads_closing") end = lambda status:callable(callback) and callback(status) if self.__closed: end(False) self._print("warn", "ltp_threads_already_closed") return False self.__closed = True self.remove_all() self._print("ok", "ltp_threads_closed") end(True) return True def __execute(self, i): while self.__threads[i] != None: if self.__threads[i]["working"]: date = date_in_seconds() * 1000 if date - self.__threads[i]["date"] >= self.__threads[i]["timer"]: self.__threads[i]["date"] = date self.__threads[i]["callback"](i) # not i and print(self.__threads[i]) if not self.__threads[i] or not self.__threads[i]["bucle"]: self.__threads[i] = None break sleep(self.__threads[i]["subtimer"] / 1000.0) def add(self, callback, _input = None): error = None try: error = ( (( 1 << 0 if callback == None else 1 << 1 if not callable(callback) else 1 << 2 if len([key for i, key in enumerate(get_full_arguments_properties(callback)[0]) if i or key != "self"]) != 1 else 0) << 0) | (( 0 if _input == None else 1 << 1 if not isinstance(_input, (list, tuple, dict, bool, int, float)) else 0) << 3) | 0) << 1 if self.validate( error, ( "exception", "callback_null", "callback_not_function", "callback_need_one_parameter", "input_null", "input_bad_type" ), {}, self.__show_add_error_message and "ltp_threads_add_error", self.__show_add_ok_message and "ltp_threads_add_ok" ): if isinstance(_input, bool): _input = {"autostart" : _input} elif isinstance(_input, (int, float)): _input = {"timer" : _input} elif _input != None and not isinstance(_input, (list, tuple, dict)): _input = None i = None l = len(self.__threads) thread = { "working" : self.settings(("threads_autostart", "autostart"), _input), "callback" : callback, "timer" : self.settings(("threads_timer", "timer"), _input), "subtimer" : self.settings(("threads_subtimer", "subtimer"), _input), "date" : 0 if self.settings(("threads_start_now", "start_now"), _input) else date_in_seconds() * 1000, "bucle" : self.settings(("threads_bucle", "threads_repeat", "bucle", "repeat"), _input) } for j, subthread in enumerate(self.__threads): if subthread == None: i = j break if i == None: i = len(self.__threads) thread["i"] = i thread["thread"] = Thread(target = lambda:self.__execute(i)) if i == l: self.__threads += [thread] else: self.__threads[i] = thread thread["thread"].start() except Exception as exception: error |= 1 << 0 self.exception(exception, self.__show_add_exception_message and "ltp_threads_add_exception") return error def __validate_i(self, i): return ( 1 << 0 if i == None else 1 << 1 if not isinstance(i, int) else 1 << 2 if i < 0 else 1 << 3 if i >= len(self.__threads) else 1 << 4 if self.__threads[i] == None else 0) def remove(self, i): try: error = self.__validate_i(i) << 1 if self.validate( error, ( "exception", "i_null", "i_not_integer", "i_lower_0", "i_too_high", "thread_null" ), {"i" : i}, self.__show_remove_error_message and "ltp_threads_remove_error", self.__show_remove_ok_message and "ltp_threads_remove_ok" ): self.__threads[i] = None except Exception as exception: error |= 1 << 0 self.exception(exception, self.__show_remove_exception_message and "ltp_threads_remove_exception", {"i" : i}) return error def remove_all(self): self.__show_removing_all_message and self._print("info", "ltp_threads_removing_all") error = None try: for i in range(len(self.__threads)): if self.__threads[i] != None: self.__threads[i] = None except Exception as exception: error |= 1 << 0 self.exception(exception, self.__show_remove_all_exception_message and "ltp_threads_remove_all_exception") return error def stop(self, i): error = None try: error = (self.__validate_i(i) or (0 if self.__threads[i]["working"] else 1 << 5)) << 1 if self.validate( error, ( "exception", "i_null", "i_not_integer", "i_lower_0", "i_too_high", "thread_null", "thread_not_working" ), {"i" : i}, self.__show_stop_error_message and "ltp_threads_stop_error", self.__show_stop_ok_message and "ltp_threads_stop_ok" ): self.__threads[i]["working"] = False except Exception as exception: error |= 1 << 0 self.exception(exception, self.__show_stop_exception_message and "ltp_threads_stop_exception", {"i" : i}) return error def play(self, i): error = None try: error = (self.__validate_i(i) or (1 << 5 if self.__threads[i]["working"] else 0)) << 1 if self.validate( error, ( "exception", "i_null", "i_not_integer", "i_lower_0", "i_too_high", "thread_null", "thread_working" ), {"i" : i}, self.__show_play_error_message and "ltp_threads_play_error", self.__show_play_ok_message and "ltp_threads_play_ok" ): self.__threads[i]["working"] = True except Exception as exception: error |= 1 << 0 self.exception(exception, self.__show_play_exception_message and "ltp_threads_play_exception", {"i" : i}) return error def get(self, i, show_errors = None): return self.__threads[i] if i != None and isinstance(i, int) and i >= 0 and i < len(self.__threads) else None LibreTranslatePlus.Threads = Anonymous del globals()["Anonymous"]