sh-edraft.de/cpl
5
0
Fork 0
Code Issues 5 Pull Requests 2 Actions Packages Projects 1 Releases 12 Wiki Activity

Fixed spinner thread & remove log thread
All checks were successful
Build on push / prepare (push) Successful in 8s
Details
Build on push / core (push) Successful in 18s
Details
Build on push / query (push) Successful in 19s
Details
Build on push / translation (push) Successful in 18s
Details
Build on push / mail (push) Successful in 18s
Details

Browse Source
This commit is contained in:
Sven Heidemann
2025-09-16 18:41:57 +02:00
parent 8ad3e3bdb4
commit cd7dfaf2b4
7 changed files with 63 additions and 117 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
src/cpl-core/cpl/core/console/__init__.py
View File

@@ -1,5 +1,4 @@
from .background_color_enum import BackgroundColorEnum
from .console import Console
from .console_call import ConsoleCall
from ._call import ConsoleCall
from .foreground_color_enum import ForegroundColorEnum
from .spinner_thread import SpinnerThread

0
src/cpl-core/cpl/core/console/console_call.py → src/cpl-core/cpl/core/console/_call.py
View File

31
src/cpl-core/cpl/core/console/spinner_thread.py → src/cpl-core/cpl/core/console/_spinner.py
View File

