"""
Logging config
"""
import logging
import logging.config
import socket
import threading
from pathlib import Path
logger = logging.getLogger(__name__)
[docs]def logConfig(level=logging.INFO, filename=None, remote=False):
"""
Configs logging for seq module
Args:
level(int): logging level.
filename(str): logging file name
Returns:
None
"""
logger = logging.getLogger(__name__)
if remote:
d = LOGGING_REMOTE
else:
d = LOGGING
d["loggers"]["seq"]["level"] = level
if filename is None:
filename='seq.log'
p = Path(filename)
p = p.with_name('seq_user.log')
if ('filename' not in d["handlers"]["logfile"]):
d["handlers"]["logfile"]["filename"] = filename
if ('filename' not in d["handlers"]["logfileUser"]):
d["handlers"]["logfileUser"]["filename"] = p.as_posix()
logging.config.dictConfig(d)
seq_logger = logging.getLogger("seq")
seq_logger.info("seq logger initialized")
seq_lib_logger = logging.getLogger("seq.lib")
seq_lib_logger.info("seq.lib logger initialized")
user_logger = logging.getLogger("seq.user")
logger.info('log filename: %s', d["handlers"]["logfile"]["filename"])
user_logger.info("User's Logging initialized")
[docs]def getUserLogger():
"""Gets the user's Logger."""
return logging.getLogger("seq.user")
LOGGING = {
"version": 1,
'disable_existing_loggers': True,
"formatters": {
"simple": {
"format": "%(levelname)s:(%(module)s.%(funcName)s): %(message)s"
},
"verbose": {
"format": "%(asctime)s %(levelname)s:(%(module)s.%(funcName)s):%(lineno)-d: %(message)s"
},
"timed": {
"format": "%(asctime)s %(levelname)s: %(message)s",
},
},
"handlers": {
"console": {
"class": "logging.StreamHandler",
"formatter": "simple",
},
"logfile": {
"class": "logging.FileHandler",
"formatter": "timed",
# "filename": "dummy.log",
# "filename": "%(filename)s",
"mode": "w",
},
"logfileUser": {
"class": "logging.FileHandler",
"formatter": "timed",
# "filename": "dummy.log",
# "filename": "%(filename)s",
"mode": "w",
},
},
"loggers": {
"seq": {
"level": logging.INFO,
"handlers": ["console", "logfile"],
"propagate": False,
"formatter": "simple",
},
"seq.lib": {
"level": logging.INFO,
"handlers": ["logfile"],
"propagate": False,
"formatter": "timed",
},
"seq.user": {
"level": logging.INFO,
"handlers": ["logfileUser"],
"propagate": False,
"formatter": "timed",
},
},
}
LOGGING_REMOTE = {
"version": 1,
'disable_existing_loggers': True,
"formatters": {
"simple": {
"format": "%(levelname)s:(%(module)s.%(funcName)s): %(message)s",
"datefmt": "%Y-%m-%dT%H:%M:%S%"
},
"verbose": {
"format": "%(asctime)s.%(msecs)03d %(levelname)s:(%(module)s.%(funcName)s):%(lineno)-d: %(message)s",
"datefmt": "%Y-%m-%dT%H:%M:%S%"
},
"timed": {
"format": "%(asctime)s.%(msecs)03d %(levelname)s: %(message)s",
"datefmt": "%Y-%m-%dT%H:%M:%S"
},
},
"handlers": {
"console": {
"class": "logging.StreamHandler",
"formatter": "timed",
},
"logfile": {
"class": "logging.FileHandler",
"formatter": "timed",
# "filename": "dummy.log",
# "filename": "%(filename)s",
"mode": "w",
},
"logfileUser": {
"class": "logging.FileHandler",
"formatter": "timed",
# "filename": "dummy.log",
# "filename": "%(filename)s",
"mode": "w",
},
"socket": {
"class": "seq.lib.ListeningSocketHandler",
"formatter": "verbose",
"port": 8113,
"ipv6": False,
},
},
"loggers": {
"": { # Allows to use any logger name in sequencer scripts
"level": logging.INFO,
"handlers": ["console", "socket"],
"propagate": False,
"formatter": "timed",
},
"seq": {
"level": logging.INFO,
"handlers": ["console", "logfile", "socket"],
"propagate": False,
"formatter": "timed",
},
"seq.lib": {
"level": logging.INFO,
"handlers": ["logfile", "socket"],
"propagate": False,
"formatter": "timed",
},
"seq.user": {
"level": logging.INFO,
"handlers": ["logfileUser", "socket"],
"propagate": False,
"formatter": "timed",
},
},
}
# Workaround for http://bugs.python.org/issue14308
# http://stackoverflow.com/questions/13193278/understand-python-threading-bug
threading._DummyThread._Thread__stop = lambda x: 42
[docs]class ListeningSocketHandler(logging.Handler):
"""
ListeningSocketHandler
======================
A python logging handler.
logging.handlers.SocketHandler is a TCP Socket client that sends log
records to a tcp server. This class is the opposite.
When a TCP client connects (e.g. telnet or netcat), log records are
streamed through the connection.
Author: Ben Cordero
Source: https://github.com/bencord0/ListeningSocketHandler
License: APL2.0
https://github.com/bencord0/ListeningSocketHandler/blob/master/LICENSE.txt
"""
def __init__(self, port=8113, ipv6=False):
super(ListeningSocketHandler, self).__init__()
self.port = port
self.ipv6 = ipv6
self.clients = set()
if self.ipv6:
self.socket = socket.socket(socket.AF_INET6)
self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
self.socket.bind(("::", self.port))
else:
self.socket = socket.socket(socket.AF_INET)
self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
self.socket.bind(("0.0.0.0", self.port))
self.socket.listen(5)
print ("ListeningSocketHandler on port: {}".format(self.socket.getsockname()[1]))
def start_accepting(self):
while True:
conn, addr = self.socket.accept()
self.clients.add(conn)
self._accept_thread = threading.Thread(target=start_accepting, args=(self,))
self._accept_thread.daemon = True
self._accept_thread.start()
[docs] def emit(self, record: logging.LogRecord):
closed_clients = set()
for client in self.clients:
try:
timestamp = "%s.%03d" % (record.__getattribute__('asctime'),record.__getattribute__('msecs'))
except AttributeError:
timestamp = ''
try:
try:
# Python3
message = bytes("Cref={}.{}:{} Date={} LogType={} Logger={} Message={}\r\n".format(record.module, record.funcName, record.lineno, timestamp, record.levelname, record.name, record.getMessage()), 'UTF-8')
except TypeError:
# Python2
message = bytes("Cref={}.{}:{} Date={} LogType={} Logger={} Message={}\r\n".format(record.module, record.funcName, record.lineno, timestamp, record.levelname, record.name, record.getMessage())).encode('UTF-8')
client.sendall(message)
except socket.error:
closed_clients.add(client)
for client in closed_clients:
client.close() # just to be sure
self.clients.remove(client)
[docs] def getsockname(self):
return self.socket.getsockname()