DR_Team/Difficult-Rocket
DR_Team/Difficult-Rocket
6
2
Fork 0
Code Issues Pull Requests Projects Releases 25 Wiki Activity

more update

Browse Source
This commit is contained in:
shenjack 2022-08-12 21:07:36 +08:00
parent 6543989c3a
commit 7b8dbabc9b
13 changed files with 263 additions and 135 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
DR.py
View File

@ -11,7 +11,6 @@ import traceback
# TODO 默认位置配置文件
# TODO 可自定义工作路径
hi = """Difficult Rocket is writen by shenjackyuanjie
mail: 3695888@qq.com or shyj3695888@163.com
QQ: 3695888"""
@ -24,6 +23,8 @@ error_format = {
}
if __name__ == '__main__':
start_time_ns = time.time_ns()
start_time_perf_ns = time.perf_counter_ns()
print(f'{__file__=}')
print(f'{sys.path[0]=}')
print(f'{sys.argv[0]=}')
@ -39,28 +40,30 @@ if __name__ == '__main__':
print(sys.path) # 输出路径
print(hi) # hi！
DEBUGGING = False # 是否在 DEBUG
from Difficult_Rocket.exception import TestError
from Difficult_Rocket.crash import crash
from Difficult_Rocket import DR_option
try:
start_time = time.perf_counter_ns() # 记录启动时间
import pyglet # 导入pyglet
pyglet.resource.path = ['/textures/']
pyglet.resource.reindex()
from Difficult_Rocket import main
from Difficult_Rocket import main, DR_runtime
DR_runtime.start_time_ns = start_time_ns
from libs.pyglet.gl import glClearColor # 调整背景颜色
glClearColor(0.5, 0.5, 0.5, 0)
game = main.Game() # 实例化一个游戏
print(time.perf_counter_ns() - start_time) # 输出一下启动用时
print(time.perf_counter_ns() - start_time_perf_ns, (time.perf_counter_ns() - start_time_perf_ns) / (10 ** 9), 'start') # 输出一下启动用时
cprofile = False # 是否使用cprofile
if cprofile:
cProfile.run('game.start()', sort='calls') # 使用 cprofile 启动
else:
game.start() # 直接启动
if DEBUGGING:
if DR_option.crash_report_test:
raise TestError('debugging') # debug 嘛，试试crash
except Exception as exp: # 出毛病了
print(error_format['error.happen']) #

86
Difficult_Rocket/__init__.py
View File

