build.rsudp.add_debug_handler() - Code Metrics - raspishake/rsudp - Measure and Improve Code Quality continuously with Scrutinizer

build.rsudp.add_debug_handler() A
last analyzed 2023-09-18 10:54 UTC

↳ Parent: build.rsudp

Complexity

Conditions

Size

Total Lines	29
Code Lines	17

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	12
CRAP Score	2

Importance

Changes

Metric	Value
eloc	17
dl	0
loc	29
ccs	12
cts	12
cp	1
rs	9.55
c	0
b	0
f	0
cc	2
nop	1
crap	2

import os, sys
import logging
from time import gmtime
from . import _version

# saved in case needed in the future
# warnings.filterwarnings('ignore', category=UserWarning, module='rsudp')
# warnings.filterwarnings('ignore', category=FutureWarning, module='obspy')


'''
Contains logging and formatting resources for command line and logfile output of rsudp.
'''

name = 'rsudp'
__version__ = _version.version

default_loc = '%s/.config/rsudp' % os.path.expanduser('~').replace('\\', '/')
settings_loc = os.path.join(default_loc, 'rsudp_settings.json').replace('\\', '/')
os.makedirs(default_loc, exist_ok=True)
log_dir = os.path.abspath('/tmp/rsudp')
log_name = 'rsudp.log'
log_loc = os.path.join(log_dir, log_name)
os.makedirs(log_dir, exist_ok=True)

# formatter settings
logging.Formatter.converter = gmtime
LOG = logging.getLogger('main')
LOGFORMAT = '%(asctime)-15s %(msg)s'
TIME_FORMAT = '%Y-%m-%d %H:%M:%S'

output_dir = False
data_dir = False
scap_dir = False
ms_path = False

COLOR = {
	'purple': '\033[95m',
	'blue': '\033[94m',
	'green': '\033[92m',
	'yellow': '\033[93m',
	'red': '\033[91m',
	'white': '\033[0m',
	'bold': "\033[1m"
}


def make_colors_friendly():
	'''
	Makes colors Windows-friendly if necessary.
	'''
	global COLOR
	if os.name == 'posix':
		pass	# terminal colors will work in this case
	else:
		for color in COLOR:
			COLOR[color] = ''

make_colors_friendly()


class LevelFormatter(logging.Formatter):
	'''
	.. |so_lf| raw:: html

		<a href="https://stackoverflow.com/a/28636024" target="_blank">this stackoverflow answer</a>

	A class that formats messages differently depending on their level.
	Adapted from |so_lf|.

	:param fmt: Format of message strings (see :py:mod:`logging`; example: ``'%(asctime)-15s %(msg)s'``)
	:type fmt: str or None
	:param datefmt: Date strings in strftime format (see :py:mod:`logging` example: ``'%Y-%m-%d %H:%M:%S'``)
	:type datefmt: str or None
	:param level_fmts: Dictionary of log levels and associated formats ``{logging.INFO: 'infoformat', logging.WARNING: 'warnformat', logging.ERROR: 'errformat'}``
	:type level_fmts: dict

	'''

	def __init__(self, fmt=None, datefmt=None, level_fmts={}):
		self._level_formatters = {}
		for level, format in level_fmts.items():
			# Could optionally support level names too
			self._level_formatters[level] = logging.Formatter(fmt=format, datefmt=datefmt)
		# self._fmt will be the default format
		super(LevelFormatter, self).__init__(fmt=fmt, datefmt=datefmt)
	# format records
	def format(self, record):
		if record.levelno in self._level_formatters:
			return self._level_formatters[record.levelno].format(record)
		return super(LevelFormatter, self).format(record)


def init_dirs(odir):
	'''
	Initialize the write directories if they do not already exist.

	:param str odir: output directory
	:return: ``True``
	:rtype: bool
	'''
	global output_dir, data_dir, scap_dir
	output_dir = odir
	data_dir = os.path.join(odir, 'data')
	scap_dir = os.path.join(odir, 'screenshots')
	try:
		os.makedirs(odir, exist_ok=True)
		os.makedirs(data_dir, exist_ok=True)
		os.makedirs(scap_dir, exist_ok=True)
	except OSError as e:
		print(COLOR['red'] + 'Error creating output directory structure. Are you sure you have permission to write to the output folder?' + COLOR['white'])
		print(COLOR['red'] + 'More info: %s' + COLOR['white'] % e)
		exit(2)

	return True


