plumd.replace_chars() - Code Metrics - Inspection of "cleanup graphite writer plugin" - s4z/plumd - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 822bc5...a4197e )

by Kenny

created 2016-02-21 06:47 UTC

plumd.replace_chars() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
cc	4
dl	0
loc	18
rs	9.2

# -*- 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


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.

    :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.

    :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.

    :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


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
20
21			def get_hostname():
22			"""
23			Returns the hostname of the system the code is running on. Eventually this
24			may include more checks.
25
26			:rtype: str
27			"""
28			return socket.gethostname().split(".")[0]
29
30
31			def escape_tag(tag):
32			"""Escapes a string for use in eg. an influxdb measurement tag.
33
34			:param tag: a string
35			:type tag: str
36			:rtype: str
37			"""
38			rpls = [(" ", "\ "), (",", "\,"), ("\n", "\\n")]
39			for rpl in rpls:
40			tag = tag.replace(*rpl)
41			return tag
42
43
44			def escape_field(field):
45			"""Escapes a string for use in eg. an influxdb measurement field.
46
47			:param field: a string
48			:type field: str
49			:rtype: str
50			"""
51			rpls = [(" ", "\ "), (",", "\,"), ("\n", "\\n")]
52			for rpl in rpls:
53			field = field.replace(*rpl)
54			return '""{0}""'.format(field)
55
56
57			def replace_chars(tag, rpls=None, rpl_with=None):
58			"""Rewrites a string to make it suitable for use in eg. a graphite metric.
59
60			:param tag: the tag to process
61			:type tag: str
62			:param repl_chars: a list of strings to replace
63			:type repl_chars: list
64			:param repl_char: a string to replace each repl_chars with
65			:type repl_char: str
66			:rtype: str
67			"""
68			if not rpls:
69			rpls = [" ", ".", "\n", "="]
70			if not rpl_with:
71			rpl_with = "_"
72			for rpl in rpls:
73			tag = tag.replace(rpl, rpl_with)
74			return tag
75
76
77			class Interval(object):
78			"""A simple helper class to ensure a loop occours every ltime seconds
79			while optionally providing a means to interupt the loop with a
80			:class:`threading.Event`.
81
82
83			:param ltime: The target duration of a loop
84			:type ltime: int or float
85			:param evt: An event object that will cause the loop to complete early if
86			the event is triggered while waiting.
87			:type evt: threading.Event
88			:raises: ValueError if the passed loop time is not an int or float or is < 0
89
90
91			As an example, use it like this to loop on an interval:
92
93			:Example:
94
95			>>> import threading
96			>>> evt = threading.Event()
97			>>> while loop:
98			>>> with Interval(111.1, evt) as timer:
99			>>> do_things_here()
100
101			.. note:: if the do_things_here() call returns before the loop timer
102			completes the timer object will either evt.wait() if evt was defined or
103			time.sleep() for the time remaining in the interval.
104			"""
105
106			def __init__(self, ltime, evt=None):
107			if ltime is None or ltime < 0:
108			raise ValueError("an interval must be defined")
109			self.ltime = ltime
110			self.evt = evt
111			self.lstart = None
112
113
114			@property
115			def remaining(self):
116			"""Return the remaining time in the loop.
117
118			:rtype: float
119			"""
120			if self.lstart is not None:
121			return self.ltime - (time.time() - self.lstart)
122			return 0.0
123
124
125			@property
126			def elapsed(self):
127			"""Return the elapsed time of the loop.
128
129			:rtype: float
130			"""
131			if self.lstart is not None:
132			return time.time() - self.lstart
133			return 0.0
134
135
136			def __str__(self):
137			"""Return a nicely formatted string.
138
139			:rtype: str
140			"""
141			elapsed = None
142			if self.lstart is not None:
143			elapsed = time.time() - self.lstart
144			s = "Interval {0} seconds, {1} elapsed, {2} started"
145			return s.format(self.ltime, elapsed, self.lstart)
146
147
148			def __repr__(self):
149			"""Return a nicely formatted string.
150
151			:rtype: str
152			"""
153			elapsed = None
154			if self.lstart is not None:
155			elapsed = time.time() - self.lstart
156			s = "Interval {0} seconds, {1} elapsed, {2} started"
157			return s.format(self.ltime, elapsed, self.lstart)
158
159
160			def __enter__(self):
161			"""Called on entrance to with() statement - simply records loop start
162			time.
163
164			:rtype: Interval"""
165			self.lstart = time.time()
166			return self
167
168
169			def __exit__(self, *args):
170			"""Called on exit from a with() statement and either waits on the
171			self.evt :class:`threading.Event` event if defined or time.sleep()'s
172			the remaining interval time."""
173			if self.lstart is None:
174			self.ltstart = time.time()
175			# block on the event for the remaining time, if any
176			wtime = self.ltime - (time.time() - self.lstart)
177			if wtime > 0:
178			if self.evt is not None:
179			self.evt.wait(wtime)
180			else:
181			time.sleep(wtime)
182
183
184			class SignalWaiter(object):
185			"""A helper class that makes waiting on signals very easy. Define the
186			list of signals to wait for and to ignore and then call wait(). Instead of
187			using signal handler functions it uses the signal.set_wakeup_fd() function
188			and blocks on read().
189
190			See http://www.pocketnix.org/doc/Fighting_set__wakeup__fd/ and others for
191			more details.
192
193			:param sig_wait: :class:`list` signals to wait for
194			:type sig_wait: list
195			:param sig_ignore: :class:`list` signals to ignore
196			:type sig_ignore: list
197			"""
198
199			def __init__(self, sig_wait, sig_ignore):
200			"""Create a :class:`SignalWaiter`"""
201			self.sig_wait = sig_wait
202			self.sig_ignore = sig_ignore
203
204			# use a pipe to listen for signals
205			self.rpipe, self.wpipe = os.pipe()
206			# set non blocking mode on the write end of the pipe
207			flags = fcntl.fcntl(self.wpipe, fcntl.F_GETFL, 0)
208			flags = flags \| os.O_NONBLOCK
209			flags = fcntl.fcntl(self.wpipe, fcntl.F_SETFL, flags)
210
211			# set the write end as a signal wakeup fd - we read signals from it
212			signal.set_wakeup_fd(self.wpipe)
213
214			# install dummy signal handlers that do nothing
215			for signum in self.sig_wait:
216			signal.signal(signum, lambda x,y: None)
217
218			# install the ignore signal handler for sig_ignore signals
219			for signum in self.sig_ignore:
220			signal.signal(signum, signal.SIG_IGN)
221
222			# now a call to wait() blocks the caller until a signal is received
223			# only the signals we've registered handlers for will trigger
224
225
226			def wait(self):
227			"""Blocks the calling thread until signal(s) are received.
228
229			:rtype: :class:`int` or :class:`Exception`
230			"""
231			try:
232			return os.read(self.rpipe, 1)
233			except Exception as e:
234			return e
235

s4z / plumd

Push — master ( 822bc5...a4197e )

plumd.replace_chars() A

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like