Utilities/common
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: 'Timer' logic

Browse Source
This commit is contained in:
Break27
2026-08-11 20:59:21 +08:00
parent 0b359ee4e2
commit bfb45a971e
5 changed files with 59 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pyproject.toml
+1 -1
View File
@@ -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" },
src/common/actionflow.py
+17 -17
View File
@@ -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 }
src/common/jsonrpc2/server.py
+23 -23
View File
@@ -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()
src/common/timer.py
+11 -11
View File
@@ -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
src/common/utils/selenium.py
+7 -7
View File
@@ -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