Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 8 additions & 7 deletions packages/helpermodules/logger.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@
from pathlib import Path
import sys
import threading
from typing import Optional, Union
import typing_extensions
import re
import io
Expand Down Expand Up @@ -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.

Expand Down Expand Up @@ -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():
Expand All @@ -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')
Expand All @@ -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):
Expand Down Expand Up @@ -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"]}
Expand Down
2 changes: 1 addition & 1 deletion packages/helpermodules/utils/_thread_handler.py
Original file line number Diff line number Diff line change
Expand Up @@ -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]
Expand Down
8 changes: 4 additions & 4 deletions packages/modules/configuration.py
Original file line number Diff line number Diff line change
@@ -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
Expand Down Expand Up @@ -184,17 +184,17 @@ 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:
Pub().pub("openWB/set/system/configurable/devices_components",
_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):
Expand Down
Loading