286 lines
10 KiB
Python
286 lines
10 KiB
Python
#!/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"] |