@ -11,33 +11,87 @@ github: @shenjackyuanjie
gitee: @shenjackyuanjie
"""
from typing import Any
from typing import Any, Dict, Callable, Union
from libs.MCDR.version import Version
game_version = Version("0.6.2")
__version__ = game_version
DR_options = {
'InputBox_use_TextEntry': False,
'playing': False,
'debugging': False
}
_DR_options_type = {
'InputBox_use_TextEntry': bool,
'playing': bool,
'debugging': bool
}
class Options:
"""
Difficult Rocket 的游戏配置的存储基类
"""
_options: Dict[str, Union[Callable, object]] = {}
def option(self) -> dict:
values = {}
for option, a_fun in self._options.items():
if isinstance(a_fun, Callable):
values[option] = a_fun() # 例子: {'language': self.language}
else:
values[option] = a_fun
return {**values, **self.__dict__}
@classmethod
def add_option(cls, name, value: Union[Callable, object]) -> Dict:
cls._options[name] = value
return cls._options
def DR_option_type(config_name: str):
if config_name in _DR_options_type:
return _DR_options_type[config_name]
return Any
class _DR_option(Options):
"""
DR 的整体配置存储类
"""
# runtime options
InputBox_use_TextEntry = False
use_local_logging = False
record_threads = True
# tests
playing = False
debugging = False
crash_report_test = True
if DR_options['playing']:
class _DR_runtime(Options):
"""
DR 的运行时配置
"""
# run status
start_time_ns: int = None
client_setup_time_ns: int = None
server_setup_time_ns: int = None
# game options
_language = 'zh-CN'
@property
def language(self):
return self._language
@language.setter
def language(self, value: str):
if value == self._language:
return
assert isinstance(value, str), "DR language MUST be string"
self._language = value
from Difficult_Rocket.utils import translate
translate.tr._update_lang()
_DR_runtime.add_option('language', _DR_runtime.language)
DR_option = _DR_option()
DR_runtime = _DR_runtime()
print(_DR_runtime.__dict__)
if DR_option.playing:
from .utils import new_thread

7
Difficult_Rocket/client/__init__.py
View File

@ -20,6 +20,7 @@ import traceback
from decimal import Decimal
# Difficult_Rocket function
from Difficult_Rocket import Options
from Difficult_Rocket.command import line, tree
from Difficult_Rocket.utils import new_thread
from Difficult_Rocket.utils.translate import tr
@ -76,6 +77,10 @@ def pyglet_load_fonts_folder(folder) -> None:
pyglet_load_fonts_folder(os.path.join(folder, obj))
class _DR_Client_option(Options):
...
class ClientWindow(Window):
def __init__(self, net_mode='local', *args, **kwargs):
@ -93,8 +98,6 @@ class ClientWindow(Window):
self.net_mode = net_mode
self.run_input = False
# configs
pyglet.resource.path = ['/textures/']
pyglet.resource.reindex()
self.set_icon(pyglet.image.load('./textures/icon.png'))
self.main_config = tools.load_file('./configs/main.toml')
self.game_config = tools.load_file('./configs/game.config')

28
Difficult_Rocket/crash/crash.py
View File

@ -12,6 +12,7 @@ gitee: @shenjackyuanjie
"""
import os
import sys
import time
import platform
import traceback
@ -29,25 +30,26 @@ from typing import Optional
import Difficult_Rocket
Head_message = """# ----- Difficult Rocket Crash Report -----
## Time: {now_time}
## Traceback
"""
Run_message = """## Difficult Rocket running status\n"""
Run_message = """\n## Difficult Rocket running status\n"""
DR_configs = """### game config"""
DR_configs = """\n### game config\n"""
Process_message = """## Process info\n"""
Process_message = """\n## Process info\n"""
Thread_message = """## Thread info\n"""
Thread_message = """\n## Thread info\n"""
Python_message = """## Python info\n"""
Python_message = """\n## Python info\n"""
System_message = """## System info\n"""
System_message = """\n## System info\n"""
all_thread = [threading.main_thread()]
all_process = [multiprocessing.current_process()]
record_thread = True
def crash_info_handler(info: str = None) -> str:
@ -69,7 +71,7 @@ def to_code(string: str):
return f'`{string}`'
def create_crash_report(info: str = None) -> None:
def create_crash_report(info: str = None) -> None:
crash_info = crash_info_handler(info)
if 'crash_report' not in os.listdir('./'):
os.mkdir('./crash_report')
@ -81,17 +83,17 @@ def create_crash_report(info: str = None) -> None:
# 崩溃信息
crash_file.write(crash_info)
# 运行状态信息
from Difficult_Rocket import DR_option, DR_runtime
crash_file.write(Run_message)
crash_file.write(markdown_line_handler(f'DR Version: {Difficult_Rocket.game_version}', level=1))
crash_file.write(markdown_line_handler(f'DR language: {DR_option.language}', level=1))
crash_file.write(markdown_line_handler(f'Running Dir: {os.path.abspath(os.curdir)}', level=1))
crash_file.write(markdown_line_handler(f': {os.name=}', level=1))
crash_file.write(markdown_line_handler(f'DR Version: {str(Difficult_Rocket.Version)}', level=1))
crash_file.write(markdown_line_handler(f'DR Version: {str(Difficult_Rocket.Version)}', level=1))
# # DR 的游戏设置
crash_file.write(DR_configs)
for key, value in Difficult_Rocket.DR_options.items():
crash_file.write(markdown_line_handler(f'Option: {to_code(key)} Type: {to_code(Difficult_Rocket.DR_option_type(key))}', level=1))
for key, value in Difficult_Rocket.DR_option.option().items():
crash_file.write(markdown_line_handler(f'Option: {to_code(key)} Type: {to_code(type(key))}', level=1))
crash_file.write(markdown_line_handler(f'Value: {to_code(value)}', level=2))
print(DR_option.option(), DR_runtime.option(), DR_option.__dict__)
# 多进程信息
crash_file.write(Process_message)
for process in all_process:

