get_file_map() - Code Metrics - Inspection of "- Added /proc readers for: cpu, diskspace, diskusa..." - s4z/plumd - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( ee7b96...42a2db )

by Kenny

created 2016-08-02 16:04 UTC

get_file_map() B

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	29
Ratio	100 %

Importance

Changes

Metric	Value
cc	5
c	0
b	0
f	0
dl	29
loc	29
rs	8.0894

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

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

"""

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

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 escape_tag(tag):
    """Escapes a string for use in eg. an influxdb measurement tag.
    todo: regexp replacements

    :param tag: a string
    :type tag: str
    :rtype: str
    """
    rpls = [(" ", "\ "), (",", "\,"), ("\n", "\\n")]
    for rpl in rpls:
        tag = tag.replace(*rpl)
    return tag


def escape_field(field):
    """Escapes a string for use in eg. an influxdb measurement field.
    todo: regexp replacements

    :param field: a string
    :type field: str
    :rtype: str
    """
    rpls = [(" ", "\ "), (",", "\,"), ("\n", "\\n"), ('"', '\"')]
    for rpl in rpls:
        field = field.replace(*rpl)
    return '""{0}""'.format(field)


def replace_chars(tag, rpls=None, rpl_with=None):
    """Rewrites a string to make it suitable for use in eg. a graphite metric.
    todo: regexp replacements

    :param tag: the tag to process
    :type tag: str
    :param repl_chars: a list of strings to replace
    :type repl_chars: list
    :param repl_char: a string to replace each repl_chars with
    :type repl_char: str
    :rtype: str
    """
    if not rpls:
        rpls = [" ", ".", "\n", "="]
    if not rpl_with:
        rpl_with = "_"
    for rpl in rpls:
        tag = tag.replace(rpl, rpl_with)
    return tag


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

s4z / plumd

Push — master ( ee7b96...42a2db )

get_file_map() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like