diff --git a/packages/helpermodules/logger.py b/packages/helpermodules/logger.py index 73fdf53719..ae13db41b2 100644 --- a/packages/helpermodules/logger.py +++ b/packages/helpermodules/logger.py @@ -4,6 +4,7 @@ from pathlib import Path import sys import threading +from typing import Optional, Union import typing_extensions import re import io @@ -31,7 +32,7 @@ URL_CREDENTIALS_PATTERN = (r'(\w+://[^/\s:@]*):[^/\s]+@', r'\1:***REDACTED***@') -def redact_sensitive_info(message: str, additional_fields: list = None) -> str: +def redact_sensitive_info(message: str, additional_fields: Optional[list] = None) -> str: """ Redacts sensitive information from the given message. @@ -163,7 +164,7 @@ def clear(self): self.line_count = 0 -def clear_in_memory_log_handler(logger_name: str = None) -> None: +def clear_in_memory_log_handler(logger_name: Optional[str] = None) -> None: if logger_name is None: # Clear all in-memory log handlers for handler in in_memory_log_handlers.values(): @@ -174,8 +175,8 @@ def clear_in_memory_log_handler(logger_name: str = None) -> None: in_memory_log_handlers[logger_name].clear() -def write_logs_to_file(logger_name: str = None) -> None: - def rotate_logs(base_path: str, name: str): +def write_logs_to_file(logger_name: Optional[str] = None) -> None: + def rotate_logs(base_path: Union[str, os.PathLike], name: str): # Rotate the log files for i in range(NUMBER_OF_LOGFILES-1, 0, -1): src = os.path.join(base_path, f'{name}.previous{i}.log') @@ -187,7 +188,7 @@ def rotate_logs(base_path: str, name: str): if os.path.exists(current_log): shutil.move(current_log, os.path.join(base_path, f'{name}.previous1.log')) - def combine_logs(base_path: str, name: str): + def combine_logs(base_path: Union[str, os.PathLike], name: str): latest_log_path = os.path.join(base_path, f'{name}.latest.log') with open(latest_log_path, 'w') as latest_log: for i in range(NUMBER_OF_LOGFILES-1, -1, -1): @@ -257,8 +258,8 @@ def setup_generate_totals_logging(include_process_log: bool = False) -> None: def setup_logging() -> None: - def mb_to_bytes(megabytes: int) -> int: - return megabytes * 1000000 + def mb_to_bytes(megabytes: float) -> int: + return int(megabytes * 1000000) global in_memory_log_handlers in_memory_log_handlers = {name: InMemoryLogHandler() for name in ["main", "internal_chargepoint"]} diff --git a/packages/helpermodules/utils/_thread_handler.py b/packages/helpermodules/utils/_thread_handler.py index cea839dd20..eb3f4929ae 100644 --- a/packages/helpermodules/utils/_thread_handler.py +++ b/packages/helpermodules/utils/_thread_handler.py @@ -6,7 +6,7 @@ log = logging.getLogger(__name__) -def joined_thread_handler(threads: List[Thread], timeout: Optional[int]) -> List[str]: +def joined_thread_handler(threads: List[Thread], timeout: Optional[float]) -> List[str]: def split_chunks(to_split, n): for i in range(0, len(to_split), n): yield to_split[i:i + n] diff --git a/packages/modules/configuration.py b/packages/modules/configuration.py index 7122f1d2d6..e85fed327e 100644 --- a/packages/modules/configuration.py +++ b/packages/modules/configuration.py @@ -1,7 +1,7 @@ import importlib import logging from pathlib import Path -from typing import Dict, List +from typing import Dict, List, Optional import dataclass_utils from helpermodules.pub import Pub @@ -184,9 +184,9 @@ def _pub_configurable_soc_modules() -> None: }) Pub().pub("openWB/set/system/configurable/soc_modules", soc_modules) except Exception as e: - log.exception(f"Fehler {e} im configuration-Modul {path}") + log.exception(f"Fehler {e} im configuration-Modul") if hasattr(sys, '_called_from_test'): - print(f"Fehler {e} im configuration-Modul {path}") + print(f"Fehler {e} im configuration-Modul") def _pub_configurable_devices_components() -> None: @@ -194,7 +194,7 @@ def _pub_configurable_devices_components() -> None: _get_configurable_devices_components(["*bat*", "*counter*", "*inverter*"])) -def _get_configurable_devices_components(component_pattern: List[str]) -> None: +def _get_configurable_devices_components(component_pattern: List[str]) -> Optional[Dict]: def update_nested_dict(dictionary: Dict, update: Dict) -> Dict: for key, value in update.items(): if isinstance(value, dict):