@@ -1,7 +1,8 @@
import os
import sys
import threading
import multiprocessing
import time
from multiprocessing import Process
from termcolor import colored
@@ -9,8 +10,8 @@ from cpl.core.console.background_color_enum import BackgroundColorEnum
from cpl.core.console.foreground_color_enum import ForegroundColorEnum
class SpinnerThread(threading.Thread):
r"""Thread to show spinner in terminal
class Spinner(Process):
r"""Process to show spinner in terminal
Parameter:
msg_len: :class:`int`
@@ -22,7 +23,7 @@ class SpinnerThread(threading.Thread):
"""
def __init__(self, msg_len: int, foreground_color: ForegroundColorEnum, background_color: BackgroundColorEnum):
threading.Thread.__init__(self)
Process.__init__(self)
self._msg_len = msg_len
self._foreground_color = foreground_color
@@ -50,29 +51,26 @@ class SpinnerThread(threading.Thread):
return color_args
def run(self) -> None:
r"""Entry point of thread, shows the spinner"""
r"""Entry point of process, shows the spinner"""
columns = 0
if sys.platform == "win32":
columns = os.get_terminal_size().columns
else:
term_rows, term_columns = os.popen("stty size", "r").read().split()
values = os.popen("stty size", "r").read().split()
term_rows, term_columns = values if len(values) == 2 else (0, 0)
columns = int(term_columns)
end_msg = "done"
end_msg_pos = columns - self._msg_len - len(end_msg)
if end_msg_pos > 0:
print(f'{"" : >{end_msg_pos}}', end="")
padding = columns - self._msg_len - len(end_msg)
if padding > 0:
print(f'{"" : >{padding}}', end="")
else:
print("", end="")
first = True
spinner = self._spinner()
while self._is_spinning:
if first:
first = False
print(colored(f"{next(spinner): >{len(end_msg) - 1}}", *self._get_color_args()), end="")
else:
print(colored(f"{next(spinner): >{len(end_msg)}}", *self._get_color_args()), end="")
print(colored(f"{next(spinner): >{len(end_msg)}}", *self._get_color_args()), end="")
time.sleep(0.1)
back = ""
for i in range(0, len(end_msg)):
@@ -84,9 +82,10 @@ class SpinnerThread(threading.Thread):
if not self._exit:
print(colored(end_msg, *self._get_color_args()), end="")
def stop_spinning(self):
def stop(self):
r"""Stops the spinner"""
self._is_spinning = False
super().terminate()
time.sleep(0.1)
def exit(self):

8
src/cpl-core/cpl/core/console/console.py
View File

@@ -10,9 +10,9 @@ from tabulate import tabulate
from termcolor import colored
from cpl.core.console.background_color_enum import BackgroundColorEnum
from cpl.core.console.console_call import ConsoleCall
from cpl.core.console._call import ConsoleCall
from cpl.core.console.foreground_color_enum import ForegroundColorEnum
from cpl.core.console.spinner_thread import SpinnerThread
from cpl.core.console._spinner import Spinner
class Console:
@@ -464,7 +464,7 @@ class Console:
cls.set_hold_back(True)
spinner = None
if not cls._disabled:
spinner = SpinnerThread(len(message), spinner_foreground_color, spinner_background_color)
spinner = Spinner(len(message), spinner_foreground_color, spinner_background_color)
spinner.start()
return_value = None
@@ -476,7 +476,7 @@ class Console:
cls.close()
if spinner is not None:
spinner.stop_spinning()
spinner.stop()
cls.set_hold_back(False)
cls.set_foreground_color(ForegroundColorEnum.default)

92
src/cpl-core/cpl/core/log/_log_writer.py
View File

@@ -1,92 +0,0 @@
import multiprocessing
import os
from datetime import datetime
from typing import Self
from cpl.core.console import Console
from cpl.core.log.log_level_enum import LogLevelEnum
class LogWriter:
_instance = None
# ANSI color codes for different log levels
_COLORS = {
LogLevelEnum.trace: "\033[37m", # Light Gray
LogLevelEnum.debug: "\033[94m", # Blue
LogLevelEnum.info: "\033[92m", # Green
LogLevelEnum.warning: "\033[93m", # Yellow
LogLevelEnum.error: "\033[91m", # Red
LogLevelEnum.fatal: "\033[95m", # Magenta
}
def __init__(self, file_prefix: str, level: LogLevelEnum = LogLevelEnum.info):
self._file_prefix = file_prefix
self._level = level
self._queue = multiprocessing.Queue()
self._process = multiprocessing.Process(target=self._log_worker, daemon=True)
self._create_log_dir()
self._process.start()
@property
def level(self) -> LogLevelEnum:
return self._level
@level.setter
def level(self, value: LogLevelEnum):
assert isinstance(value, LogLevelEnum), "Log level must be an instance of LogLevelEnum"
self._level = value
@classmethod
def get_instance(cls, file_prefix: str, level: LogLevelEnum = LogLevelEnum.info) -> Self:
if cls._instance is None:
cls._instance = LogWriter(file_prefix, level)
return cls._instance
@staticmethod
def _create_log_dir():
if os.path.exists("logs"):
return
os.makedirs("logs")
def _log_worker(self):
"""Worker process that writes log messages from the queue to the file."""
while True:
content = self._queue.get()
if content is None: # Shutdown signal
break
self._write_log_to_file(content)
Console.write_line(f"{self._COLORS.get(self._level, '\033[0m')}{content}\033[0m")
@property
def log_file(self):
return f"logs/{self._file_prefix}_{datetime.now().strftime('%Y-%m-%d')}.log"
def _ensure_file_size(self):
log_file = self.log_file
if not os.path.exists(log_file) or os.path.getsize(log_file) <= 0.5 * 1024 * 1024:
return
# if exists and size is greater than 300MB, create a new file
os.rename(
log_file,
f"{log_file.split('.log')[0]}_{datetime.now().strftime('%H-%M-%S')}.log",
)
def _write_log_to_file(self, content: str):
self._ensure_file_size()
with open(self.log_file, "a") as log_file:
log_file.write(content + "\n")
log_file.close()
def log(self, content: str):
"""Enqueue log message without blocking main app."""
self._queue.put(content)
def close(self):
"""Gracefully stop the logging process."""
self._queue.put(None)
self._process.join()

45
src/cpl-core/cpl/core/log/logger.py
View File

@@ -3,7 +3,6 @@ import traceback
from datetime import datetime
from cpl.core.console import Console
from cpl.core.log._log_writer import LogWriter
from cpl.core.log.log_level_enum import LogLevelEnum
from cpl.core.log.logger_abc import LoggerABC
from cpl.core.typing import Messages, Source
@@ -13,6 +12,16 @@ class Logger(LoggerABC):
_level = LogLevelEnum.info
_levels = [x for x in LogLevelEnum]
# ANSI color codes for different log levels
_COLORS = {
LogLevelEnum.trace: "\033[37m", # Light Gray
LogLevelEnum.debug: "\033[94m", # Blue
LogLevelEnum.info: "\033[92m", # Green
LogLevelEnum.warning: "\033[93m", # Yellow
LogLevelEnum.error: "\033[91m", # Red
LogLevelEnum.fatal: "\033[95m", # Magenta
}
def __init__(self, source: Source, file_prefix: str = None):
LoggerABC.__init__(self)
assert source is not None and source != "", "Source cannot be None or empty"
@@ -22,7 +31,18 @@ class Logger(LoggerABC):
file_prefix = "app"
self._file_prefix = file_prefix
self._writer = LogWriter.get_instance(self._file_prefix)
self._create_log_dir()
@property
def log_file(self):
return f"logs/{self._file_prefix}_{datetime.now().strftime('%Y-%m-%d')}.log"
@staticmethod
def _create_log_dir():
if os.path.exists("logs"):
return
os.makedirs("logs")
@classmethod
def set_level(cls, level: LogLevelEnum):
@@ -31,6 +51,24 @@ class Logger(LoggerABC):
else:
raise ValueError(f"Invalid log level: {level}")
@staticmethod
def _ensure_file_size(log_file: str):
if not os.path.exists(log_file) or os.path.getsize(log_file) <= 0.5 * 1024 * 1024:
return
# if exists and size is greater than 300MB, create a new file
os.rename(
log_file,
f"{log_file.split('.log')[0]}_{datetime.now().strftime('%H-%M-%S')}.log",
)
def _write_log_to_file(self, content: str):
file = self.log_file
self._ensure_file_size(file)
with open(file, "a") as log_file:
log_file.write(content + "\n")
log_file.close()
def _log(self, level: LogLevelEnum, *messages: Messages):
try:
if self._levels.index(level) < self._levels.index(self._level):
@@ -39,7 +77,8 @@ class Logger(LoggerABC):
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")
formatted_message = self._format_message(level.value, timestamp, *messages)
self._writer.log(formatted_message)
self._write_log_to_file(formatted_message)
Console.write_line(f"{self._COLORS.get(self._level, '\033[0m')}{formatted_message}\033[0m")
except Exception as e:
print(f"Error while logging: {e} -> {traceback.format_exc()}")