escape_field() - Code Metrics - Inspection of "documentation updates, misc cleanup" - s4z/plumd - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( faefc5...f161ca )

by Kenny

created 2016-08-13 07:53 UTC

escape_field() A

↳ Parent: get_file_map()

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	2
Bugs	1	Features	0

Metric	Value
cc	2
c	2
b	1
f	0
dl	0
loc	12
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 re
import os
import os.path
import time
import fcntl
import socket
import signal
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().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


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

s4z / plumd

Push — master ( faefc5...f161ca )

escape_field() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like