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

sync pyglet 0.1.8

Browse Source
This commit is contained in:
shenjack 2023-11-05 23:49:44 +08:00
parent b28ef5c580
commit c7140e0ef2
Signed by: shenjack
GPG Key ID: 7B1134A979775551
8 changed files with 314 additions and 80 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
libs/lib_not_dr/__init__.py
View File

@ -4,4 +4,4 @@
# All rights reserved
# -------------------------------
__version__ = '0.1.7'
__version__ = '0.1.8'

37
libs/lib_not_dr/logger/formatter/__init__.py
View File

@ -5,17 +5,22 @@
# -------------------------------
import time
from pathlib import Path
from string import Template
from typing import List, Union, Optional, Dict, Tuple
from typing import List, Union, Optional, Dict, Tuple, TYPE_CHECKING
from lib_not_dr.types.options import Options
from lib_not_dr.logger.structers import LogMessage, FormattingMessage
from lib_not_dr.logger.structure import LogMessage, FormattingMessage
if TYPE_CHECKING:
from lib_not_dr.logger.formatter.colors import BaseColorFormatter
class BaseFormatter(Options):
name = 'BaseFormatter'
sub_formatter: List['BaseFormatter'] = []
color_formatters: List['BaseColorFormatter'] = []
default_template: str = '${log_time}|${logger_name}|${logger_tag}|${level}|${messages}'
@classmethod
@ -176,7 +181,7 @@ class TraceFormatter(BaseFormatter):
def _format(self, message: FormattingMessage) -> FormattingMessage:
if message[0].stack_trace is None:
return message
message[1]['log_source'] = message[0].stack_trace.f_code.co_filename
message[1]['log_source'] = Path(message[0].stack_trace.f_code.co_filename)
message[1]['log_line'] = message[0].stack_trace.f_lineno
message[1]['log_function'] = message[0].stack_trace.f_code.co_name
return message
@ -187,21 +192,40 @@ class StdFormatter(BaseFormatter):
enable_color: bool = True
sub_formatter = [TimeFormatter(),
sub_formatter: List[BaseFormatter] = [TimeFormatter(),
LevelFormatter(),
TraceFormatter()]
from lib_not_dr.logger.formatter.colors import (LevelColorFormatter,
LoggerColorFormatter,
TimeColorFormatter,
TraceColorFormatter,
MessageColorFormatter)
color_formatters = [LevelColorFormatter(),
color_formatters: List[BaseFormatter] = [LevelColorFormatter(),
LoggerColorFormatter(),
TimeColorFormatter(),
TraceColorFormatter(),
MessageColorFormatter()]
def __init__(self,
enable_color: bool = True,
sub_formatter: Optional[List[BaseFormatter]] = None,
color_formatters: Optional[List[BaseFormatter]] = None,
**kwargs) -> None:
"""
Initialize the StdFormatter
:param enable_color: enable color
:param sub_formatter: list of sub formatter
:param color_formatters: list of color formatter
:param kwargs: other options
"""
# 同 structures.LogMessage.__init__ 的注释 (逃)
self.enable_color = enable_color
if sub_formatter is not None:
self.sub_formatter = sub_formatter
if color_formatters is not None:
self.color_formatters = color_formatters
super().__init__(**kwargs)
def _format(self, message: FormattingMessage) -> FormattingMessage:
super()._format(message)
@ -259,4 +283,3 @@ if __name__ == '__main__':
print(std_format.format_message(log_message), end='')
print(std_format.as_markdown())

2
libs/lib_not_dr/logger/formatter/colors.py
View File

@ -7,7 +7,7 @@
from typing import Dict, Tuple
from lib_not_dr.logger.formatter import BaseFormatter
from lib_not_dr.logger.structers import FormattingMessage
from lib_not_dr.logger.structure import FormattingMessage
__all__ = [

2
libs/lib_not_dr/logger/logger.py
View File

@ -9,7 +9,7 @@ import inspect
from types import FrameType
from typing import List, Optional
from lib_not_dr.logger.structers import LogMessage
from lib_not_dr.logger.structure import LogMessage
from lib_not_dr.logger.outstream import BaseOutputStream
from lib_not_dr.types.options import Options

178
libs/lib_not_dr/logger/outstream.py
View File

@ -4,23 +4,35 @@
# All rights reserved
# -------------------------------
import io
import sys
import time
import string
import atexit
import threading
from pathlib import Path
from typing import Optional
from lib_not_dr.types.options import Options
from lib_not_dr.logger.structers import LogMessage
from lib_not_dr.logger.structure import LogMessage
from lib_not_dr.logger.formatter import BaseFormatter, StdFormatter
__all__ = [
'BaseOutputStream'
'BaseOutputStream',
'StdioOutputStream',
'FileCacheOutputStream'
]
class BaseOutputStream(Options):
name = 'BaseOutputStream'
level: int = 20
level: int = 10
enable: bool = True
formatter: BaseFormatter
def write_stdout(self, message: LogMessage) -> None:
raise NotImplementedError(f'{self.__class__.__name__}.write_stdout is not implemented')
@ -30,10 +42,14 @@ class BaseOutputStream(Options):
def flush(self) -> None:
raise NotImplementedError(f'{self.__class__.__name__}.flush is not implemented')
def close(self) -> None:
self.enable = False
class StdioOutputStream(BaseOutputStream):
name = 'StdioOutputStream'
level: int = 10
formatter: BaseFormatter = StdFormatter()
def write_stdout(self, message: LogMessage) -> None:
@ -60,3 +76,159 @@ class StdioOutputStream(BaseOutputStream):
print('', end='', flush=True)
print('', end='', flush=True, file=sys.stderr)
return None
class FileCacheOutputStream(BaseOutputStream):
name = 'FileCacheOutputStream'
level: int = 10
formatter: BaseFormatter = StdFormatter(enable_color=False)
text_cache: io.StringIO = None
flush_counter: int = 0
# 默认 10 次 flush 一次
flush_count_limit: int = 10
flush_time_limit: int = 10 # time limit in sec, 0 means no limit
flush_lock: threading.Lock = None
flush_timer: threading.Timer = None
file_path: Optional[Path] = Path('./logs')
file_name: str
# file mode: always 'a'
file_encoding: str = 'utf-8'
# do file swap or not
file_swap: bool = False
at_exit_register: bool = False
file_swap_counter: int = 0
file_swap_name_template: str = '${name}-${counter}.log'
# ${name} -> file_name
# ${counter} -> file_swap_counter
# ${log_time} -> time when file swap ( round(time.time()) )
# ${start_time} -> start time of output stream ( round(time.time()) )
current_file_name: str = None
file_start_time: int = None
# log file swap triggers
# 0 -> no limit
file_size_limit: int = 0 # size limit in kb
file_time_limit: int = 0 # time limit in sec 0
file_swap_on_both: bool = False # swap file when both size and time limit reached
def init(self, **kwargs) -> bool:
self.file_start_time = round(time.time())
if self.text_cache is None:
self.text_cache = io.StringIO()
self.flush_lock = threading.Lock()
self.get_file_path()
return False
def _write(self, message: LogMessage) -> None:
"""
write message to text cache
默认已经检查过了
:param message: message to write
:return: None
"""
self.text_cache.write(self.formatter.format_message(message))
self.flush_counter += 1
if message.flush or self.flush_counter >= self.flush_count_limit:
self.flush()
else:
if self.flush_time_limit > 0:
if self.flush_timer is None or not self.flush_timer.is_alive():
self.flush_timer = threading.Timer(self.flush_time_limit, self.flush)
self.flush_timer.daemon = True
self.flush_timer.start()
if not self.at_exit_register:
atexit.register(self.flush)
self.at_exit_register = True
return None
def write_stdout(self, message: LogMessage) -> None:
if not self.enable:
return None
if message.level < self.level:
return None
self._write(message)
return None
def write_stderr(self, message: LogMessage) -> None:
if not self.enable:
return None
if message.level < self.level:
return None
self._write(message)
return None
def get_file_path(self) -> Path:
"""
get file path
:return:
"""
if (current_file := self.current_file_name) is None:
if not self.file_swap:
# 直接根据 file name 生成文件
current_file = Path(self.file_path) / self.file_name
self.current_file_name = str(current_file)
return current_file
template = string.Template(self.file_swap_name_template)
file_name = template.safe_substitute(name=self.file_name,
counter=self.file_swap_counter,
log_time=round(time.time()),
start_time=self.file_start_time)
current_file = Path(self.file_path) / file_name
self.current_file_name = str(current_file)
else:
current_file = Path(current_file)
return current_file
def check_flush(self) -> Path:
current_file = self.get_file_path()
# 获取当前文件的路径
if not self.file_swap:
# 不需要 swap 直接返回
return current_file
# 检查是否需要 swap
size_pass = True
if self.file_size_limit > 0:
file_size = current_file.stat().st_size / 1024 # kb
if file_size > self.file_size_limit: # kb
size_pass = False
time_pass = True
if self.file_time_limit > 0:
file_time = round(time.time()) - current_file.stat().st_mtime
if file_time > self.file_time_limit:
time_pass = False
if (self.file_swap_on_both and size_pass and time_pass) or \
(not self.file_swap_on_both and (size_pass or time_pass)):
# 两个都满足
# 或者只有一个满足
if size_pass and time_pass:
self.file_swap_counter += 1
# 生成新的文件名
return self.get_file_path()
def flush(self) -> None:
new_cache = io.StringIO() # 创建新的缓存
self.flush_counter = 0 # atomic, no lock
with self.flush_lock:
text = self.text_cache.getvalue()
old_cache, self.text_cache = self.text_cache, new_cache
old_cache.close() # 关闭旧的缓存
if text == '':
return None
current_file = self.check_flush()
if not current_file.exists():
current_file.parent.mkdir(parents=True, exist_ok=True)
current_file.touch(exist_ok=True)
with current_file.open('a', encoding=self.file_encoding) as f:
f.write(text)
return None
def close(self) -> None:
super().close()
self.flush()
self.text_cache.close()
atexit.unregister(self.flush)
return None

60
libs/lib_not_dr/logger/structers.py
View File

@ -1,60 +0,0 @@
# -------------------------------
# Difficult Rocket
# Copyright © 2020-2023 by shenjackyuanjie 3695888@qq.com
# All rights reserved
# -------------------------------
import time
from types import FrameType
from typing import List, Optional, Tuple, Dict
from lib_not_dr.types.options import Options
__all__ = ['LogMessage',
'FormattingMessage']
class LogMessage(Options):
name = 'LogMessage'
# 消息内容本身的属性
messages: List[str] = []
end: str = '\n'
split: str = ' '
# 消息的属性
flush: bool = True
level: int = 20
log_time: float = None # time.time_ns()
logger_name: str = 'root'
logger_tag: Optional[str] = None
stack_trace: Optional[FrameType] = None
def init(self, **kwargs) -> bool:
if self.log_time is None:
self.log_time = time.time_ns()
return False
def format_message(self) -> str:
return self.split.join(self.messages) + self.end
def format_for_message(self) -> Dict[str, str]:
basic_info = self.option()
if self.logger_tag is None:
basic_info['logger_tag'] = ' '
basic_info['messages'] = self.format_message()
return basic_info
@property
def create_msec_3(self) -> int:
return int(self.log_time / 1000000) % 1000
FormattingMessage = Tuple[LogMessage, Dict[str, str]]
if __name__ == '__main__':
print(LogMessage().as_markdown())

98
libs/lib_not_dr/logger/structure.py Normal file
View File

@ -0,0 +1,98 @@
# -------------------------------
# Difficult Rocket
# Copyright © 2020-2023 by shenjackyuanjie 3695888@qq.com
# All rights reserved
# -------------------------------
import time
from types import FrameType
from typing import List, Optional, Tuple, Dict
from lib_not_dr.types.options import Options
__all__ = ['LogMessage',
'FormattingMessage']
class LogMessage(Options):
name = 'LogMessage'
# 消息内容本身的属性
messages: List[str] = []
end: str = '\n'
split: str = ' '
# 消息的属性
flush: bool = None
level: int = 20
log_time: float = None # time.time_ns()
logger_name: str = 'root'
logger_tag: Optional[str] = None
stack_trace: Optional[FrameType] = None
def __init__(self,
messages: Optional[List[str]] = None,
end: Optional[str] = '\n',
split: Optional[str] = ' ',
flush: Optional[bool] = None,
level: Optional[int] = 20,
log_time: Optional[float] = None,
logger_name: Optional[str] = 'root',
logger_tag: Optional[str] = None,
stack_trace: Optional[FrameType] = None,
**kwargs) -> None:
"""
Init for LogMessage
:param messages: message list for log
:param end: end of message
:param split: split for messages
:param flush: do flush or not
:param level: level of message
:param log_time: time of message (default: time.time_ns())
:param logger_name: name of logger
:param logger_tag: tag of logger
:param stack_trace: stack trace of logger
:param kwargs: other options
"""
# 为了方便使用 单独覆盖了 __init__ 方法来提供代码补全的选项
super().__init__(messages=messages,
end=end,
split=split,
flush=flush,
level=level,
log_time=log_time,
logger_name=logger_name,
logger_tag=logger_tag,
stack_trace=stack_trace,
**kwargs)
def init(self, **kwargs) -> bool:
if self.log_time is None:
self.log_time = time.time_ns()
if not isinstance(self.flush, bool) and self.flush is not None:
self.flush = True if self.flush else False
return False
def format_message(self) -> str:
return self.split.join(self.messages) + self.end
def format_for_message(self) -> Dict[str, str]:
basic_info = self.option()
if self.logger_tag is None:
basic_info['logger_tag'] = ' '
basic_info['messages'] = self.format_message()
return basic_info
@property
def create_msec_3(self) -> int:
return int(self.log_time / 1000000) % 1000
FormattingMessage = Tuple[LogMessage, Dict[str, str]]
if __name__ == '__main__':
print(LogMessage().as_markdown())

3
libs/lib_not_dr/types/options.py
View File

@ -64,6 +64,7 @@ class Options:
"""
name = 'Option Base'
cached_options: Dict[str, Union[str, Any]] = {}
_check_options: bool = True
def __init__(self, **kwargs):
"""
@ -75,7 +76,7 @@ class Options:
self._options: Dict[str, Union[Callable, object]] = {}
self.flush_option()
for option, value in kwargs.items():
if option not in self.cached_options:
if option not in self.cached_options and self._check_options:
raise OptionNameNotDefined(f"option: {option} with value: {value} is not defined")
setattr(self, option, value)
run_load_file = True