6
Difficult_Rocket/guis/widgets.py
View File

@ -13,7 +13,7 @@ gitee: @shenjackyuanjie
from utils import translate
from Difficult_Rocket.guis.format import html
from Difficult_Rocket import DR_options
from Difficult_Rocket import DR_option
# from libs import pyglet
from libs.pyglet import font
@ -51,7 +51,7 @@ class Parts(widgets.WidgetBase):
self._value = 0
if not DR_options['InputBox_use_TextEntry']:
if not DR_option.InputBox_use_TextEntry:
class InputBox(widgets.WidgetBase):
"""
input box
@ -86,7 +86,7 @@ if not DR_options['InputBox_use_TextEntry']:
dpi=font_dpi)
self.font_height = self.font.ascent - self.font.descent
self.out_bound = out_line
if DR_options['InputBox_use_TextEntry']:
if DR_option.InputBox_use_TextEntry:
# 基于IncrementalTextLayout的处理系统
self._doc = FormattedDocument(message)
# self._doc.set_style()

5
Difficult_Rocket/main.py
View File

@ -22,10 +22,11 @@ if __name__ == '__main__': # been start will not run this
sys.path.append('/bin/libs')
sys.path.append('/bin')
from Difficult_Rocket import client, server
from Difficult_Rocket import client, server, DR_option
from Difficult_Rocket.utils import tools
from Difficult_Rocket.utils.translate import tr
class Game:
def __init__(self):
# basic config
@ -34,7 +35,7 @@ class Game:
self.start_time = time.strftime('%Y-%m-%d %H-%M-%S', time.gmtime(time.time()))
# lang_config
self.language = tools.load_file('configs/main.config', 'runtime')['language']
tr.set_language(self.language)
DR_option.language = self.language
# logging config
log_config = tools.load_file('configs/logger.toml')
file_name = log_config['handlers']['file']['filename']

2
Difficult_Rocket/server/__init__.py
View File

@ -30,8 +30,6 @@ from Difficult_Rocket.utils.new_thread import new_thread
class Server:
def __init__(self, net_mode='local', Dev: Delivery = Delivery):
# father class __init__()
# mp.Process.__init__(self)
# logging
self.logger = logging.getLogger('server')
# value

4
Difficult_Rocket/utils/new_thread.py
View File

@ -7,7 +7,7 @@
import functools
import inspect
import threading
from Difficult_Rocket import crash
from Difficult_Rocket import crash, DR_option
from typing import Optional, Callable
"""
@ -76,7 +76,7 @@ def new_thread(thread_name: Optional[str or Callable] = None,
thread = FunctionThread(target=func, args=args, kwargs=kwargs, name=thread_name)
thread.daemon = daemon
thread.start()
if log_thread:
if log_thread and DR_option.record_threads:
crash.all_thread.append(thread)
return thread

193
libs/msdnicrosoft_logger/shenjack.py → Difficult_Rocket/utils/shenjack.py
View File

@ -2,12 +2,17 @@
@author shenjackyuanjie
@contact 3695888@qq.com
"""
import os
# if __name__ == "__main__":
# os.chdir('../')
import atexit
import threading
from os import PathLike
from time import strftime
from typing import Optional, Union, Dict, Iterable, Tuple, List, Callable
from logging import NOTSET, DEBUG, INFO, WARNING, ERROR, FATAL
from typing import Optional, Union, Dict, Iterable, Tuple, List, Callable
from Difficult_Rocket.utils.thread import ThreadLock
@ -42,11 +47,87 @@ ALL = NOTSET
TRACE = 5
FINE = 7
level_name_map = {
ALL: 'ALL', # NOTSET
TRACE: 'TRACE',
FINE: 'FINE',
DEBUG: 'DEBUG',
INFO: 'INFO',
WARNING: 'WARNING', # WARN
ERROR: 'ERROR',
FATAL: 'FATAL'
}
name_level_map = {
'NOTSET': ALL,
'ALL': ALL,
'TRACE': TRACE,
'FINE': FINE,
'DEBUG': DEBUG,
'INFO': INFO,
'WARNING': WARNING,
'WARN': WARNING,
'ERROR': ERROR,
'CRITICAL': FATAL,
'FATAL': FATAL
}
class ListCache:
"""一个线程安全的列表缓存"""
def __init__(self, lock: ThreadLock):
self._cache = []
self.with_thread_lock = lock
def append(self, value: Union[str, Iterable]):
if isinstance(value, str):
with self.with_thread_lock:
self._cache.append(value)
elif isinstance(value, Iterable):
with self.with_thread_lock:
self._cache.append(*value)
else:
raise TypeError(f"cache must be string or Iterable. not a {type(value)}")
def __getitem__(self, item):
assert isinstance(item, int)
with self.with_thread_lock:
try:
return self._cache[item]
except IndexError as exp:
print(f'cache:{self.cache}')
raise IndexError(f'there is no cache at {item}!\ncache:{self.cache}')
def __call__(self, *args, **kwargs):
return self.cache
def __iter__(self):
with self.with_thread_lock:
self._iter_cache = self._cache.copy()
self._iter_len = len(self.cache)
return self
def __next__(self):
if self._iter_cache == -1:
raise StopIteration
returns = self._iter_cache[-self._iter_len]
self._iter_cache -= 1
return returns
def __bool__(self):
with self.with_thread_lock:
return True if len(self.cache) > 0 else False
@property
def cache(self):
return self._cache
class LogFileCache:
"""日志文件缓存"""
def __init__(self, file_name: str = 'logs//log.log', flush_time: Optional[Union[int, float]] = 1, log_cache_lens_max: int = 10):
def __init__(self, file_name: PathLike = 'logs//log.log', flush_time: Optional[Union[int, float]] = 1, log_cache_lens_max: int = 10):
"""
@param file_name: 日志文件名称
@ -57,89 +138,74 @@ class LogFileCache:
self._logfile_name = file_name # log 文件名称
self.flush_time = flush_time # 缓存刷新时长
self.cache_entries_num = log_cache_lens_max
# 日志缓存表
self._log_cache = []
self.started = False
self.log
# 同步锁
self.cache_lock = threading.Lock() # 主锁
self.write_lock = threading.Lock() # 写入锁
self.with_thread_lock = ThreadLock(self.cache_lock, time_out=1 / 60) # 直接用于 with 的主锁
self.threaded_write = threading.Timer(1, self._log_file_time_write) # 基于 timer 的多线程
# 日志缓存表
self.log_cache = ListCache(self.with_thread_lock)
def end_thread(self) -> None:
"""结束日志写入进程，顺手把目前的缓存写入"""
self.cache_lock.acquire(blocking=True)
self.write_lock.acquire(blocking=True)
self.threaded_write.cancel()
self.started = False
self._log_file_time_write()
atexit.unregister(self.end_thread)
def start_thread(self) -> None:
self.threaded_write.start()
self.started = True
atexit.register(self.end_thread)
@property
def logfile_name(self) -> str:
def logfile_name(self) -> PathLike:
return self._logfile_name
@logfile_name.setter
def logfile_name(self, value: str) -> None:
def logfile_name(self, value: PathLike) -> None:
with self.with_thread_lock:
self._logfile_name = value
@property
def log_caches(self) -> list:
return self._log_cache
@log_caches.setter
def log_caches(self, value: Union[str, Iterable[str]]):
if type(value) == str:
with self.with_thread_lock:
self._log_cache.append(value)
return
elif isinstance(value, Iterable):
with self.with_thread_lock:
list(map(self._log_cache.append, value))
...
def _log_file_time_write(self) -> None:
"""使用 threading.Timer 调用的定时写入日志文件的函数"""
if self.log_caches:
if self.log_cache:
with self.with_thread_lock:
if self.log_caches:
...
...
if self.log_cache:
with open(file=self.logfile_name, encoding='utf-8', mode='a') as log_file:
log_file.writelines(self.log_cache)
def write_logs(self, string: str, wait_for_cache: bool = True) -> None:
if wait_for_cache:
if not wait_for_cache:
with self.with_thread_lock and open(file=self.logfile_name, encoding='utf-8', mode='a') as log_file:
log_file.writelines(self._log_cache)
if self.log_cache:
log_file.writelines(self.log_cache)
log_file.write(string)
...
else:
...
self.log_cache.append(string)
class Logger:
"""shenjack logger"""
def __init__(self, **kwargs) -> None:
def __init__(self, name: str = None, level: int = None, file_name: PathLike = None, **kwargs) -> None:
"""
配置模式: 使用 kwargs 配置
@param config: 字典格式的配置
@param kwargs: key word 格式的配置
@param name: logger 名称 默认为 root
@param level: logging 输出等级 默认为 DEBUG(10)
@param file_name: logging 写入文件名称 默认为 None(不写入)
"""
self.name = 'root'
self.level = DEBUG
self.name = name or 'root'
self.level = level or DEBUG
self.colors = None
if kwargs is not None: # 使用 kwargs 尝试配置
if name := kwargs.pop('name', False): # 顺手把获取到的配置填入临时变量 如果成功获取再填入 self
self.name = name
if level := kwargs.pop('level', False):
self.level = level
if file_name:
self.file_cache = LogFileCache(file_name=file_name)
else:
...
self.file_cache = LogFileCache()
self.file_cache = False
self.warn = self.warning
def make_log(self, *values: object,
@ -149,10 +215,14 @@ class Logger:
flush: Optional[bool] = False) -> None:
if level < self.level:
return None
print(level, values, sep, end, flush, sep='|')
# print(level, values, sep, end, flush, sep='|')
write_text = sep.join(i if type(i) is str else str(i) for i in values).__add__(end)
print(write_text, end='')
...
# self.file_cache.write_logs()
if self.file_cache:
self.file_cache: LogFileCache
if not self.file_cache.started:
self.file_cache.start_thread()
def trace(self, *values: object,
sep: Optional[str] = ' ',
@ -221,20 +291,20 @@ logger_configs = {
'Logger': {
'root': {
'level': TRACE,
'color': {
DEBUG: '\033[0m'
},
'color': 'main_color',
'file': 'main_log_file',
},
},
'Color': {
TRACE: {'info': '\033[34;40m', 'message': '\033[48;2;40;40;40m'},
FINE: {'info': '', 'message': '\033[35m'},
DEBUG: {'info': '', 'message': '\033[38;2;133;138;149m'},
INFO: {'info': '\033[32;40m', 'message': ''},
WARNING: {'info': '', 'message': '\033[33m'},
ERROR: {'info': '', 'message': '\033[31m'},
FATAL: {'info': '', 'message': '\033[33;41'}
'main_color': {
TRACE: {'info': '\033[34;40m', 'message': '\033[48;2;40;40;40m'},
FINE: {'info': '', 'message': '\033[35m'},
DEBUG: {'info': '', 'message': '\033[38;2;133;138;149m'},
INFO: {'info': '\033[32;40m', 'message': ''},
WARNING: {'info': '', 'mes sage': '\033[33m'},
ERROR: {'info': '', 'message': '\033[31m'},
FATAL: {'info': '', 'message': '\033[33;41'}
}
},
'File': {
'main_log_file': {
@ -287,8 +357,17 @@ def get_logger(name: str = 'name') -> Logger:
if __name__ == "__main__":
import os
os.chdir('../../')
# 在这里可以使用 add_kwargs_to_global
some_logger = Logger(name='aaa')
some_logger.level = ALL
some_logger.warn('aaaa', 'aaaa')
...
a_lock = threading.Lock()
a_with_lock = ThreadLock(a_lock)
a_cache = ListCache(a_with_lock)
a_cache.append('123123')
print(a_cache[0])
print(a_cache)

6
Difficult_Rocket/utils/tools.py
View File

@ -36,12 +36,10 @@ file_error = {FileNotFoundError: 'no {filetype} file was founded!:\n file name:
Exception: 'get some {error_type} error when read {filetype} file {filename}! \n file type: {} \n file name: {} \n stack: {stack}'}
def load_file(file_name: str, stack:Union[str, list, dict] = None, raise_error: bool = True) -> Union[dict, list]:
def load_file(file_name: str, stack: Union[str, list, dict] = None, raise_error: bool = True) -> Union[dict, list]:
f_type = file_name[file_name.rfind('.') + 1:] # 从最后一个.到末尾 (截取文件格式)
get_file = NotImplementedError('解析失败，请检查文件类型/文件内容/文件是否存在！')
try:
get_file = NotImplementedError('解析失败，请检查文件类型/文件内容/文件是否存在！')
if f_type == 'xml':
xml_load = parse(file_name)
if stack is not None:

42
Difficult_Rocket/utils/translate.py
View File

@ -16,6 +16,7 @@ import objprint
from typing import Union
from Difficult_Rocket import DR_runtime, DR_option
from Difficult_Rocket.utils import tools
from Difficult_Rocket.exception.language import *
@ -60,14 +61,13 @@ class Lang:
lang.lang(xxx, xxx)来获取翻译过的值
"""
def __init__(self, language: str = 'zh-CN') -> None:
self.语言 = language
self.翻译结果 = tools.load_file(f'configs/lang/{language}.toml')
def __init__(self) -> None:
self.翻译结果 = tools.load_file(f'configs/lang/{DR_runtime.language}.toml')
self.默认翻译 = tools.load_file('configs/lang/zh-CN.toml')
self.直接返回原始数据 = True
def __str__(self) -> str:
return self.语言
return DR_option.language
def __getitem__(self, item) -> Union[int, str, list, dict]:
try:
@ -76,26 +76,7 @@ class Lang:
try:
return self.默认翻译[item]
except KeyError:
raise TranslateKeyNotFound(f'there\'s no key {item} in both {self.语言} and zh-CN')
def __setitem__(self, key, value) -> None:
if key == 'language' or key == 'lang':
try:
self.翻译结果 = tools.load_file(f'configs/lang/{value}.toml')
self.语言 = value
except FileNotFoundError:
if self.直接返回原始数据:
return None
raise TranslateKeyNotFound(f'{value}\'s language toml file not found')
else:
raise NotImplementedError
def set_language(self, language) -> None:
try:
self.翻译结果 = tools.load_file(f'configs/lang/{language}.toml')
self.语言 = language
except FileNotFoundError:
raise TranslateKeyNotFound(f'{language}\'s language toml file not found')
raise TranslateKeyNotFound(f'there\'s no key {item} in both {DR_option.language} and zh-CN')
def lang(self, *args) -> Union[int, str, list, dict]:
# frame = inspect.currentframe()
@ -117,13 +98,22 @@ class Lang:
except KeyError:
if self.直接返回原始数据:
return args
raise TranslateKeyNotFound(f'there\'s no key {args} in both {self.语言} and zh-CN')
raise TranslateKeyNotFound(f'there\'s no key {args} in both {DR_option.language} and zh-CN')
def 翻译(self, *args) -> Union[int, str, list, dict]:
return self.lang(args)
def _update_lang(self) -> str:
"""
用于更新语言(内部调用)
@return: 设置完成后的语言
"""
self.翻译结果 = tools.load_file(f'configs/lang/{DR_option.language}.toml')
return DR_option.language
tr = Lang('zh-CN')
if not __name__ == '__main__':
tr = Lang()
# font's value

2
libs/MCDR/serializer.py
View File

@ -8,7 +8,7 @@ from enum import EnumMeta
from threading import Lock
from typing import Union, TypeVar, List, Dict, Type, get_type_hints, Any
from MCDR.version import Version
from libs.MCDR.version import Version
"""
This part of code come from MCDReforged(https://github.com/Fallen-Breath/MCDReforged)

2
requirement.txt
View File

@ -1,4 +1,4 @@
psutil
pillow
objprint
selenium
# selenium