def start_logging(log_name=log_name, testing=False):
	'''
	Creates a handler for logging info and warnings to file.

	:param bool testing: whether or not testing is active (adds a "TESTING" label to messages)
	:return: ``True``
	:rtype: bool
	'''

	global LOG, LOGFORMAT
	LOG.setLevel('INFO')
	# logging formatters
	
	if testing:
		LOGFORMAT = '%(asctime)-15s TESTING %(msg)s'

	formatter = logging.Formatter(fmt=LOGFORMAT, datefmt=TIME_FORMAT)

	# this initializes logging to file
	f = logging.FileHandler(os.path.join(log_dir, log_name))
	f.setLevel('INFO')
	f.setFormatter(formatter)
	# warnings also go to file
	# initialize logging
	LOG.addHandler(f)
	printM('Logging initialized successfully.', sender='Init')
	return True


def add_debug_handler(testing=False):
	'''
	Creates an additional handler for logging info and warnings to the command line.

	:param bool testing: whether or not testing is active (adds a "TESTING" label to messages)
	:return: ``True``
	:rtype: bool
	'''
	global LOGFORMAT

	if testing:
		LOGFORMAT = '%(asctime)-15s TESTING %(msg)s'

	# terminal formats
	termformat = '\x1b[2K\r' + LOGFORMAT		# note '\x1b[2K' erases current line and \r returns to home
	# warning format
	warnformat = '\x1b[2K\r' + COLOR['yellow'] + LOGFORMAT + COLOR['white']
	# error format
	failformat = '\x1b[2K\r' + COLOR['red'] + LOGFORMAT + COLOR['white']
	termformatter = LevelFormatter(fmt=LOGFORMAT,
								   datefmt=TIME_FORMAT,
								   level_fmts={logging.INFO: termformat,
											   logging.WARNING: warnformat,
											   logging.ERROR: failformat},)
	s = logging.StreamHandler(sys.stdout)
	s.setLevel('INFO')
	s.setFormatter(termformatter)
	logging.getLogger('main').addHandler(s)
	return True


def get_scap_dir():
	'''
	This function returns the screen capture directory from the init function.
	This allows the variable to be more threadsafe.

	.. code-block:: python

		>>> get_scap_dir()
		'/home/pi/rsudp/screenshots/'

	:return: the path of the screenshot directory
	'''
	return scap_dir


def printM(msg, sender='', announce=False):
	'''
	Prints messages with datetime stamp and sends their output to the logging handlers.

	:param str msg: message to log
	:param str sender: the name of the class or function sending the message
	'''
	msg = u'[%s] %s' % (sender, msg) if sender != '' else msg
	# strip emoji from unicode by converting to ascii
	msg = msg.encode('ascii', 'ignore').decode('ascii')
	LOG.info(msg)


def printW(msg, sender='', announce=True, spaces=False):
	'''
	Prints warnings with datetime stamp and sends their output to the logging handlers.

	:param str msg: message to log
	:param str sender: the name of the class or function sending the message
	:param bool announce: whether or not to display "WARNING" before the message
	:param bool spaces: whether or not to display formatting spaces before the message
	'''
	if spaces:
		announce = False

	if announce:
		msg = u'[%s] WARNING: %s' % (sender, msg) if sender != '' else msg
	else:
		if spaces:
			msg = u'[%s]          %s' % (sender, msg) if sender != '' else msg
		else:
			msg = u'[%s] %s' % (sender, msg) if sender != '' else msg
	# strip emoji from unicode by converting to ascii
	msg = msg.encode('ascii', 'ignore').decode('ascii')
	LOG.warning(msg)


def printE(msg, sender='', announce=True, spaces=False):
	'''
	Prints errors with datetime stamp and sends their output to the logging handlers.

	:param str msg: message to log
	:param str sender: the name of the class or function sending the message
	:param bool announce: whether or not to display "WARNING" before the message
	:param bool spaces: whether or not to display formatting spaces before the message
	'''
	if spaces:
		announce = False

	if announce:
		msg = u'[%s] ERROR: %s' % (sender, msg) if sender != '' else msg
	else:
		if spaces:
			msg = u'[%s]        %s' % (sender, msg) if sender != '' else msg
		else:
			msg = u'[%s] %s' % (sender, msg) if sender != '' else msg
	# strip emoji from unicode by converting to ascii
	msg = msg.encode('ascii', 'ignore').decode('ascii')
	LOG.error(msg)


1	1	import os, sys
2	1	import logging
3	1	from time import gmtime
4	1	from . import _version
5
6		# saved in case needed in the future
7		# warnings.filterwarnings('ignore', category=UserWarning, module='rsudp')
8		# warnings.filterwarnings('ignore', category=FutureWarning, module='obspy')
9
10
11		'''
12		Contains logging and formatting resources for command line and logfile output of rsudp.
13		'''
14
15	1	name = 'rsudp'
16	1	__version__ = _version.version
17
18	1	default_loc = '%s/.config/rsudp' % os.path.expanduser('~').replace('\\', '/')
19	1	settings_loc = os.path.join(default_loc, 'rsudp_settings.json').replace('\\', '/')
20	1	os.makedirs(default_loc, exist_ok=True)
21	1	log_dir = os.path.abspath('/tmp/rsudp')
22	1	log_name = 'rsudp.log'
23	1	log_loc = os.path.join(log_dir, log_name)
24	1	os.makedirs(log_dir, exist_ok=True)
25
26		# formatter settings
27	1	logging.Formatter.converter = gmtime
28	1	LOG = logging.getLogger('main')
29	1	LOGFORMAT = '%(asctime)-15s %(msg)s'
30	1	TIME_FORMAT = '%Y-%m-%d %H:%M:%S'
31
32	1	output_dir = False
33	1	data_dir = False
34	1	scap_dir = False
35	1	ms_path = False
36
37	1	COLOR = {
38		'purple': '\033[95m',
39		'blue': '\033[94m',
40		'green': '\033[92m',
41		'yellow': '\033[93m',
42		'red': '\033[91m',
43		'white': '\033[0m',
44		'bold': "\033[1m"
45		}
46
47
48	1	def make_colors_friendly():
49		'''
50		Makes colors Windows-friendly if necessary.
51		'''
52		global COLOR
53	1	if os.name == 'posix':
54	1	pass # terminal colors will work in this case
55		else:
56		for color in COLOR:
57		COLOR[color] = ''
58
59	1	make_colors_friendly()
60
61
62	1	class LevelFormatter(logging.Formatter):
63		'''
64		.. \|so_lf\| raw:: html
65
66		<a href="https://stackoverflow.com/a/28636024" target="_blank">this stackoverflow answer</a>
67
68		A class that formats messages differently depending on their level.
69		Adapted from \|so_lf\|.
70
71		:param fmt: Format of message strings (see :py:mod:`logging`; example: ``'%(asctime)-15s %(msg)s'``)
72		:type fmt: str or None
73		:param datefmt: Date strings in strftime format (see :py:mod:`logging` example: ``'%Y-%m-%d %H:%M:%S'``)
74		:type datefmt: str or None
75		:param level_fmts: Dictionary of log levels and associated formats ``{logging.INFO: 'infoformat', logging.WARNING: 'warnformat', logging.ERROR: 'errformat'}``
76		:type level_fmts: dict
77
78		'''
79
80	1	def __init__(self, fmt=None, datefmt=None, level_fmts={}):
81	1	self._level_formatters = {}
82	1	for level, format in level_fmts.items():
83		# Could optionally support level names too
84	1	self._level_formatters[level] = logging.Formatter(fmt=format, datefmt=datefmt)
85		# self._fmt will be the default format
86	1	super(LevelFormatter, self).__init__(fmt=fmt, datefmt=datefmt)
87		# format records
88	1	def format(self, record):
89	1	if record.levelno in self._level_formatters:
90	1	return self._level_formatters[record.levelno].format(record)
91		return super(LevelFormatter, self).format(record)
92
93
94	1	def init_dirs(odir):
95		'''
96		Initialize the write directories if they do not already exist.
97
98		:param str odir: output directory
99		:return: ``True``
100		:rtype: bool
101		'''
102		global output_dir, data_dir, scap_dir
103	1	output_dir = odir
104	1	data_dir = os.path.join(odir, 'data')
105	1	scap_dir = os.path.join(odir, 'screenshots')
106	1	try:
107	1	os.makedirs(odir, exist_ok=True)
108	1	os.makedirs(data_dir, exist_ok=True)
109	1	os.makedirs(scap_dir, exist_ok=True)
110		except OSError as e:
111		print(COLOR['red'] + 'Error creating output directory structure. Are you sure you have permission to write to the output folder?' + COLOR['white'])
112		print(COLOR['red'] + 'More info: %s' + COLOR['white'] % e)
113		exit(2)
114
115	1	return True
116
117
118	1	def start_logging(log_name=log_name, testing=False):
119		'''
120		Creates a handler for logging info and warnings to file.
121
122		:param bool testing: whether or not testing is active (adds a "TESTING" label to messages)
123		:return: ``True``
124		:rtype: bool
125		'''
126
127		global LOG, LOGFORMAT
128	1	LOG.setLevel('INFO')
129		# logging formatters
130
131	1	if testing:
132	1	LOGFORMAT = '%(asctime)-15s TESTING %(msg)s'
133
134	1	formatter = logging.Formatter(fmt=LOGFORMAT, datefmt=TIME_FORMAT)
135
136		# this initializes logging to file
137	1	f = logging.FileHandler(os.path.join(log_dir, log_name))
138	1	f.setLevel('INFO')
139	1	f.setFormatter(formatter)
140		# warnings also go to file
141		# initialize logging
142	1	LOG.addHandler(f)
143	1	printM('Logging initialized successfully.', sender='Init')
144	1	return True
145
146
147	1	def add_debug_handler(testing=False):
148		'''
149		Creates an additional handler for logging info and warnings to the command line.
150
151		:param bool testing: whether or not testing is active (adds a "TESTING" label to messages)
152		:return: ``True``
153		:rtype: bool
154		'''
155		global LOGFORMAT
156
157	1	if testing:
158	1	LOGFORMAT = '%(asctime)-15s TESTING %(msg)s'
159
160		# terminal formats
161	1	termformat = '\x1b[2K\r' + LOGFORMAT # note '\x1b[2K' erases current line and \r returns to home
162		# warning format
163	1	warnformat = '\x1b[2K\r' + COLOR['yellow'] + LOGFORMAT + COLOR['white']
164		# error format
165	1	failformat = '\x1b[2K\r' + COLOR['red'] + LOGFORMAT + COLOR['white']
166	1	termformatter = LevelFormatter(fmt=LOGFORMAT,
167		datefmt=TIME_FORMAT,
168		level_fmts={logging.INFO: termformat,
169		logging.WARNING: warnformat,
170		logging.ERROR: failformat},)
171	1	s = logging.StreamHandler(sys.stdout)
172	1	s.setLevel('INFO')
173	1	s.setFormatter(termformatter)
174	1	logging.getLogger('main').addHandler(s)
175	1	return True
176
177
178	1	def get_scap_dir():
179		'''
180		This function returns the screen capture directory from the init function.
181		This allows the variable to be more threadsafe.
182
183		.. code-block:: python
184
185		>>> get_scap_dir()
186		'/home/pi/rsudp/screenshots/'
187
188		:return: the path of the screenshot directory
189		'''
190	1	return scap_dir
191
192
193	1	def printM(msg, sender='', announce=False):
194		'''
195		Prints messages with datetime stamp and sends their output to the logging handlers.
196
197		:param str msg: message to log
198		:param str sender: the name of the class or function sending the message
199		'''
200	1	msg = u'[%s] %s' % (sender, msg) if sender != '' else msg
201		# strip emoji from unicode by converting to ascii
202	1	msg = msg.encode('ascii', 'ignore').decode('ascii')
203	1	LOG.info(msg)
204
205
206	1	def printW(msg, sender='', announce=True, spaces=False):
207		'''
208		Prints warnings with datetime stamp and sends their output to the logging handlers.
209
210		:param str msg: message to log
211		:param str sender: the name of the class or function sending the message
212		:param bool announce: whether or not to display "WARNING" before the message
213		:param bool spaces: whether or not to display formatting spaces before the message
214		'''
215	1	if spaces:
216	1	announce = False
217
218	1	if announce:
219	1	msg = u'[%s] WARNING: %s' % (sender, msg) if sender != '' else msg
220		else:
221	1	if spaces:
222	1	msg = u'[%s] %s' % (sender, msg) if sender != '' else msg
223		else:
224	1	msg = u'[%s] %s' % (sender, msg) if sender != '' else msg
225		# strip emoji from unicode by converting to ascii
226	1	msg = msg.encode('ascii', 'ignore').decode('ascii')
227	1	LOG.warning(msg)
228
229
230	1	def printE(msg, sender='', announce=True, spaces=False):
231		'''
232		Prints errors with datetime stamp and sends their output to the logging handlers.
233
234		:param str msg: message to log
235		:param str sender: the name of the class or function sending the message
236		:param bool announce: whether or not to display "WARNING" before the message
237		:param bool spaces: whether or not to display formatting spaces before the message
238		'''
239		if spaces:
240		announce = False
241
242		if announce:
243		msg = u'[%s] ERROR: %s' % (sender, msg) if sender != '' else msg
244		else:
245		if spaces:
246		msg = u'[%s] %s' % (sender, msg) if sender != '' else msg
247		else:
248		msg = u'[%s] %s' % (sender, msg) if sender != '' else msg
249		# strip emoji from unicode by converting to ascii
250		msg = msg.encode('ascii', 'ignore').decode('ascii')
251		LOG.error(msg)
252

raspishake / rsudp

build.rsudp.add_debug_handler() A last analyzed 2023-09-18 10:54 UTC

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like

build.rsudp.add_debug_handler() A
last analyzed 2023-09-18 10:54 UTC