mirror of
https://github.com/LmeSzinc/StarRailCopilot.git
synced 2024-11-16 14:31:16 +00:00
197 lines
5.1 KiB
Python
197 lines
5.1 KiB
Python
|
import random
|
||
|
import re
|
||
|
from functools import wraps
|
||
|
from typing import Callable, Generic, TypeVar
|
||
|
|
||
|
T = TypeVar("T")
|
||
|
|
||
|
|
||
|
class Config:
|
||
|
"""
|
||
|
Decorator that calls different function with a same name according to config.
|
||
|
|
||
|
func_list likes:
|
||
|
func_list = {
|
||
|
'func1': [
|
||
|
{'options': {'ENABLE': True}, 'func': 1},
|
||
|
{'options': {'ENABLE': False}, 'func': 1}
|
||
|
]
|
||
|
}
|
||
|
"""
|
||
|
func_list = {}
|
||
|
|
||
|
@classmethod
|
||
|
def when(cls, **kwargs):
|
||
|
"""
|
||
|
Args:
|
||
|
**kwargs: Any option in AzurLaneConfig.
|
||
|
|
||
|
Examples:
|
||
|
@Config.when(USE_ONE_CLICK_RETIREMENT=True)
|
||
|
def retire_ships(self, amount=None, rarity=None):
|
||
|
pass
|
||
|
|
||
|
@Config.when(USE_ONE_CLICK_RETIREMENT=False)
|
||
|
def retire_ships(self, amount=None, rarity=None):
|
||
|
pass
|
||
|
"""
|
||
|
from module.logger import logger
|
||
|
options = kwargs
|
||
|
|
||
|
def decorate(func):
|
||
|
name = func.__name__
|
||
|
data = {'options': options, 'func': func}
|
||
|
if name not in cls.func_list:
|
||
|
cls.func_list[name] = [data]
|
||
|
else:
|
||
|
override = False
|
||
|
for record in cls.func_list[name]:
|
||
|
if record['options'] == data['options']:
|
||
|
record['func'] = data['func']
|
||
|
override = True
|
||
|
if not override:
|
||
|
cls.func_list[name].append(data)
|
||
|
|
||
|
@wraps(func)
|
||
|
def wrapper(self, *args, **kwargs):
|
||
|
"""
|
||
|
Args:
|
||
|
self: ModuleBase instance.
|
||
|
*args:
|
||
|
**kwargs:
|
||
|
"""
|
||
|
for record in cls.func_list[name]:
|
||
|
|
||
|
flag = [value is None or self.config.__getattribute__(key) == value
|
||
|
for key, value in record['options'].items()]
|
||
|
if not all(flag):
|
||
|
continue
|
||
|
|
||
|
return record['func'](self, *args, **kwargs)
|
||
|
|
||
|
logger.warning(f'No option fits for {name}, using the last define func.')
|
||
|
return func(self, *args, **kwargs)
|
||
|
|
||
|
return wrapper
|
||
|
|
||
|
return decorate
|
||
|
|
||
|
|
||
|
class cached_property(Generic[T]):
|
||
|
"""
|
||
|
cached-property from https://github.com/pydanny/cached-property
|
||
|
Add typing support
|
||
|
|
||
|
A property that is only computed once per instance and then replaces itself
|
||
|
with an ordinary attribute. Deleting the attribute resets the property.
|
||
|
Source: https://github.com/bottlepy/bottle/commit/fa7733e075da0d790d809aa3d2f53071897e6f76
|
||
|
"""
|
||
|
|
||
|
def __init__(self, func: Callable[..., T]):
|
||
|
self.func = func
|
||
|
|
||
|
def __get__(self, obj, cls) -> T:
|
||
|
if obj is None:
|
||
|
return self
|
||
|
|
||
|
value = obj.__dict__[self.func.__name__] = self.func(obj)
|
||
|
return value
|
||
|
|
||
|
|
||
|
def del_cached_property(obj, name):
|
||
|
"""
|
||
|
Delete a cached property safely.
|
||
|
|
||
|
Args:
|
||
|
obj:
|
||
|
name (str):
|
||
|
"""
|
||
|
try:
|
||
|
del obj.__dict__[name]
|
||
|
except KeyError:
|
||
|
pass
|
||
|
|
||
|
|
||
|
def has_cached_property(obj, name):
|
||
|
"""
|
||
|
Check if a property is cached.
|
||
|
|
||
|
Args:
|
||
|
obj:
|
||
|
name (str):
|
||
|
"""
|
||
|
return name in obj.__dict__
|
||
|
|
||
|
|
||
|
def function_drop(rate=0.5, default=None):
|
||
|
"""
|
||
|
Drop function calls to simulate random emulator stuck, for testing purpose.
|
||
|
|
||
|
Args:
|
||
|
rate (float): 0 to 1. Drop rate.
|
||
|
default: Default value to return if dropped.
|
||
|
|
||
|
Examples:
|
||
|
@function_drop(0.3)
|
||
|
def click(self, button, record_check=True):
|
||
|
pass
|
||
|
|
||
|
30% possibility:
|
||
|
INFO | Dropped: module.device.device.Device.click(REWARD_GOTO_MAIN, record_check=True)
|
||
|
70% possibility:
|
||
|
INFO | Click (1091, 628) @ REWARD_GOTO_MAIN
|
||
|
"""
|
||
|
from module.logger import logger
|
||
|
|
||
|
def decorate(func):
|
||
|
@wraps(func)
|
||
|
def wrapper(*args, **kwargs):
|
||
|
if random.uniform(0, 1) > rate:
|
||
|
return func(*args, **kwargs)
|
||
|
else:
|
||
|
cls = ''
|
||
|
arguments = [str(arg) for arg in args]
|
||
|
if len(arguments):
|
||
|
matched = re.search('<(.*?) object at', arguments[0])
|
||
|
if matched:
|
||
|
cls = matched.group(1) + '.'
|
||
|
arguments.pop(0)
|
||
|
arguments += [f'{k}={v}' for k, v in kwargs.items()]
|
||
|
arguments = ', '.join(arguments)
|
||
|
logger.info(f'Dropped: {cls}{func.__name__}({arguments})')
|
||
|
return default
|
||
|
|
||
|
return wrapper
|
||
|
|
||
|
return decorate
|
||
|
|
||
|
|
||
|
def run_once(f):
|
||
|
"""
|
||
|
Run a function only once, no matter how many times it has been called.
|
||
|
|
||
|
Examples:
|
||
|
@run_once
|
||
|
def my_function(foo, bar):
|
||
|
return foo + bar
|
||
|
|
||
|
while 1:
|
||
|
my_function()
|
||
|
|
||
|
Examples:
|
||
|
def my_function(foo, bar):
|
||
|
return foo + bar
|
||
|
|
||
|
action = run_once(my_function)
|
||
|
while 1:
|
||
|
action()
|
||
|
"""
|
||
|
|
||
|
def wrapper(*args, **kwargs):
|
||
|
if not wrapper.has_run:
|
||
|
wrapper.has_run = True
|
||
|
return f(*args, **kwargs)
|
||
|
|
||
|
wrapper.has_run = False
|
||
|
return wrapper
|