diff --git a/pyproject.toml b/pyproject.toml index f1ca813..f191b2c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -5,7 +5,7 @@ build-backend = "setuptools.build_meta" [project] name = "common" description = "Commonly reusable code" -version = "0.1.19" +version = "0.1.20" requires-python = ">=3.13" authors = [ { name="BreakerBear", email="breakerbear@autistic.men" }, diff --git a/src/common/actionflow.py b/src/common/actionflow.py index c4cbd4f..efbf204 100644 --- a/src/common/actionflow.py +++ b/src/common/actionflow.py @@ -5,12 +5,12 @@ class NotAllowed(Exception): pass class Action(Exception, metaclass=ABCMeta): def __init__(self, *args): super().__init__(*args) - + @classmethod @abstractmethod def prepare(cls) -> bool: pass - + @classmethod def perform(cls): raise cls() @@ -20,11 +20,11 @@ class ActionFlow: self.indices: dict[str, int] = {} self.actions: list[type[Action]] = [] self.on: list[bool] = [] - + def __iter__(self): for key in self.indices.keys(): yield (key, self[key]) - + def __getitem__(self, key): index = self.indices[key] def inner(*args, **kwargs): @@ -32,61 +32,61 @@ class ActionFlow: if self.on[index] is None: raise NotAllowed(key) self.on[index] = self.actions[index].prepare(*args, **kwargs) return self.on[index] - + return inner - + def queue(self, action: type[Action]) -> bool: try: return self[action.__name__]() except: return False - + def react(self, *stage: type[Action]): for Props in stage: self.allow(Props) self.queue(Props) - + while any(self.on): for index in self.indices.values(): if (self.on[index]): self.on[index] = None self.on[index] = self.actions[index].perform() - + def index(self, action: type[Action]) -> int: name = action.__name__ index = self.indices.get(name) assert index is not None, "Action '%s' is not registered" % name return index - + def allow(self, *actions: type[Action]): for Item in actions: index = self.index(Item) self.on[index] = False - + def deter(self, *actions: type[Action]): for Item in actions: index = self.index(Item) self.on[index] = None - + def stage(self, *preset: type[Action]): for Props in preset: self.append(Props) - + def append(self, action: type[Action]): name = action.__name__ index = len(self.indices) assert name not in self.indices, "Action '%s' is already registered" % name - + self.indices[name] = index self.actions.insert(index, action) self.on.insert(index, None) - + def remove(self, action: type[Action]): name = action.__name__ index = self.index(action) - + self.indices.pop(name) self.actions.pop(index) self.on.pop(index) - + def capabilities(self) -> dict[str, bool]: items = self.indices.items() return { k: not self.on[v] for k, v in items if self.on[v] is not None } diff --git a/src/common/jsonrpc2/server.py b/src/common/jsonrpc2/server.py index cc231c6..3e3e229 100644 --- a/src/common/jsonrpc2/server.py +++ b/src/common/jsonrpc2/server.py @@ -18,20 +18,20 @@ class Request[T]: class ParamsError(Exception): pass class NotFound(Exception): pass class Invalid(Exception): pass - + @classmethod def load(cls, data: str) -> Self: result = json.loads(data) return cls(**result) - + def handle(self, handlers: dict[str, Any]) -> T: args = self.params handler: Callable[..., T] = handlers.get(self.method) - + if handler is None: raise self.NotFound(self.method) argcount = handler.__code__.co_argcount argnames = handler.__code__.co_varnames[:argcount] - + if self.params is None: if argcount > 0: raise self.ParamsError('Too less') return handler() @@ -41,14 +41,14 @@ class Request[T]: if isinstance(args, dict): if args.keys() != set(argnames): raise self.ParamsError(args) return handler(**args) - + raise self.Invalid(type(args)) class Response[T]: def __init__(self, id: str|int|None, inner: T): self.id = id self.inner = inner - + def __str__(self): data = dict() data['id'] = self.id @@ -62,17 +62,17 @@ class Error: def __init__(self, code, data=None): self.code: Error.Code = code self.data = data - + class Code(Enum): PARSE_ERROR = -32700 INVALID_REQUEST = -32600 METHOD_NOT_FOUND = -32601 INVALID_PARAMS = -32602 INTERNAL_ERROR = -32603 - + def message(self) -> str: return self.code.name.capitalize().replace('_', ' ') - + def response(self): result = dict() result['code'] = self.code.value @@ -85,14 +85,14 @@ class History(logging.Handler): def __init__(self): super().__init__() self.records = [] - + def emit(self, record): keys = set(['name', 'levelno', 'levelname', 'msg', 'exc_text', 'stack_info', 'created']) values = vars(record) values['msg'] %= values['args'] result = { key: values[key] for key in keys } self.records.append(result) - + def truncate(self) -> list: copy = self.records.copy() self.records.clear() @@ -108,10 +108,10 @@ class Server(ThreadingHTTPServer): class RequestHandler(BaseHTTPRequestHandler): server: Server protocol_version = 'HTTP/1.1' - + def log_message(self, format, *args): pass - + def do_GET(self): try: match self.headers.get('Sec-Fetch-Dest'): @@ -128,16 +128,16 @@ class RequestHandler(BaseHTTPRequestHandler): except Exception as e: response = b'Internal error: %b' % str(e).encode(self.server.encoding) self.send_response(500) - + self.send_header('Content-Length', str(len(response))) self.end_headers() self.wfile.write(response) - + def do_HEAD(self): self.send_response(200) self.send_header('X-Powered-By', app) if (app := self.server.application) else None self.end_headers() - + def do_POST(self, request=None): try: size = int(self.headers.get('Content-Length', '0')) @@ -154,7 +154,7 @@ class RequestHandler(BaseHTTPRequestHandler): message = Error(Error.Code.METHOD_NOT_FOUND, e) except Exception as e: message = Error(Error.Code.INTERNAL_ERROR, e) - + response = Response(request.id if request is not None else None, message) buffer = bytes(str(response), encoding=self.server.encoding) self.send_response(200) @@ -168,7 +168,7 @@ class ServiceProvider: self.server = Server((self.options['host'], self.options['port']), self.options['handler']) self.server.application = self.options['application'] self.server.encoding = self.options['encoding'] - + if not self.options['setup']: return import sys, _thread as t logger = logging.getLogger() @@ -176,7 +176,7 @@ class ServiceProvider: logger.addHandler(history) self.set('logs', lambda: history.truncate()) self.set('exit', lambda: (t.interrupt_main(), sys.exit(0))) - + @dataclasses.dataclass class Options: host : str = '127.0.0.1' @@ -186,18 +186,18 @@ class ServiceProvider: encoding : str = 'UTF-8' interval : float = 0.2 application : str = '' - + def add(self, *handlers: Callable[..., Any], pairs: Iterable[tuple[str, Callable[..., Any]]] = None): for handler in handlers: self.set(handler.__name__, handler) for k, v in pairs or tuple(): self.set(k, v) - + def set(self, method: str, handler: Callable[..., Any]): if method in self.server.handlers: raise KeyError(method) self.server.handlers[method] = handler - + def pop(self, method: str) -> Callable[..., Any]: return self.server.handlers.pop(method) - + def run(self) -> str: thread = Thread(target=lambda: self.server.serve_forever(self.options['interval']), daemon=True) thread.start() diff --git a/src/common/timer.py b/src/common/timer.py index e7bc1a7..b90a18f 100644 --- a/src/common/timer.py +++ b/src/common/timer.py @@ -4,19 +4,19 @@ from time import time class Timer: def __init__(self): self.clear() - + def clear(self): - self.checkpoint = None self.accumulator = 0 - + self.checkpoint = 0 + def start(self): - seconds = floor(time()) - self.checkpoint = seconds - - def pause(self): - assert self.checkpoint is not None, "Uninitialized" self.accumulator = self.delta() - + self.checkpoint = floor(time()) + + def pause(self): + self.accumulator = self.delta() + self.checkpoint = 0 + def delta(self): - assert self.checkpoint is not None, "Uninitialized" - return floor(time()) - self.checkpoint + self.accumulator + excess = floor(time()) - rhs if (rhs := self.checkpoint) else 0 + return excess + self.accumulator diff --git a/src/common/utils/selenium.py b/src/common/utils/selenium.py index 9291c0d..eb8d774 100644 --- a/src/common/utils/selenium.py +++ b/src/common/utils/selenium.py @@ -23,7 +23,7 @@ def sleep(seconds: float): try: driver.switch_to.alert except NoAlertPresentException: pass except: raise KeyboardInterrupt() - + try: WebDriverWait(driver, seconds, seconds).until(lambda _: False) except TimeoutException: pass @@ -32,15 +32,15 @@ def locate(selector: str, wait=True, condition=True) -> WebElement: try: locator = (By.CSS_SELECTOR, selector) if not wait: return driver.find_element(*locator) - + presence = EC.presence_of_element_located(locator) element = WebDriverWait(driver, parameters.get('timeout', 0)).until(presence, 'Timeout') driver.execute_script("arguments[0].scrollIntoView({ block: 'center', inline: 'nearest' });", element) - + if condition is not None and condition != False: predicate = condition if callable(condition) else EC.visibility_of_element_located element = WebDriverWait(driver, parameters.get('timeout', 0)).until(predicate(locator), 'Timeout') - + return element except StaleElementReferenceException: pass @@ -48,13 +48,13 @@ def locate(selector: str, wait=True, condition=True) -> WebElement: def click(selector: str|WebElement, wait=True, condition=False): predicate = None if condition is None else condition or EC.element_to_be_clickable error = False - + element = locate(selector, wait, predicate) if isinstance(selector, str) else selector counter = lambda: int(element.get_attribute(unique) or '0') value = counter() driver.execute_script("window.__%s__ = () => { arguments[0].setAttribute('%s', arguments[1] + 1) };" % ((unique,) * 2), element, value) driver.execute_script("arguments[0].addEventListener('click', __%s__);" % unique, element) - + for _ in range(parameters.get('attempts', 0)): try: if not error: element.click() @@ -69,7 +69,7 @@ def click(selector: str|WebElement, wait=True, condition=False): break except TimeoutException: continue except: break - + try: driver.execute_script("arguments[0].removeEventListener('click', __%s__);" % unique, element) except: pass