def get_logger(self, level : int) -> RootLogger:
formatter = logging.Formatter(fmt='%(levelname)s (%(threadName)-10s) :%(name)s: %(message)s '
'(%(asctime)s; %(filename)s:%(lineno)d)',
datefmt="%Y-%m-%d %H:%M:%S")
handlers = [
logging.handlers.RotatingFileHandler(self.__filename,
encoding='utf8',
maxBytes=self.__max_bytes,
backupCount=3),
logging.StreamHandler()
]
self.__root_logger = logging.getLogger()
self.__root_logger.setLevel(level)
for handler in handlers:
handler.setFormatter(formatter)
handler.setLevel(level)
self.__root_logger.addHandler(handler)
return self.__root_logger
python类RootLogger()的实例源码
def __init__(self, url, method='POST', fallback_file=None, loglevel=LOG_DEBUG, servicename=__default_servicename__):
import urlparse
self.loglevel = loglevel
self.fallback_file = fallback_file
p = urlparse.urlparse(url)
netloc = p.netloc
urlpath = p.path
if p.query:
urlpath += '?' + query
h = HTTPHandler(host=netloc, url=urlpath, method=method)
h.setLevel(loglevel)
fmt = logging.Formatter(servicename+"[%(process)d]: %(levelname)s: %(message)s")
h.setFormatter(fmt)
self._logger = RootLogger(loglevel)
self._logger.addHandler(h)
if self.fallback_file:
self._fallback = FileLogger(logfile=self.fallback_file,
loglevel=self.loglevel,
servicename=servicename)
else:
self._fallback = DummyLogger()
def root_logger(self) -> RootLogger:
logging_config = LoggingConfig(general.logging['log_file'], general.logging['log_entries'])
sys.excepthook = logging_config.set_error_hadler
return logging_config.get_logger(logging.INFO)
def __init__(self, change_actuator_request_event: ChangeActuatorRequestEvent,
text_communication_enhancer: TextCommunicationEnhancer,
sound_api : SoundApi, logging: RootLogger):
self.__change_actuator_request_event = change_actuator_request_event
self.__text_communication_enhancer = text_communication_enhancer
self.__sound_api = sound_api
self.__logging = logging
def __init__(self, ifttt_rules: IftttRulesRepository, command_executor: CommandExecutor,
tokenizer: Tokenizer, logging: RootLogger):
threading.Thread.__init__(self)
self.__ifttt_rules = ifttt_rules
self.__command_executor = command_executor
self.__tokenizer = tokenizer
self.__logging = logging
self.shutdown = False
def __init__(self, root_logger: RootLogger) -> None:
self.__root_logger = root_logger
self.__token_converters = []
def __init__(self, tokenizer: Tokenizer, logger: RootLogger):
self.__tokenizer = tokenizer
self.__current_token_index = 0
self.__logger = logger
def __init__(self, async_jobs: AsyncActuatorCommands, change_actuator_request_event: ChangeActuatorRequestEvent,
logging: RootLogger):
threading.Thread.__init__(self)
self.__async_jobs = async_jobs
self.__change_actuator_request_event = change_actuator_request_event
self.__logging = logging
self.shutdown = False
IncommingZwaveCommunicationThread.py 文件源码
项目:home-automation
作者: danionescu0
项目源码
文件源码
阅读 22
收藏 0
点赞 0
评论 0
def __init__(self, sensor_update_event: SensorUpdateEvent, sensors_repository: SensorsRepository,
actuators_repository: ActuatorsRepository, zwave_device: ZWaveDevice, logger: RootLogger):
threading.Thread.__init__(self)
self.__sensor_update_event = sensor_update_event
self.__sensors_repository = sensors_repository
self.__actuators_repository = actuators_repository
self.__zwave_device = zwave_device
self.__logger = logger
self.shutdown = False
def __init__(self, root_logger: RootLogger, port: str, openzwave_config_path: str) -> None:
self.__root_logger = root_logger
self.__port = port
self.__openzwave_config_path = openzwave_config_path
self.__network = None
self.__state_change_callback = None
SerialCommunicatorRegistry.py 文件源码
项目:home-automation
作者: danionescu0
项目源码
文件源码
阅读 21
收藏 0
点赞 0
评论 0
def __init__(self, configuration, logger: RootLogger):
self.__configuration = configuration
self.__logger = logger
self.__communicators = {}
def set_logger(self, logger: RootLogger):
self.__logger = logger
def get_logger(self) -> RootLogger:
return self.__logger
def __init__(self, root_logger: RootLogger) -> None:
self.__root_logger = root_logger
self.__env = None
def test_logger(self):
import livebridge.config
livebridge.config.LOGFILE = "/dev/null"
from livebridge.logger import logger
assert type(logger) is logging.RootLogger
assert type(logger.handlers[0]) is logging.StreamHandler
assert type(logger.handlers[1]) is logging.FileHandler
logger.handlers = []
def __init__(self, loglevel=LOG_DEBUG, servicename=__default_servicename__):
self.loglevel = loglevel
h = logging.StreamHandler()
h.setLevel(loglevel)
fmt = logging.Formatter("%(asctime)s "+servicename+"[%(process)d]: %(levelname)s: %(message)s")
h.setFormatter(fmt)
self._logger = RootLogger(loglevel)
self._logger.addHandler(h)
def __init__(self, addr='/dev/log', syslogfacility="local0", \
loglevel=LOG_DEBUG, servicename=__default_servicename__):
if ':' in addr:
host, port = addr.split(':', 1)
port = int(port)
addr = (host, port)
fac = Syslog.facility_names[syslogfacility]
h = Syslog(address=addr, facility=fac)
h.setLevel(loglevel)
fmt = logging.Formatter(servicename+"[%(process)d]: %(levelname)s: %(message)s")
h.setFormatter(fmt)
self._logger = RootLogger(loglevel)
self._logger.addHandler(h)
def __init__(self, logfile='/tmp/%s.log' % __default_servicename__, \
loglevel=LOG_DEBUG, servicename=__default_servicename__):
h = logging.FileHandler(filename=logfile)
h.setLevel(loglevel)
fmt = logging.Formatter("%(asctime)s "+servicename+"[%(process)d]: %(levelname)s: %(message)s")
h.setFormatter(fmt)
self._logger = RootLogger(loglevel)
self._logger.addHandler(h)
def main(connection, logger, content_view=None, all_views=False, keep=0, **kwargs):
"""
:type keep: int or bool
:type all_views: bool
:type connection: KatelloConnection
:type content_view: list
:param logger: Logger object
:type logger: logging.RootLogger
"""
if not keep:
keep = None
if all_views:
logger.debug('Going to clean all content views as instructed, keeping {} versions'.format(keep))
logger.debug('Getting all content views')
view_names = [view['name'] for view in connection.content_views]
logger.debug('Found: {} to clean'.format(all_views))
else:
try:
logger.debug('Checking if {} is defined in config file')
view_names = [c.strip() for c in kwargs['config_obj'].get(content_view[0], 'views').split(',')]
logger.debug('Yes, we are going to clean {}'.format(view_names))
except NoSectionError:
logger.debug('No, we are cleaning {}'.format(content_view))
view_names = content_view
for view_name in view_names:
try:
logger.debug('Getting info for {}'.format(view_name))
view_dict = get_components(connection.content_views, ('name', view_name))
except TypeError:
logger.error('View %s not found!' % view_name)
raise NotFoundError('View {} not found'.format(view_name))
ids_to_remove = [version['id'] for version in view_dict['versions'] if not version['environment_ids']]
try:
ids_to_remove = sorted(ids_to_remove)[:-keep]
except TypeError:
pass
logger.debug("We are going to remove the following versions id's: {}".format(ids_to_remove))
for version_id in ids_to_remove:
logging.info('Removing version_id %s' % version_id)
connection.remove_view_version(version_id)
def _configureLogger(name, force=False):
try:
logger = _Loggers[name]
except KeyError:
# add unknown logger names
logger = getLogger(name)
_debug("Added unknown logger name '%s'" % name)
# If not forcing, skip loggers that already have handlers installed
if not force and logger.handlers:
return
global _LogLevels
if not _LogLevels:
setLogLevels(getParam('GCAM.LogLevel') or 'WARN')
if name in _LogLevels:
level = _LogLevels[name]
_debug("Configuring %s, level=%s" % (name, level))
logger.setLevel(level)
else:
parent = logger.parent
# Handle case of user plugins, which aren't in pygcam package
# but we'd like them to default to the 'pygcam' loglevel
if isinstance(parent, logging.RootLogger):
parent = getLogger(PKGNAME)
logger.setLevel(parent.level)
logger.propagate = False
# flush and remove all handlers
for handler in logger.handlers:
if not isinstance(handler, logging.NullHandler):
handler.flush()
logger.removeHandler(handler)
logConsole = getParamAsBoolean('GCAM.LogConsole')
if logConsole:
consoleFormat = getParam('GCAM.LogConsoleFormat')
_addHandler(logger, consoleFormat)
logFile = getParam('GCAM.LogFile')
if logFile:
fileFormat = getParam('GCAM.LogFileFormat')
_addHandler(logger, fileFormat, logFile=logFile)
if not logger.handlers:
logger.addHandler(logger, logging.NullHandler())
_debug("Added NullHandler to root logger")