Skip to content
Snippets Groups Projects
Commit 4025ce5d authored by Christoph Schmidt's avatar Christoph Schmidt
Browse files

Changed the logger implementation.

parent d0995eee
No related branches found
No related tags found
No related merge requests found
......@@ -13,8 +13,8 @@ from LaserConfig import LaserConfig
class ApplicationConfig(cfg.ConfigNode):
def __init__(self, internal_log=True, internal_log_level= logging.DEBUG) -> None:
super().__init__(internal_log=internal_log, internal_log_level=internal_log_level)
def __init__(self) -> None:
super().__init__()
# self.output_directory: cfg.Field[Path] = cfg.Field(Path("C:\\{wafer_list1}"))
#
......
......@@ -3,15 +3,17 @@ import sys
import time
from PySide6 import QtWidgets
from PySide6.QtCore import Signal
from PySide6.QtCore import Signal, QObject
from PySide6.QtWidgets import QApplication, QMainWindow, QWidget, QTreeWidget
from rich.logging import RichHandler
from ApplicationConfig import ApplicationConfig
class TestClass:
class TestClass(QObject):
signal = Signal(int)
def __init__(self):
self.logger = logging.getLogger(f"{__name__} - fallback")
super().__init__()
self._wafer = 0
@property
......@@ -21,7 +23,7 @@ class TestClass:
@wafer.setter
def wafer(self, value):
self._wafer = value
print("Wafer changed to", value)
self.logger.info(f"Wafer changed to {value}")
self.signal.emit(value)
......@@ -29,12 +31,18 @@ if __name__ == "__main__":
app = QApplication(sys.argv)
# setup the logging module
FORMAT = "%(name)s %(message)s"
logging.basicConfig(
level=logging.DEBUG, format=FORMAT, datefmt="[%X]", handlers=[RichHandler()]
)
config = ApplicationConfig()
config.module_log_enabled = False
config.module_log_level = logging.DEBUG
testclass = TestClass()
time.sleep(1)
#config.autosave(enable=True, path='./configs_autosave')
(config.load('./configs/ApplicationConfig.yaml'))
config.autosave(enable=True, path='./configs_autosave/ApplicationConfig.yaml')
(config.load('./configs/ApplicationConfig.yaml', as_auto_save=True))
#print(config.wafer_version)
#config.wafer_version.get()
#config.wafer_number.get()
......
......@@ -13,50 +13,69 @@ from rich.logging import RichHandler
class CObject:
def __init__(self, name: str = ""):
def __init__(self, module_log, module_log_level, name: str = "",):
if name is None or name == "":
self.name = f"{self.__class__.__name__.replace('Field', '')}({hash(self)})"
else:
self.name = f"{name}({os.getpid()})"
self._internal_logger, self._internal_log_handler = self.create_new_logger(f"(cfg) {self.name}")
self._module_logger = self.create_new_logger(f"(cfg) {self.name}", module_log, module_log_level)
# ==================================================================================================================
# Logging
# ==================================================================================================================
def create_new_logger(self, name: str) -> (logging.Logger, logging.Handler):
qh = RichHandler(rich_tracebacks=True)
def create_new_logger(self, name: str, enable: bool = True, level: int = 0, propagate: bool=True) -> logging.Logger:
_internal_logger = logging.getLogger(name)
_internal_logger.handlers = [qh]
_internal_logger.setLevel(logging.DEBUG)
formatter = logging.Formatter(
'%(name)s %(message)s')
qh.setFormatter(formatter)
return _internal_logger, qh
_internal_logger.handlers = [logging.NullHandler()]
_internal_logger.setLevel(level)
_internal_logger.disabled = not enable
_internal_logger.propagate = propagate
return _internal_logger
@property
def internal_log_enabled(self):
return not self._internal_logger.disabled
def module_log_enabled(self):
return not self._module_logger.disabled
@internal_log_enabled.setter
def internal_log_enabled(self, enable: bool) -> None:
@module_log_enabled.setter
def module_log_enabled(self, enable: bool) -> None:
"""
Enables or disables internal logging. If disabled, the internal logger will be disabled and no messages will be
emitted to the state queue.
:param enable: True to enable, False to disable
"""
self._internal_logger.disabled = not enable
if enable:
self._module_logger.disabled = False
self._module_logger.debug(f"Logger { self._module_logger.name} enabled (Level {self._module_logger.level}).")
else:
self._module_logger.debug(f"Logger {self._module_logger.name} disabled.")
self._module_logger.disabled = True
@property
def internal_log_level(self):
return self._internal_logger.level
def module_log_level(self):
return self._module_logger.level
@internal_log_level.setter
def internal_log_level(self, level: int) -> None:
@module_log_level.setter
def module_log_level(self, level: int) -> None:
"""
Sets the internal logging level.
:param level:
:return:
"""
self._internal_logger.setLevel(level)
\ No newline at end of file
self._module_logger.setLevel(level)
if self._module_logger is not None:
if level == logging.DEBUG:
self._module_logger.debug(f"Module log level of {self.__class__.__name__} has been set to DEBUG.")
elif level == logging.INFO:
self._module_logger.debug(f"Module log level of {self.__class__.__name__} has been set to INFO.")
elif level == logging.WARNING:
self._module_logger.debug(f"Module log level of {self.__class__.__name__} has been set to WARNING.")
elif level == logging.ERROR:
self._module_logger.debug(f"Module log level of {self.__class__.__name__} has been set to ERROR.")
elif level == logging.CRITICAL:
self._module_logger.debug(f"Module log level of {self.__class__.__name__} has been set to CRITICAL.")
else:
self._module_logger.debug(
f"Module log level of {self.__class__.__name__} has been set to level {level}.")
else:
raise Exception("Can't set internal log level. Internal logger not initialized")
......@@ -7,6 +7,8 @@ Description:
"""
import datetime
import logging
import os
import pathlib
import yaml
......@@ -22,17 +24,15 @@ class ConfigNode(CObject):
cur_time = datetime.datetime.now()
def __init__(self, internal_log, internal_log_level):
super().__init__()
def __init__(self, module_log=True, module_log_level=logging.WARNING):
super().__init__(module_log, module_log_level)
self._autosave = False
self.name = self.__class__.__name__
self.config_file: pathlib.Path = pathlib.Path(f"./{self.name}.yaml")
self.name = self.__class__.__name__
self.logger, self.log_handler = self.create_new_logger(self.name)
self.internal_log_enabled = internal_log
self.internal_log_level = internal_log_level
self.logger = self.create_new_logger(self.name)
self.owner = None
self._level = 0
......@@ -49,8 +49,6 @@ class ConfigNode(CObject):
self.field_changed.connect(self._on_field_changed)
# ==================================================================================================================
#
# ==================================================================================================================
......@@ -102,24 +100,56 @@ class ConfigNode(CObject):
def deserialize(self, content):
"""Deserializes the content of the config based on the yaml file"""
self._internal_logger.info(f"Deserializing {content}")
self._module_logger.info(f"Deserializing {content}")
for attr, val in content.items():
# Check if the attribute is not of type GenericConfig
# therefore get the attribute type of this class
# print(f"Parsing {attr} with content: {val}")
if attr == self.name:
print(f"Found own config")
self.deserialize(val)
elif attr in self.__dict__:
if not isinstance(getattr(self, attr), ConfigNode):
self._internal_logger.info(f"Deserializing field {attr} with content: {val}")
self._module_logger.info(f"Deserializing field {attr} with content: {val}")
val = getattr(self, attr)._field_parser(val)
getattr(self, attr).set(**val, force_emit=True)
else:
self._internal_logger.info(f"Deserializing config {attr} with content: {val}")
self._module_logger.info(f"Deserializing config {attr} with content: {val}")
getattr(self, attr).deserialize(val)
@property
def module_log_level(self):
return self._module_logger.level
@module_log_level.setter
def module_log_level(self, level: int) -> None:
self._module_logger.setLevel(level)
for attr, val in self.__dict__.items():
if isinstance(val, Field):
self.fields[attr] = val
val.module_log_level = self.module_log_level
@property
def module_log_enabled(self):
return not self._module_logger.disabled
@module_log_enabled.setter
def module_log_enabled(self, enable: bool) -> None:
"""
Enables or disables internal logging. If disabled, the internal logger will be disabled and no messages will be
emitted to the state queue.
:param enable: True to enable, False to disable
"""
if enable:
self._module_logger.disabled = False
self._module_logger.debug(
f"Logger {self._module_logger.name} enabled (Level {self._module_logger.level}).")
else:
self._module_logger.debug(f"Logger {self._module_logger.name} disabled.")
self._module_logger.disabled = True
for attr, val in self.__dict__.items():
if isinstance(val, Field):
self.fields[attr] = val
val.module_log_enabled = self.module_log_enabled
# ==================================================================================================================
......@@ -136,8 +166,8 @@ class ConfigNode(CObject):
self.fields[attr] = val
# val.register(self.keywords, self.view.keywords_changed)
val.register(self.__class__.__name__, attr, self.keywords, self.field_changed)
val.internal_log_enabled = self.internal_log_enabled
val.internal_log_level = self.internal_log_level
val.module_log_enabled = self.module_log_enabled
val.module_log_level = self.module_log_level
self.view.keywords_changed.emit(self.keywords)
def _register_config(self):
......@@ -153,51 +183,57 @@ class ConfigNode(CObject):
# I/O Operations
# ==================================================================================================================
def save(self, file: str=None, background_save=True):
if file is None:
file = f"{self._path}/{self.__class__.__name__}.yaml"
if file is not None:
self.config_file = pathlib.Path(file)
# write the string to a file
with open(file, 'w+') as stream:
with open(self.config_file, 'w+') as stream:
stream.write(self.serialize())
# print(self.serialize())
if not background_save:
self._internal_logger.debug(f"Saved config to {file}")
# with open(file, 'w+') as stream:
# yaml.dump(self, stream) # Dump it as a xaml file
# with open(file, 'w+') as stream:
# stream.write(
# print(self._dump(cfg))
self._module_logger.debug(f"Saved config to {file}")
def autosave(self, enable: bool = False, path: str = None):
self._autosave = enable
if self._autosave:
if path is None:
self._path = pathlib.Path(".")
if path is not None:
# Check if the given path is a file or folder
_path = pathlib.Path(f"./{path}")
if _path.suffix == "" or path[-1].strip() == "/":
self.config_file = pathlib.Path(_path) / f"{self.name}.yaml"
else:
self._path = pathlib.Path(path)
self.config_file = pathlib.Path(_path)
self.config_file.parent.mkdir(parents=True, exist_ok=True)
self._module_logger.info(
f"Autosave enabled. File will be saved to {self.config_file.absolute().as_posix()}")
# Check if the path exists otherwise create it
if not self._path.exists():
self._path.mkdir(parents=True, exist_ok=True)
def load(self, file: str):
def load(self, file: str, as_auto_save: bool = False):
# load the yaml file
with open(file, 'r') as stream:
_file = pathlib.Path(file)
self._module_logger.info(f"Loading config from {_file}")
with open(str(_file.absolute().as_posix()), 'r') as stream:
content = yaml.load(stream, Loader=yaml.FullLoader)
self.deserialize(content)
if as_auto_save:
self._module_logger.debug(f"Configuration will be saved to {file}")
self.config_file = pathlib.Path(file)
# ==================================================================================================================
# Functions that happens on a change
# ==================================================================================================================
def _on_field_changed(self, *args, **kwargs):
# Emit that a field has changed, thus the keywords have changed
# print(f"Field changed {self.keywords}")
for attr, val in self.fields.items():
val: Field
val._on_keyword_changed()
if self._level == 0 and self._autosave:
file = f"{self._path}/{self.__class__.__name__}.yaml"
# Saves on every field change
self.save(file=file, background_save=True)
# Saves on every field change
self.save(file=str(self.config_file.as_posix()), background_save=True)
#self._module_logger.debug(f"Autosave to {self.config_file.absolute().as_posix()}")
......@@ -30,14 +30,13 @@ class Field(Generic[T], CObject):
changed = CSignal()
def __init__(self, value: T, friendly_name: str = None, description: str = None,
internal_log=True, internal_log_level=logging.INFO):
super().__init__()
self.internal_log_enabled = internal_log
self.internal_log_level = internal_log_level
module_log=True, module_log_level=logging.WARNING):
super().__init__(module_log, module_log_level)
self.logger, self.log_handler = self.create_new_logger(self.name)
self.field_name = self.__class__.__name__
self.logger = self.create_new_logger(self.name)
self._data = FieldData(self.name, value, friendly_name, description)
self._data = FieldData(self.field_name, value, friendly_name, description)
self._friendly_name: str = friendly_name
self._description: str = description
......@@ -51,7 +50,7 @@ class Field(Generic[T], CObject):
# Connected properties that should bet set if the field changes
self.props = []
self._internal_logger.debug(f"Field {self.__class__.__name__} created with value {value} of type {type(value)}")
self._module_logger.debug(f"Field {self.field_name} created with value {value} of type {type(value)}")
def __new__(cls, value, friendly_name: str = None, description: str = None):
# print(f"Field {cls.__name__} created with value {value} of type {type(value)} -> {isinstance(value, int)}")
......@@ -82,7 +81,7 @@ class Field(Generic[T], CObject):
def serialize(self):
"""Used for serializing instances. Returns the current field as a yaml-line."""
return f"{self.name}: {self._yaml_repr()} # {self.friendly_name}: {self.description}"
return f"{self.name}.{self.field_name}: {self._yaml_repr()} # {self.friendly_name}: {self.description}"
def connect_property(self, instance, prop: property):
self.props.append((instance, prop))
......@@ -99,23 +98,20 @@ class Field(Generic[T], CObject):
# ==================================================================================================================
# Register the field to a configuration
# ==================================================================================================================
def register(self, owner, name, keywords, csig_field_changed: CSignal):
def register(self, owner, field_name, keywords, csig_field_changed: CSignal):
"""
Register the keyword for the field. This is used for updating the keywords when the value is changed.
Should only be called from a configuration class
:param owner: The owner (parent) of the field
:param name: The name of the field
:param field_name: The name of the field
:param keywords: The keywords dict
:param csig_field_changed: The signal that is emitted when the keywords are changed
"""
self.name = name
self.field_name = field_name
self.owner = owner
formatter = logging.Formatter(f'%(name)s [{self.name}] %(message)s')
self._internal_log_handler.setFormatter(formatter)
if self._friendly_name is None:
self._friendly_name = self.name
self._friendly_name = self.field_name
# Assigns the global keywords dict to the field
self.keywords = keywords
......@@ -124,8 +120,8 @@ class Field(Generic[T], CObject):
# self.keywords_changed = keyword_changed
# self.keywords_changed.connect(self._on_keyword_changed)
self.set_keywords()
self._internal_logger.info(f"Field '{self.name}' assigned to {self.owner}")
self._module_logger.info(f"Field '{self.field_name}' assigned to {self.owner}")
self._module_logger.name = f"(cfg) {self.name}.{self.field_name}"
# ==================================================================================================================
# Set the keywords for the field
# ==================================================================================================================
......@@ -133,7 +129,7 @@ class Field(Generic[T], CObject):
"""Set the keywords for the field. Also updates the keywords dict if a value of a field is changed."""
# self.keywords["{" + self.name + "}"] = self.value
# self._internal_logger.info(f"Setting keywords for {self.name} to {self.value}")
self.keywords[self.name] = str(self.value).replace(' ', '_').replace('.', '_').replace(',', '_')
self.keywords[self.field_name] = str(self.value).replace(' ', '_').replace('.', '_').replace(',', '_')
self.csig_field_changed.emit()
def replace_keywords(self, fr: str):
......@@ -184,7 +180,7 @@ class Field(Generic[T], CObject):
def set(self, value: T, *args, force_emit: bool = False, **kwargs):
if not self._value_to_emit == value or force_emit:
self._internal_logger.info(f"{self.name} = {value} ({type(value)})")
self._module_logger.info(f"{self.field_name} = {value} ({type(value)})")
self._set_all_props(value)
self._set(value, *args, **kwargs)
self.set_keywords()
......
......@@ -37,10 +37,10 @@ class FieldPath(Field):
# Overwritten function, to replace the @Path keyword
match = re.findall(r'@Path:<([^>]*)>', val)
if len(match) > 0:
self._internal_logger.info(f"Found @Path: {val}. Check syntax, multiple @Path: are not allowed in one field.")
self._module_logger.info(f"Found @Path: {val}. Check syntax, multiple @Path: are not allowed in one field.")
return {"value": Path(match[0])}
elif len(match) == 0:
self._internal_logger.debug(f"No @Path: found in {val}. Please check field.")
self._module_logger.debug(f"No @Path: found in {val}. Please check field.")
return {"value": Path('./')}
def __str__(self):
return str(Path(self.value).as_posix())
\ No newline at end of file
......@@ -52,7 +52,7 @@ class FieldViewPath(FieldView):
grd.addWidget(btn_open, 0, 1)
grd.addWidget(self.ui_edit_fields_lbl[-1], 1, 0, 1, 2)
self.parent_field._internal_logger.info(f"Registered QEditField for {self.ui_edit_fields[-1]}")
self.parent_field._module_logger.info(f"Registered QEditField for {self.ui_edit_fields[-1]}")
wdg.setLayout(grd)
......
......@@ -31,14 +31,14 @@ class FieldViewString(FieldView):
le: QLineEdit = view
le.setToolTip(f"({self.parent_field.name}) {self.parent_field._description}")
self.ui_edit_fields.append(le)
self.parent_field._internal_logger.debug(f"Registering LineEdit {le}")
self.parent_field._module_logger.debug(f"Registering LineEdit {le}")
self.ui_edit_fields[-1].textEdited.connect(lambda d: self._on_text_edited(le, d))
# new
return le
def _on_text_edited(self, f, value):
self.parent_field._internal_logger.debug(f"LineEdit {f} changed to {value}.")
self.parent_field._module_logger.debug(f"LineEdit {f} changed to {value}.")
self.parent_field.set(value)
def _on_value_changed_partial(self, value):
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment