Filter.process() - Code Metrics - Inspection of "misc improvements for linux proc readers, fixed py..." - s4z/plumd - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 8451c1...a45803 )

by Kenny

created 2016-08-15 13:56 UTC

Filter.process() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	1
c	1
b	0
f	0
dl	0
loc	6
rs	9.4285

# -*- coding: utf-8 -*-
"""Various helper classes.

.. moduleauthor:: Kenny Freeman <[email protected]>

"""

__author__ = 'Kenny Freeman'
__email__ = '[email protected]'
__license__     = "ISCL"
__docformat__   = 'reStructuredText'

import os

import sys
PY3 = sys.version_info > (3,)

import os.path
import time
import fcntl
import socket
import signal
import string
from collections import deque


def get_hostname():
    """
    Returns the hostname of the system the code is running on. Eventually this
    may include more checks.

    :rtype: str
    """
    return socket.gethostname().split(".")[0]


def get_file_map(fname, col, skip):
    """Return a dict representation of a file - use on small files only.

    eg. file contents of:

    8       1 sda1 284 166 2673 1724 58 18 9056 369 0 1122 2091

    with a call of col=2, skip=0 returns:

    {"sda1": [8, 1, 284, 166, 2673, 1724, 58, 18, 9056, 369, 0, 1122, 2091]}

    :param fname: Full pathname to a file.
    :type fname: str
    :param col: Column number to use as the map
    :type col: int
    :param skip: Skip this many lines of the file.
    :type skip: int
    :rtype: dict
    """
    fdat = {}
    if os.access(fname, os.R_OK):
        with open(fname, 'r') as fd:
            dat = fd.read().split("\n")
            for line in dat[skip:]:
                vals = line.split()
                if len(vals) > col:
                    map_val = vals.pop(col)
                    fdat[map_val] = deque(vals)
    return fdat


def get_file_map_list(fname, col, skip):
    """Return a dict representation of a file - use on small files only.

    eg. file contents of:

    8       1 sda1 284 166 2673 1724 58 18 9056 369 0 1122 2091

    with a call of col=2, skip=0 returns:

    {"sda1": [8, 1, 284, 166, 2673, 1724, 58, 18, 9056, 369, 0, 1122, 2091]}

    :param fname: Full pathname to a file.
    :type fname: str
    :param col: Column number to use as the map

    :type col: int
    :param skip: Skip this many lines of the file.
    :type skip: int
    :rtype: dict
    """
    fdat = {}
    if os.access(fname, os.R_OK):
        with open(fname, 'r') as fd:
            dat = fd.read().split("\n")
            for line in dat[skip:]:
                vals = line.split()
                if len(vals) > col:
                    map_val = vals.pop(col)
                    fdat[map_val] = vals
    return fdat


def get_file_list(fname):
    """Return the file as a string - use on small files only.

    :param fname: Full pathname to a file.
    :type fname: str
    :rtype: dict
    """
    if os.access(fname, os.R_OK):
        with open(fname, 'r') as fd:
            return deque(fd.read().strip().split("\n"))
    return []


def get_file(fname):

    """Return the file as a string - use on small files only.

    :param fname: Full pathname to a file.
    :type fname: str
    :rtype: dict
    """
    if os.access(fname, os.R_OK):
        with open(fname, 'r') as fd:
            return fd.read()
    return []


class Interval(object):
    """A simple helper class to ensure a loop occours every ltime seconds
    while optionally providing a means to interupt the loop with a
    :class:`threading.Event`.


    :param ltime: The target duration of a loop
    :type ltime: int or float
    :param evt: An event object that will cause the loop to complete early if
        the event is triggered while waiting.
    :type evt: threading.Event
    :raises: ValueError if the passed loop time is not an int or float or is < 0


    As an example, use it like this to loop on an interval:

    :Example:

        >>> import threading
        >>> evt = threading.Event()
        >>> while loop:
        >>>     with Interval(111.1, evt) as timer:
        >>>         do_things_here()

    .. note:: if the do_things_here() call returns before the loop timer
        completes the timer object will either evt.wait() if evt was defined or
        time.sleep() for the time remaining in the interval.
    """

    def __init__(self, ltime, evt=None):
        if ltime is None or ltime < 0:
            raise ValueError("an interval must be defined")
        self.ltime = ltime
        self.evt = evt
        self.lstart = None


    @property
    def remaining(self):
        """Return the remaining time in the loop.

        :rtype: float
        """
        if self.lstart is not None:
            return self.ltime - (time.time() - self.lstart)
        return 0.0


    @property
    def elapsed(self):
        """Return the elapsed time of the loop.

        :rtype: float
        """
        if self.lstart is not None:
            return time.time() - self.lstart
        return 0.0


    def __str__(self):
        """Return a nicely formatted string.

        :rtype: str
        """
        elapsed = None
        if self.lstart is not None:
            elapsed = time.time() - self.lstart
        s = "Interval {0} seconds, {1} elapsed, {2} started"
        return s.format(self.ltime, elapsed, self.lstart)


    def __repr__(self):
        """Return a nicely formatted string.

        :rtype: str
        """
        elapsed = None
        if self.lstart is not None:
            elapsed = time.time() - self.lstart
        s = "Interval {0} seconds, {1} elapsed, {2} started"
        return s.format(self.ltime, elapsed, self.lstart)


    def __enter__(self):
        """Called on entrance to with() statement - simply records loop start
        time.

        :rtype: Interval"""
        self.lstart = time.time()
        return self


    def __exit__(self, *args):
        """Called on exit from a with() statement and either waits on the
        self.evt :class:`threading.Event` event if defined or time.sleep()'s
        the remaining interval time."""
        if self.lstart is None:
            self.ltstart = time.time()
        # block on the event for the remaining time, if any
        wtime = self.ltime - (time.time() - self.lstart)
        if wtime > 0:
            if self.evt is not None:
                self.evt.wait(wtime)
            else:
                time.sleep(wtime)


class SignalWaiter(object):
    """A helper class that makes waiting on signals very easy. Define the
    list of signals to wait for and to ignore and then call wait(). Instead of
    using signal handler functions it uses the signal.set_wakeup_fd() function
    and blocks on read().

    See http://www.pocketnix.org/doc/Fighting_set__wakeup__fd/ and others for
    more details.

    :param sig_wait: :class:`list` signals to wait for
    :type sig_wait: list
    :param sig_ignore: :class:`list` signals to ignore
    :type sig_ignore: list
    """

    def __init__(self, sig_wait, sig_ignore):
        """Create a :class:`SignalWaiter`"""
        self.sig_wait = sig_wait
        self.sig_ignore = sig_ignore

        # use a pipe to listen for signals
        self.rpipe, self.wpipe = os.pipe()
        # set non blocking mode on the write end of the pipe
        flags = fcntl.fcntl(self.wpipe, fcntl.F_GETFL, 0)
        flags = flags | os.O_NONBLOCK
        flags = fcntl.fcntl(self.wpipe, fcntl.F_SETFL, flags)

        # set the write end as a signal wakeup fd - we read signals from it
        signal.set_wakeup_fd(self.wpipe)

        # install dummy signal handlers that do nothing
        for signum in self.sig_wait:
            signal.signal(signum, lambda x,y: None)

        # install the ignore signal handler for sig_ignore signals
        for signum in self.sig_ignore:
            signal.signal(signum, signal.SIG_IGN)

        # now a call to wait() blocks the caller until a signal is received
        # only the signals we've registered handlers for will trigger


    def wait(self):
        """Blocks the calling thread until signal(s) are received.

        :rtype: :class:`int` or :class:`Exception`
        """
        try:
            return os.read(self.rpipe, 1)
        except Exception as e:
            return e


class Filter(object):
    """A helper class that efficiently filters characters from strings.

    :param keep_chars: A string containing all characters to keep in strings
    :type keep_chars: str
    """

    def __init__(self, keep_chars=None):
        """Create a :class:`Filter`"""
        if not keep_chars:
            # translate is different in python3, so is string
            if PY3:
                keep_chars = string.ascii_letters + string.digits + "_-"
            else:
                keep_chars = string.letters + string.digits + "_-"
        self.filter_chars = "".join(char for char in map(chr, range(256)) \
                                   if not char in keep_chars)
        self.table = dict.fromkeys(map(ord, self.filter_chars), None)
        if PY3:
            self.process = self.process_py3
        else:
            self.process = self.process_py2


    def process_py3(self, filter_str):
        """Returns a filtered string, removes filter_char's from it.

        :rtype: str
        """
        return filter_str.translate(self.table)


    def process_py2(self, filter_str):
        """Returns a filtered string, removes filter_char's from it.

        :rtype: str
        """
        return filter_str.translate(None, self.filter_chars)


1		# -- coding: utf-8 --
2		"""Various helper classes.
3
4		.. moduleauthor:: Kenny Freeman <[email protected]>
5
6		"""
7
8		__author__ = 'Kenny Freeman'
9		__email__ = '[email protected]'
10		__license__ = "ISCL"
11		__docformat__ = 'reStructuredText'
12
13		import os
14
15		import sys
16		PY3 = sys.version_info > (3,)
17
18		import os.path
19		import time
20		import fcntl
21		import socket
22		import signal
23		import string
24		from collections import deque
25
26
27		def get_hostname():
28		"""
29		Returns the hostname of the system the code is running on. Eventually this
30		may include more checks.
31
32		:rtype: str
33		"""
34		return socket.gethostname().split(".")[0]
35
36
37		def get_file_map(fname, col, skip):
38		"""Return a dict representation of a file - use on small files only.
39
40		eg. file contents of:
41
42		8 1 sda1 284 166 2673 1724 58 18 9056 369 0 1122 2091
43
44		with a call of col=2, skip=0 returns:
45
46		{"sda1": [8, 1, 284, 166, 2673, 1724, 58, 18, 9056, 369, 0, 1122, 2091]}
47
48		:param fname: Full pathname to a file.
49		:type fname: str
50		:param col: Column number to use as the map
51		:type col: int
52		:param skip: Skip this many lines of the file.
53		:type skip: int
54		:rtype: dict
55		"""
56		fdat = {}
57		if os.access(fname, os.R_OK):
58		with open(fname, 'r') as fd:
59		dat = fd.read().split("\n")
60		for line in dat[skip:]:
61		vals = line.split()
62		if len(vals) > col:
63		map_val = vals.pop(col)
64		fdat[map_val] = deque(vals)
65		return fdat
66
67
68		def get_file_map_list(fname, col, skip):
69		"""Return a dict representation of a file - use on small files only.
70
71		eg. file contents of:
72
73		8 1 sda1 284 166 2673 1724 58 18 9056 369 0 1122 2091
74
75		with a call of col=2, skip=0 returns:
76
77		{"sda1": [8, 1, 284, 166, 2673, 1724, 58, 18, 9056, 369, 0, 1122, 2091]}
78
79		:param fname: Full pathname to a file.
80		:type fname: str
81	View Code Duplication	:param col: Column number to use as the map
		1 ignored issue – show Duplication introduced 2016-08-02 16:06 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
82		:type col: int
83		:param skip: Skip this many lines of the file.
84		:type skip: int
85		:rtype: dict
86		"""
87		fdat = {}
88		if os.access(fname, os.R_OK):
89		with open(fname, 'r') as fd:
90		dat = fd.read().split("\n")
91		for line in dat[skip:]:
92		vals = line.split()
93		if len(vals) > col:
94		map_val = vals.pop(col)
95		fdat[map_val] = vals
96		return fdat
97
98
99		def get_file_list(fname):
100		"""Return the file as a string - use on small files only.
101
102		:param fname: Full pathname to a file.
103		:type fname: str
104		:rtype: dict
105		"""
106		if os.access(fname, os.R_OK):
107		with open(fname, 'r') as fd:
108		return deque(fd.read().strip().split("\n"))
109		return []
110
111
112	View Code Duplication	def get_file(fname):
		1 ignored issue – show Duplication introduced 2016-08-02 16:06 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
113		"""Return the file as a string - use on small files only.
114
115		:param fname: Full pathname to a file.
116		:type fname: str
117		:rtype: dict
118		"""
119		if os.access(fname, os.R_OK):
120		with open(fname, 'r') as fd:
121		return fd.read()
122		return []
123
124
125		class Interval(object):
126		"""A simple helper class to ensure a loop occours every ltime seconds
127		while optionally providing a means to interupt the loop with a
128		:class:`threading.Event`.
129
130
131		:param ltime: The target duration of a loop
132		:type ltime: int or float
133		:param evt: An event object that will cause the loop to complete early if
134		the event is triggered while waiting.
135		:type evt: threading.Event
136		:raises: ValueError if the passed loop time is not an int or float or is < 0
137
138
139		As an example, use it like this to loop on an interval:
140
141		:Example:
142
143		>>> import threading
144		>>> evt = threading.Event()
145		>>> while loop:
146		>>> with Interval(111.1, evt) as timer:
147		>>> do_things_here()
148
149		.. note:: if the do_things_here() call returns before the loop timer
150		completes the timer object will either evt.wait() if evt was defined or
151		time.sleep() for the time remaining in the interval.
152		"""
153
154		def __init__(self, ltime, evt=None):
155		if ltime is None or ltime < 0:
156		raise ValueError("an interval must be defined")
157		self.ltime = ltime
158		self.evt = evt
159		self.lstart = None
160
161
162		@property
163		def remaining(self):
164		"""Return the remaining time in the loop.
165
166		:rtype: float
167		"""
168		if self.lstart is not None:
169		return self.ltime - (time.time() - self.lstart)
170		return 0.0
171
172
173		@property
174		def elapsed(self):
175		"""Return the elapsed time of the loop.
176
177		:rtype: float
178		"""
179		if self.lstart is not None:
180		return time.time() - self.lstart
181		return 0.0
182
183
184		def __str__(self):
185		"""Return a nicely formatted string.
186
187		:rtype: str
188		"""
189		elapsed = None
190		if self.lstart is not None:
191		elapsed = time.time() - self.lstart
192		s = "Interval {0} seconds, {1} elapsed, {2} started"
193		return s.format(self.ltime, elapsed, self.lstart)
194
195
196		def __repr__(self):
197		"""Return a nicely formatted string.
198
199		:rtype: str
200		"""
201		elapsed = None
202		if self.lstart is not None:
203		elapsed = time.time() - self.lstart
204		s = "Interval {0} seconds, {1} elapsed, {2} started"
205		return s.format(self.ltime, elapsed, self.lstart)
206
207
208		def __enter__(self):
209		"""Called on entrance to with() statement - simply records loop start
210		time.
211
212		:rtype: Interval"""
213		self.lstart = time.time()
214		return self
215
216
217		def __exit__(self, *args):
218		"""Called on exit from a with() statement and either waits on the
219		self.evt :class:`threading.Event` event if defined or time.sleep()'s
220		the remaining interval time."""
221		if self.lstart is None:
222		self.ltstart = time.time()
223		# block on the event for the remaining time, if any
224		wtime = self.ltime - (time.time() - self.lstart)
225		if wtime > 0:
226		if self.evt is not None:
227		self.evt.wait(wtime)
228		else:
229		time.sleep(wtime)
230
231
232		class SignalWaiter(object):
233		"""A helper class that makes waiting on signals very easy. Define the
234		list of signals to wait for and to ignore and then call wait(). Instead of
235		using signal handler functions it uses the signal.set_wakeup_fd() function
236		and blocks on read().
237
238		See http://www.pocketnix.org/doc/Fighting_set__wakeup__fd/ and others for
239		more details.
240
241		:param sig_wait: :class:`list` signals to wait for
242		:type sig_wait: list
243		:param sig_ignore: :class:`list` signals to ignore
244		:type sig_ignore: list
245		"""
246
247		def __init__(self, sig_wait, sig_ignore):
248		"""Create a :class:`SignalWaiter`"""
249		self.sig_wait = sig_wait
250		self.sig_ignore = sig_ignore
251
252		# use a pipe to listen for signals
253		self.rpipe, self.wpipe = os.pipe()
254		# set non blocking mode on the write end of the pipe
255		flags = fcntl.fcntl(self.wpipe, fcntl.F_GETFL, 0)
256		flags = flags \| os.O_NONBLOCK
257		flags = fcntl.fcntl(self.wpipe, fcntl.F_SETFL, flags)
258
259		# set the write end as a signal wakeup fd - we read signals from it
260		signal.set_wakeup_fd(self.wpipe)
261
262		# install dummy signal handlers that do nothing
263		for signum in self.sig_wait:
264		signal.signal(signum, lambda x,y: None)
265
266		# install the ignore signal handler for sig_ignore signals
267		for signum in self.sig_ignore:
268		signal.signal(signum, signal.SIG_IGN)
269
270		# now a call to wait() blocks the caller until a signal is received
271		# only the signals we've registered handlers for will trigger
272
273
274		def wait(self):
275		"""Blocks the calling thread until signal(s) are received.
276
277		:rtype: :class:`int` or :class:`Exception`
278		"""
279		try:
280		return os.read(self.rpipe, 1)
281		except Exception as e:
282		return e
283
284
285		class Filter(object):
286		"""A helper class that efficiently filters characters from strings.
287
288		:param keep_chars: A string containing all characters to keep in strings
289		:type keep_chars: str
290		"""
291
292		def __init__(self, keep_chars=None):
293		"""Create a :class:`Filter`"""
294		if not keep_chars:
295		# translate is different in python3, so is string
296		if PY3:
297		keep_chars = string.ascii_letters + string.digits + "_-"
298		else:
299		keep_chars = string.letters + string.digits + "_-"
300		self.filter_chars = "".join(char for char in map(chr, range(256)) \
301		if not char in keep_chars)
302		self.table = dict.fromkeys(map(ord, self.filter_chars), None)
303		if PY3:
304		self.process = self.process_py3
305		else:
306		self.process = self.process_py2
307
308
309		def process_py3(self, filter_str):
310		"""Returns a filtered string, removes filter_char's from it.
311
312		:rtype: str
313		"""
314		return filter_str.translate(self.table)
315
316
317		def process_py2(self, filter_str):
318		"""Returns a filtered string, removes filter_char's from it.
319
320		:rtype: str
321		"""
322		return filter_str.translate(None, self.filter_chars)
323

s4z / plumd

Push — master ( 8451c1...a45803 )

Filter.process() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like