TheCloud/panda3d
1
0
Fork 0
mirror of https://github.com/panda3d/panda3d.git synced 2025-09-27 23:34:57 -04:00
Code Issues Packages Projects Releases Wiki Activity

directnotify: annotate types (#1527)

Browse Source
This commit is contained in:
WMOkiishi 2023-10-09 01:46:32 -06:00 committed by GitHub
parent 526994c302
commit 1ca0e3f1ea
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
9 changed files with 333 additions and 96 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

62
direct/src/directnotify/DirectNotify.py
View File

@ -2,6 +2,10 @@
DirectNotify module: this module contains the DirectNotify class
"""
from __future__ import annotations
from panda3d.core import StreamWriter
from . import Notifier
from . import Logger
@ -12,39 +16,39 @@ class DirectNotify:
mulitple notify categories via a dictionary of Notifiers.
"""
def __init__(self):
def __init__(self) -> None:
"""
DirectNotify class keeps a dictionary of Notfiers
"""
self.__categories = {}
self.__categories: dict[str, Notifier.Notifier] = {}
# create a default log file
self.logger = Logger.Logger()
# This will get filled in later by ShowBase.py with a
# C++-level StreamWriter object for writing to standard
# output.
self.streamWriter = None
self.streamWriter: StreamWriter | None = None
def __str__(self):
def __str__(self) -> str:
"""
Print handling routine
"""
return "DirectNotify categories: %s" % (self.__categories)
#getters and setters
def getCategories(self):
def getCategories(self) -> list[str]:
"""
Return list of category dictionary keys
"""
return list(self.__categories.keys())
def getCategory(self, categoryName):
def getCategory(self, categoryName: str) -> Notifier.Notifier | None:
"""getCategory(self, string)
Return the category with given name if present, None otherwise
"""
return self.__categories.get(categoryName, None)
def newCategory(self, categoryName, logger=None):
def newCategory(self, categoryName: str, logger: Logger.Logger | None = None) -> Notifier.Notifier:
"""newCategory(self, string)
Make a new notify category named categoryName. Return new category
if no such category exists, else return existing category
@ -52,9 +56,11 @@ class DirectNotify:
if categoryName not in self.__categories:
self.__categories[categoryName] = Notifier.Notifier(categoryName, logger)
self.setDconfigLevel(categoryName)
return self.getCategory(categoryName)
notifier = self.getCategory(categoryName)
assert notifier is not None
return notifier
def setDconfigLevel(self, categoryName):
def setDconfigLevel(self, categoryName: str) -> None:
"""
Check to see if this category has a dconfig variable
to set the notify severity and then set that level. You cannot
@ -77,40 +83,42 @@ class DirectNotify:
level = 'error'
category = self.getCategory(categoryName)
assert category is not None, f'failed to find category: {categoryName!r}'
# Note - this print statement is making it difficult to
# achieve "no output unless there's an error" operation - Josh
# print ("Setting DirectNotify category: " + categoryName +
# " to severity: " + level)
if level == "error":
category.setWarning(0)
category.setInfo(0)
category.setDebug(0)
category.setWarning(False)
category.setInfo(False)
category.setDebug(False)
elif level == "warning":
category.setWarning(1)
category.setInfo(0)
category.setDebug(0)
category.setWarning(True)
category.setInfo(False)
category.setDebug(False)
elif level == "info":
category.setWarning(1)
category.setInfo(1)
category.setDebug(0)
category.setWarning(True)
category.setInfo(True)
category.setDebug(False)
elif level == "debug":
category.setWarning(1)
category.setInfo(1)
category.setDebug(1)
category.setWarning(True)
category.setInfo(True)
category.setDebug(True)
else:
print("DirectNotify: unknown notify level: " + str(level)
+ " for category: " + str(categoryName))
def setDconfigLevels(self):
def setDconfigLevels(self) -> None:
for categoryName in self.getCategories():
self.setDconfigLevel(categoryName)
def setVerbose(self):
def setVerbose(self) -> None:
for categoryName in self.getCategories():
category = self.getCategory(categoryName)
category.setWarning(1)
category.setInfo(1)
category.setDebug(1)
assert category is not None
category.setWarning(True)
category.setInfo(True)
category.setDebug(True)
def popupControls(self, tl = None):
# Don't use a regular import, to prevent ModuleFinder from picking
@ -119,5 +127,5 @@ class DirectNotify:
NotifyPanel = importlib.import_module('direct.tkpanels.NotifyPanel')
NotifyPanel.NotifyPanel(self, tl)
def giveNotify(self,cls):
def giveNotify(self, cls) -> None:
cls.notify = self.newCategory(cls.__name__)

24
direct/src/directnotify/Logger.py
View File

@ -1,27 +1,30 @@
"""Logger module: contains the logger class which creates and writes
data to log files on disk"""
from __future__ import annotations
import io
import time
import math
class Logger:
def __init__(self, fileName="log"):
def __init__(self, fileName: str = "log") -> None:
"""
Logger constructor
"""
self.__timeStamp = 1
self.__timeStamp = True
self.__startTime = 0.0
self.__logFile = None
self.__logFile: io.TextIOWrapper | None = None
self.__logFileName = fileName
def setTimeStamp(self, enable):
def setTimeStamp(self, enable: bool) -> None:
"""
Toggle time stamp printing with log entries on and off
"""
self.__timeStamp = enable
def getTimeStamp(self):
def getTimeStamp(self) -> bool:
"""
Return whether or not we are printing time stamps with log entries
"""
@ -29,24 +32,25 @@ class Logger:
# logging control
def resetStartTime(self):
def resetStartTime(self) -> None:
"""
Reset the start time of the log file for time stamps
"""
self.__startTime = time.time()
def log(self, entryString):
def log(self, entryString: str) -> None:
"""log(self, string)
Print the given string to the log file"""
if self.__logFile is None:
self.__openLogFile()
assert self.__logFile is not None
if self.__timeStamp:
self.__logFile.write(self.__getTimeStamp())
self.__logFile.write(entryString + '\n')
# logging functions
def __openLogFile(self):
def __openLogFile(self) -> None:
"""
Open a file for logging error/warning messages
"""
@ -56,14 +60,14 @@ class Logger:
logFileName = self.__logFileName + "." + st
self.__logFile = open(logFileName, "w")
def __closeLogFile(self):
def __closeLogFile(self) -> None:
"""
Close the error/warning output file
"""
if self.__logFile is not None:
self.__logFile.close()
def __getTimeStamp(self):
def __getTimeStamp(self) -> str:
"""
Return the offset between current time and log file startTime
"""

83
direct/src/directnotify/Notifier.py
View File

@ -2,11 +2,16 @@
Notifier module: contains methods for handling information output
for the programmer/user
"""
from __future__ import annotations
from .Logger import Logger
from .LoggerGlobal import defaultLogger
from direct.showbase import PythonUtil
from panda3d.core import ConfigVariableBool, NotifyCategory, StreamWriter, Notify
import time
import sys
from typing import NoReturn
class NotifierException(Exception):
@ -20,13 +25,13 @@ class Notifier:
# messages instead of writing them to the console. This is
# particularly useful for integrating the Python notify system
# with the C++ notify system.
streamWriter = None
streamWriter: StreamWriter | None = None
if ConfigVariableBool('notify-integrate', True):
streamWriter = StreamWriter(Notify.out(), False)
showTime = ConfigVariableBool('notify-timestamp', False)
def __init__(self, name, logger=None):
def __init__(self, name: str, logger: Logger | None = None) -> None:
"""
Parameters:
name (str): a string name given to this Notifier instance.
@ -42,12 +47,12 @@ class Notifier:
self.__logger = logger
# Global default levels are initialized here
self.__info = 1
self.__warning = 1
self.__debug = 0
self.__logging = 0
self.__info = True
self.__warning = True
self.__debug = False
self.__logging = False
def setServerDelta(self, delta, timezone):
def setServerDelta(self, delta: float, timezone: int) -> None:
"""
Call this method on any Notify object to globally change the
timestamp printed for each line of all Notify objects.
@ -65,7 +70,7 @@ class Notifier:
self.info("Notify clock adjusted by %s (and timezone adjusted by %s hours) to synchronize with server." % (PythonUtil.formatElapsedSeconds(delta), (time.timezone - timezone) / 3600))
def getTime(self):
def getTime(self) -> str:
"""
Return the time as a string suitable for printing at the
head of any notify message
@ -74,14 +79,14 @@ class Notifier:
# the task is out of focus on win32. time.clock doesn't have this problem.
return time.strftime(":%m-%d-%Y %H:%M:%S ", time.localtime(time.time() + self.serverDelta))
def getOnlyTime(self):
def getOnlyTime(self) -> str:
"""
Return the time as a string.
The Only in the name is referring to not showing the date.
"""
return time.strftime("%H:%M:%S", time.localtime(time.time() + self.serverDelta))
def __str__(self):
def __str__(self) -> str:
"""
Print handling routine
"""
@ -89,26 +94,26 @@ class Notifier:
(self.__name, self.__info, self.__warning, self.__debug, self.__logging)
# Severity funcs
def setSeverity(self, severity):
def setSeverity(self, severity: int) -> None:
from panda3d.core import NSDebug, NSInfo, NSWarning, NSError
if severity >= NSError:
self.setWarning(0)
self.setInfo(0)
self.setDebug(0)
self.setWarning(False)
self.setInfo(False)
self.setDebug(False)
elif severity == NSWarning:
self.setWarning(1)
self.setInfo(0)
self.setDebug(0)
self.setWarning(True)
self.setInfo(False)
self.setDebug(False)
elif severity == NSInfo:
self.setWarning(1)
self.setInfo(1)
self.setDebug(0)
self.setWarning(True)
self.setInfo(True)
self.setDebug(False)
elif severity <= NSDebug:
self.setWarning(1)
self.setInfo(1)
self.setDebug(1)
self.setWarning(True)
self.setInfo(True)
self.setDebug(True)
def getSeverity(self):
def getSeverity(self) -> int:
from panda3d.core import NSDebug, NSInfo, NSWarning, NSError
if self.getDebug():
return NSDebug
@ -120,7 +125,7 @@ class Notifier:
return NSError
# error funcs
def error(self, errorString, exception=NotifierException):
def error(self, errorString: object, exception: type[Exception] = NotifierException) -> NoReturn:
"""
Raise an exception with given string and optional type:
Exception: error
@ -134,7 +139,7 @@ class Notifier:
raise exception(errorString)
# warning funcs
def warning(self, warningString):
def warning(self, warningString: object) -> int:
"""
Issue the warning message if warn flag is on
"""
@ -148,20 +153,20 @@ class Notifier:
self.__print(string)
return 1 # to allow assert myNotify.warning("blah")
def setWarning(self, enable):
def setWarning(self, enable: bool) -> None:
"""
Enable/Disable the printing of warning messages
"""
self.__warning = enable
def getWarning(self):
def getWarning(self) -> bool:
"""
Return whether the printing of warning messages is on or off
"""
return self.__warning
# debug funcs
def debug(self, debugString):
def debug(self, debugString: object) -> int:
"""
Issue the debug message if debug flag is on
"""
@ -175,20 +180,20 @@ class Notifier:
self.__print(string)
return 1 # to allow assert myNotify.debug("blah")
def setDebug(self, enable):
def setDebug(self, enable: bool) -> None:
"""
Enable/Disable the printing of debug messages
"""
self.__debug = enable
def getDebug(self):
def getDebug(self) -> bool:
"""
Return whether the printing of debug messages is on or off
"""
return self.__debug
# info funcs
def info(self, infoString):
def info(self, infoString: object) -> int:
"""
Print the given informational string, if info flag is on
"""
@ -202,39 +207,39 @@ class Notifier:
self.__print(string)
return 1 # to allow assert myNotify.info("blah")
def getInfo(self):
def getInfo(self) -> bool:
"""
Return whether the printing of info messages is on or off
"""
return self.__info
def setInfo(self, enable):
def setInfo(self, enable: bool) -> None:
"""
Enable/Disable informational message printing
"""
self.__info = enable
# log funcs
def __log(self, logEntry):
def __log(self, logEntry: str) -> None:
"""
Determine whether to send informational message to the logger
"""
if self.__logging:
self.__logger.log(logEntry)
def getLogging(self):
def getLogging(self) -> bool:
"""
Return 1 if logging enabled, 0 otherwise
"""
return self.__logging
def setLogging(self, enable):
def setLogging(self, enable: bool) -> None:
"""
Set the logging flag to int (1=on, 0=off)
"""
self.__logging = enable
def __print(self, string):
def __print(self, string: str) -> None:
"""
Prints the string to output followed by a newline.
"""
@ -285,7 +290,7 @@ class Notifier:
self.__print(string)
return 1 # to allow assert self.notify.debugStateCall(self)
def debugCall(self, debugString=''):
def debugCall(self, debugString: object = '') -> int:
"""
If this notify is in debug mode, print the time of the
call followed by the notifier category and

48
direct/src/directnotify/RotatingLog.py
View File

@ -1,5 +1,8 @@
from __future__ import annotations
import os
import time
from typing import Iterable
class RotatingLog:
@ -8,7 +11,12 @@ class RotatingLog:
to a new file if the prior file is too large or after a time interval.
"""
def __init__(self, path="./log_file", hourInterval=24, megabyteLimit=1024):
def __init__(
self,
path: str = "./log_file",
hourInterval: int | None = 24,
megabyteLimit: int | None = 1024,
) -> None:
"""
Args:
path: a full or partial path with file name.
@ -28,33 +36,33 @@ class RotatingLog:
if megabyteLimit is not None:
self.sizeLimit = megabyteLimit*1024*1024
def __del__(self):
def __del__(self) -> None:
self.close()
def close(self):
def close(self) -> None:
if hasattr(self, "file"):
self.file.flush()
self.file.close()
self.closed = self.file.closed
del self.file
else:
self.closed = 1
self.closed = True
def shouldRotate(self):
def shouldRotate(self) -> bool:
"""
Returns a bool about whether a new log file should
be created and written to (while at the same time
stopping output to the old log file and closing it).
"""
if not hasattr(self, "file"):
return 1
return True
if self.timeLimit is not None and time.time() > self.timeLimit:
return 1
return True
if self.sizeLimit is not None and self.file.tell() > self.sizeLimit:
return 1
return 0
return True
return False
def filePath(self):
def filePath(self) -> str:
dateString = time.strftime("%Y_%m_%d_%H", time.localtime())
for i in range(26):
limit = self.sizeLimit
@ -65,7 +73,7 @@ class RotatingLog:
# Maybe we should clear the self.sizeLimit here... maybe.
return path
def rotate(self):
def rotate(self) -> None:
"""
Rotate the log now. You normally shouldn't need to call this.
See write().
@ -88,12 +96,13 @@ class RotatingLog:
#self.newlines = self.file.newlines # Python 2.3, maybe
if self.timeLimit is not None and time.time() > self.timeLimit:
assert self.timeInterval is not None
self.timeLimit=time.time()+self.timeInterval
else:
# We'll keep writing to the old file, if available.
print("RotatingLog error: Unable to open new log file \"%s\"." % (path,))
def write(self, data):
def write(self, data: str) -> int | None:
"""
Write the data to either the current log or a new one,
depending on the return of shouldRotate() and whether
@ -105,14 +114,15 @@ class RotatingLog:
r = self.file.write(data)
self.file.flush()
return r
return None
def flush(self):
def flush(self) -> None:
return self.file.flush()
def fileno(self):
def fileno(self) -> int:
return self.file.fileno()
def isatty(self):
def isatty(self) -> bool:
return self.file.isatty()
def __next__(self):
@ -131,14 +141,14 @@ class RotatingLog:
def xreadlines(self):
return self.file.xreadlines()
def seek(self, offset, whence=0):
def seek(self, offset: int, whence: int = 0) -> int:
return self.file.seek(offset, whence)
def tell(self):
def tell(self) -> int:
return self.file.tell()
def truncate(self, size):
def truncate(self, size: int | None) -> int:
return self.file.truncate(size)
def writelines(self, sequence):
def writelines(self, sequence: Iterable[str]) -> None:
return self.file.writelines(sequence)

2
direct/src/distributed/DistributedCartesianGrid.py
View File

@ -23,7 +23,7 @@ GRID_Z_OFFSET = 0.0
class DistributedCartesianGrid(DistributedNode, CartesianGridBase):
notify = directNotify.newCategory("DistributedCartesianGrid")
notify.setDebug(0)
notify.setDebug(False)
VisualizeGrid = ConfigVariableBool("visualize-cartesian-grid", False)

57
tests/directnotify/test_DirectNotify.py Normal file
View File

@ -0,0 +1,57 @@
import pytest
from panda3d import core
from direct.directnotify import DirectNotify, Logger, Notifier
CATEGORY_NAME = 'test'
@pytest.fixture
def notify():
notify = DirectNotify.DirectNotify()
notify.newCategory(CATEGORY_NAME)
return notify
def test_categories():
notify = DirectNotify.DirectNotify()
assert len(notify.getCategories()) == 0
assert notify.getCategory(CATEGORY_NAME) is None
notifier = notify.newCategory(CATEGORY_NAME, logger=Logger.Logger())
assert isinstance(notifier, Notifier.Notifier)
assert notify.getCategories() == [CATEGORY_NAME]
def test_setDconfigLevels(notify):
config = core.ConfigVariableString('notify-level-' + CATEGORY_NAME, '')
notifier = notify.getCategory(CATEGORY_NAME)
config.value = 'error'
notify.setDconfigLevels()
assert notifier.getSeverity() == core.NS_error
config.value = 'warning'
notify.setDconfigLevels()
assert notifier.getSeverity() == core.NS_warning
config.value = 'info'
notify.setDconfigLevels()
assert notifier.getSeverity() == core.NS_info
config.value = 'debug'
notify.setDconfigLevels()
assert notifier.getSeverity() == core.NS_debug
def test_setVerbose(notify):
notifier = notify.getCategory(CATEGORY_NAME)
notifier.setWarning(False)
notifier.setInfo(False)
notifier.setDebug(False)
notify.setVerbose()
assert notifier.getWarning()
assert notifier.getInfo()
assert notifier.getDebug()
def test_giveNotify(notify):
class HasNotify:
notify = None
notify.giveNotify(HasNotify)
assert isinstance(HasNotify.notify, Notifier.Notifier)

21
tests/directnotify/test_Logger.py Normal file
View File

@ -0,0 +1,21 @@
import re
from direct.directnotify import Logger
LOG_TEXT = 'Arbitrary log text'
def test_logging(tmp_path):
log_filename = str(tmp_path / 'log')
logger = Logger.Logger(log_filename)
assert logger.getTimeStamp()
logger.log(LOG_TEXT)
logger.setTimeStamp(False)
assert not logger.getTimeStamp()
logger.log(LOG_TEXT)
logger._Logger__closeLogFile()
log_file, = tmp_path.iterdir()
log_text = log_file.read_text()
pattern = rf'\d\d:\d\d:\d\d:\d\d: {LOG_TEXT}\n{LOG_TEXT}\n'
assert re.match(pattern, log_text)

87
tests/directnotify/test_Notifier.py Normal file
View File

@ -0,0 +1,87 @@
import io
import re
import time
import pytest
from panda3d import core
from direct.directnotify import Logger, Notifier
NOTIFIER_NAME = 'Test notifier'
DEBUG_LOG = 'Debug log'
INFO_LOG = 'Info log'
WARNING_LOG = 'Warning log'
ERROR_LOG = 'Error log'
@pytest.fixture
def log_io():
return io.StringIO()
@pytest.fixture
def notifier(log_io):
logger = Logger.Logger()
logger.setTimeStamp(False)
logger._Logger__logFile = log_io
notifier = Notifier.Notifier(NOTIFIER_NAME, logger)
notifier.setLogging(True)
return notifier
def test_setServerDelta():
notifier = Notifier.Notifier(NOTIFIER_NAME)
notifier.setServerDelta(4.2, time.timezone)
assert Notifier.Notifier.serverDelta == 4
Notifier.Notifier.serverDelta = 0
def test_logging(notifier, log_io):
notifier.setLogging(False)
assert not notifier.getLogging()
notifier.info(INFO_LOG)
assert log_io.getvalue() == ''
notifier.setLogging(True)
assert notifier.getLogging()
notifier.info(INFO_LOG)
assert log_io.getvalue() == f':{NOTIFIER_NAME}: {INFO_LOG}\n'
@pytest.mark.parametrize('severity', (core.NS_debug, core.NS_info, core.NS_warning, core.NS_error))
def test_severity(severity, notifier, log_io):
notifier.setSeverity(severity)
assert notifier.getSeverity() == severity
with pytest.raises(Notifier.NotifierException):
notifier.error(ERROR_LOG)
warning_return = notifier.warning(WARNING_LOG)
info_return = notifier.info(INFO_LOG)
debug_return = notifier.debug(DEBUG_LOG)
assert warning_return and info_return and debug_return
expected_logs = [
f'{Notifier.NotifierException}: {NOTIFIER_NAME}(error): {ERROR_LOG}',
f':{NOTIFIER_NAME}(warning): {WARNING_LOG}',
f':{NOTIFIER_NAME}: {INFO_LOG}',
f':{NOTIFIER_NAME}(debug): {DEBUG_LOG}',
]
del expected_logs[6-severity:]
assert log_io.getvalue() == '\n'.join(expected_logs) + '\n'
def test_custom_exception(notifier):
class CustomException(Exception):
pass
with pytest.raises(CustomException):
notifier.error(ERROR_LOG, CustomException)
def test_debugCall(notifier, log_io):
notifier.setDebug(False)
return_value = notifier.debugCall(DEBUG_LOG)
assert return_value
assert log_io.getvalue() == ''
notifier.setDebug(True)
notifier.debugCall(DEBUG_LOG)
pattern = rf':\d\d:\d\d:\d\d:{NOTIFIER_NAME} "{DEBUG_LOG}" test_debugCall\(.*\)\n'
assert re.match(pattern, log_io.getvalue())

45
tests/directnotify/test_RotatingLog.py Normal file
View File

@ -0,0 +1,45 @@
import pytest
from direct.directnotify import RotatingLog
LOG_TEXT = 'Arbitrary log text'
@pytest.fixture
def log_dir(tmp_path):
log_dir = tmp_path / 'logs'
log_dir.mkdir()
return log_dir
@pytest.fixture
def rotating_log(log_dir):
log_filename = str(log_dir / 'log')
rotating_log = RotatingLog.RotatingLog(log_filename)
yield rotating_log
rotating_log.close()
def test_rotation(rotating_log, log_dir):
rotating_log.sizeLimit = -1
rotating_log.write('1')
rotating_log.write('2')
written = [f.read_text() for f in log_dir.iterdir()]
assert written == ['1', '2'] or written == ['2', '1']
def test_wrapper_methods(rotating_log, log_dir):
rotating_log.write('')
log_file, = log_dir.iterdir()
assert rotating_log.fileno() == rotating_log.file.fileno()
assert rotating_log.isatty() == rotating_log.file.isatty()
rotating_log.writelines([LOG_TEXT] * 10)
assert not log_file.read_text()
rotating_log.flush()
assert log_file.read_text() == LOG_TEXT * 10
assert rotating_log.tell() == len(LOG_TEXT) * 10
rotating_log.seek(len(LOG_TEXT))
rotating_log.truncate(None)
assert log_file.read_text() == LOG_TEXT