# SPDX-License-Identifier: Apache-2.0
# Copyright (C) 2020 ifm electronic gmbh
#
# THE PROGRAM IS PROVIDED "AS IS" WITHOUT WARRANTY OF ANY KIND.
#
"""
This module provides the gui logging service for nexxT.
"""
import datetime
from queue import Queue
import traceback
import logging
import nexxT.shiboken
from nexxT.Qt import __version_info__ as qtversion
from nexxT.Qt.QtCore import Qt, QTimer, QAbstractItemModel, QModelIndex, Slot
from nexxT.Qt.QtWidgets import QTableView, QHeaderView, QStyleOptionViewItem
from nexxT.Qt.QtGui import QColor, QAction, QActionGroup
from nexxT.services.ConsoleLogger import ConsoleLogger
from nexxT.interface import Services
from nexxT.core.Utils import assertMainThread
logger = logging.getLogger(__name__)
MAX_ENTRIES = 1500
MIN_ENTRIES = 1000
[docs]
class LogHandler(logging.Handler):
"""
Python logging handler which passes python log records to the gui.
"""
[docs]
def __init__(self, logView):
super().__init__()
self.logView = logView
[docs]
def emit(self, record):
"""
called when a new log record is created
:param record: a log record instance (see python docs)
:return:
"""
msg = record.getMessage()
if record.exc_info is not None:
msg += "\n" + "".join(traceback.format_exception(*record.exc_info))
if msg[-1] == "\n":
msg = msg[:-1]
items = (str(datetime.datetime.fromtimestamp(record.created)),
record.levelno,
msg,
record.name, record.filename, str(record.lineno))
self.logView.addLogRecord(items)
[docs]
class LogView(QTableView):
"""
Class implementing the GUI log display.
"""
[docs]
class LogModel(QAbstractItemModel):
"""
Model/view model for log entries. The entries are held in a python list.
"""
[docs]
def __init__(self):
super().__init__()
self.entries = []
self.singleLineMode = True
[docs]
def setSingleLineMode(self, enabled):
"""
called from the table view to indicate single line mode (in this mode only the last line of a log message
is displayed.
:param enabled: boolean
:return:
"""
self.singleLineMode = enabled
[docs]
def rowCount(self, parent):
"""
Overwritten from QAbstractItemModel
:param parent: a QModelIndex instance
:return: number of log entries
"""
if not parent.isValid():
return len(self.entries)
return 0
[docs]
def columnCount(self, parent): # pylint: disable=unused-argument
"""
Overwritten from QAbstractItemModel
:param parent: a QModelIndex instance
:return: the number of columns (constant)
"""
return 6
[docs]
def update(self, queue):
"""
add queued items to model
:param queue: a python Queue instance
:return: None
"""
toInsert = []
while not queue.empty():
items = queue.get()
toInsert.append(items)
if len(toInsert) > 0:
newCount = len(toInsert) + len(self.entries)
if newCount > MAX_ENTRIES:
remove = min(len(self.entries), newCount - MIN_ENTRIES)
self.beginRemoveRows(QModelIndex(), 0, remove - 1)
del self.entries[:remove]
self.endRemoveRows()
self.beginInsertRows(QModelIndex(), len(self.entries), len(self.entries) + len(toInsert) - 1)
self.entries.extend(toInsert)
self.endInsertRows()
[docs]
def clear(self):
"""
removes all entries from the list
:return: None
"""
if len(self.entries) > 0:
self.beginRemoveRows(QModelIndex(), 0, len(self.entries)-1)
self.entries = []
self.endRemoveRows()
[docs]
def index(self, row, column, parent):
"""
Overwritten from QAbstractItemModel
:param row: integer
:param column: integer
:param parent: a QModelIndex instance
:return: a QModelIndex for the specified item
"""
if not self.hasIndex(row, column, parent):
#print("index invalid", row, column, parent.isValid())
return QModelIndex()
if not parent.isValid():
#print("index valid", row, column, parent.isValid())
return self.createIndex(row, column)
#print("index invalid", row, column, parent.isValid())
return QModelIndex()
[docs]
def parent(self, index): # pylint: disable=unused-argument
"""
Overwritten from QAbstractItemModel
:param index:
:return: invalid model index (because we have a 2D table)
"""
return QModelIndex()
[docs]
def data(self, modelIndex, role):
"""
Overwritten from QAbstractItemModel
:param modelIndex: a QModelIndex instance
:param role: the role
:return: the requested data
"""
if not modelIndex.isValid():
return None
e = self.entries[modelIndex.row()]
if role in [Qt.DisplayRole, Qt.EditRole]:
if modelIndex.column() == 1:
levelno = e[1]
if levelno <= logging.INTERNAL:
return "INTERNAL"
if logging.INTERNAL < levelno <= logging.DEBUG:
return "DEBUG"
if logging.DEBUG < levelno <= logging.INFO:
return "INFO"
if logging.INFO < levelno <= logging.WARNING:
return "WARNING"
if logging.WARNING < levelno <= logging.ERROR:
return "ERROR"
return "CRITICAL"
if modelIndex.column() == 2 and self.singleLineMode:
msg = e[2]
if "\n" in msg:
msg = msg[msg.rfind("\n")+1:]
return msg
return e[modelIndex.column()]
if role == Qt.ToolTipRole:
return e[2]
if role == Qt.BackgroundRole:
levelno = e[1]
if levelno <= logging.INTERNAL:
return QColor(255, 255, 255) # white
if logging.INTERNAL < levelno <= logging.DEBUG:
return QColor(155, 155, 255) # blue
if logging.DEBUG < levelno <= logging.INFO:
return QColor(155, 255, 155) # green
if logging.INFO < levelno <= logging.WARNING:
return QColor(255, 255, 155) # yellow
if logging.WARNING < levelno <= logging.ERROR:
return QColor(255, 205, 155) # orange
return QColor(255, 155, 155) # red
return None
[docs]
def __init__(self):
super().__init__()
self.follow = True
self._model = self.LogModel()
self.setShowGrid(False)
self.queue = Queue()
self.setModel(self._model)
self._rowHeight = None
self._uniformRowHeights = True
self.horizontalHeader().setSectionResizeMode(0, QHeaderView.ResizeToContents)
self.horizontalHeader().setSectionResizeMode(1, QHeaderView.ResizeToContents)
self.horizontalHeader().setSectionResizeMode(2, QHeaderView.Stretch)
self.horizontalHeader().setSectionResizeMode(3, QHeaderView.ResizeToContents)
self.horizontalHeader().setSectionResizeMode(4, QHeaderView.ResizeToContents)
self.horizontalHeader().setSectionResizeMode(5, QHeaderView.ResizeToContents)
self.horizontalHeader().setStretchLastSection(False)
self.verticalHeader().setSectionResizeMode(QHeaderView.ResizeToContents)
self.timer = QTimer()
self.timer.setSingleShot(False)
self.timer.start(100)
self.timer.timeout.connect(self.update, Qt.QueuedConnection)
[docs]
def sizeHintForRow(self, row):
"""
return a size hint for the given row index
:param row: the row index as integer
:return: the height of the row as integer
"""
if self._uniformRowHeights:
if row < 0 or row >= self._model.rowCount(QModelIndex()):
# Mirror super implementation.
return -1
return self._getRowHeight()
return super().sizeHintForRow(row)
def _getRowHeight(self):
if self._rowHeight is None:
self._rowHeight = max(self._getCellHeights())
return self._rowHeight
[docs]
def changeEvent(self, event):
"""
This for instance happens when the style sheet changed. It may affect
the calculated row height. So invalidate:
:param event: the event causing the change
:return:
"""
self._rowHeight = None
super().changeEvent(event)
def _getCellHeights(self, row=0):
self.ensurePolished()
if qtversion[0] == 5:
option = self.viewOptions()
else:
option = QStyleOptionViewItem()
self.initViewItemOption(option)
model = self._model
for column in range(model.columnCount(QModelIndex())):
index = model.index(row, column, QModelIndex())
delegate = self.itemDelegate(index)
if delegate:
yield delegate.sizeHint(option, index).height()
[docs]
def addLogRecord(self, items):
"""
Add a log record to the synchronized queue
:param items: a tuple of (timestamp[str], level[int], message[str], modulename[str], filename[str], lineno[int])
:return: None
"""
self.queue.put(items)
[docs]
def update(self):
"""
Called periodically to synchronize model with added log records
:return: None
"""
assertMainThread()
if not nexxT.shiboken.isValid(self): # pylint: disable=no-member
return
if not self.queue.empty():
self._model.update(self.queue)
if self.follow:
self.scrollToBottom()
[docs]
def setFollow(self, follow):
"""
set follow mode
:param follow: a boolean
:return: None
"""
self.follow = follow
[docs]
def clear(self):
"""
Clears the view
:return: None
"""
self._model.clear()
[docs]
class GuiLogger(ConsoleLogger):
"""
Logging service in GUI mode.
"""
[docs]
def __init__(self):
super().__init__()
srv = Services.getService("MainWindow")
self.dockWidget = srv.newDockWidget("Log", parent=None,
defaultArea=Qt.BottomDockWidgetArea,
allowedArea=Qt.LeftDockWidgetArea | Qt.BottomDockWidgetArea)
self.logWidget = LogView()
self.dockWidget.setWidget(self.logWidget)
logMenu = srv.menuBar().addMenu("&Log")
mainLogger = logging.getLogger()
self.handler = LogHandler(self.logWidget)
mainLogger.addHandler(self.handler)
self.destroyed.connect(self.removeHandler)
self.actFollow = QAction("Follow")
self.actFollow.setCheckable(True)
self.actFollow.setChecked(True)
self.actClear = QAction("Clear")
self.actSingleLine = QAction("Single Line")
self.actSingleLine.setCheckable(True)
self.actSingleLine.setChecked(True)
self.logWidget.setUniformRowHeights(True)
self.actFollow.toggled.connect(self.logWidget.setFollow)
self.actClear.triggered.connect(self.logWidget.clear)
self.actSingleLine.toggled.connect(self.logWidget.setUniformRowHeights)
self._populateLogLevelMenu(logMenu)
self._populateLogCustomMenu(logMenu)
self._populateLogMaxEntryMenu(logMenu)
def _populateLogLevelMenu(self, logMenu):
self.actDisable = QAction("Disable")
self.actDisable.setData(100)
self.actDisable.setCheckable(True)
self.actDisable.triggered.connect(self.setLogLevel)
mainLogger = logging.getLogger()
actGroup = QActionGroup(self)
actGroup.setExclusive(True)
levelno = mainLogger.level
for lv in ["INTERNAL", "DEBUG", "INFO", "WARNING", "ERROR"]:
a = QAction(lv[:1] + lv[1:].lower())
a.setCheckable(True)
loglevel = getattr(logging, lv)
a.setData(loglevel)
setattr(self, "setLogLevel_" + lv, self.setLogLevel)
a.triggered.connect(getattr(self, "setLogLevel_" + lv))
actGroup.addAction(a)
if levelno == loglevel:
a.setChecked(True)
else:
a.setChecked(False)
logMenu.addAction(a)
logMenu.addAction(self.actDisable)
actGroup.addAction(self.actDisable)
def _populateLogCustomMenu(self, logMenu):
logMenu.addSeparator()
logMenu.addAction(self.actClear)
logMenu.addAction(self.actFollow)
logMenu.addAction(self.actSingleLine)
def _populateLogMaxEntryMenu(self, logMenu):
logMenu.addSeparator()
nlogGroup = QActionGroup(self)
nlogGroup.setExclusive(True)
for maxNumLogentries in [100, 1000, "unlimited"]:
a = QAction(f"Show {maxNumLogentries} entries")
a.setData(maxNumLogentries)
a.setCheckable(True)
setattr(self, f"setNumEntries_{maxNumLogentries}", self.setNumEntries)
a.triggered.connect(getattr(self, f"setNumEntries_{maxNumLogentries}"))
a.setChecked(maxNumLogentries == MIN_ENTRIES)
nlogGroup.addAction(a)
logMenu.addAction(a)
[docs]
@Slot()
def detach(self):
"""
This slot is called when the service is removed from the framework
"""
self.removeHandler()
[docs]
def removeHandler(self):
"""
Remove logging handler from python logging system.
"""
mainLogger = logging.getLogger()
if self.handler is not None:
mainLogger.removeHandler(self.handler)
self.handler = None
[docs]
def setLogLevel(self):
"""
Sets the current log level from the calling action.
:return: None
"""
lv = self.sender().data()
logging.getLogger().setLevel(lv)
[docs]
def setNumEntries(self):
"""
Sets the number of displayed log entries from the calling action.
"""
global MAX_ENTRIES, MIN_ENTRIES
maxNumEntries = self.sender().data()
if maxNumEntries != "unlimited":
MIN_ENTRIES = maxNumEntries
MAX_ENTRIES = int(maxNumEntries*1.1)
else:
MIN_ENTRIES = 1 << 31
MAX_ENTRIES = 1 << 32
logger.info("Number of displayed log entries=%s", maxNumEntries)