build.rsudp.c_rsam.RSAM._set_deconv() - Code Metrics - Inspection of "Consumer for RSAM amplitude analysis" - raspishake/rsudp - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#6)

unknown

created 2020-04-18 23:49 UTC

build.rsudp.c_rsam.RSAM._set_deconv() B

↳ Parent: build.rsudp.c_rsam

Complexity

Conditions

Size

Total Lines	23
Code Lines	9

Duplication

Lines	23
Ratio	100 %

Importance

Changes

Metric	Value
eloc	9
dl	23
loc	23
rs	8.6666
c	0
b	0
f	0
cc	6
nop	2

import sys, os, time
import socket as s
from datetime import datetime, timedelta
import statistics
from rsudp import printM, printW, printE
from rsudp import helpers
import rsudp.raspberryshake as rs
COLOR = {}
from rsudp import COLOR

# set the terminal text color to green
COLOR['current'] = COLOR['green']

class RSAM(rs.ConsumerThread):
	"""
	.. versionadded:: 1.0.0

	A consumer class that runs an Real-time Seismic Amplitude Measurement (RSAM).

	:param queue.Queue q: queue of data and messages sent by :class:`rsudp.c_consumer.Consumer`.
	:param bool debug: whether or not to display RSAM analysis live to the console.
	:param float interval: window of time in seconds to apply RSAM analysis.
	:param str cha: listening channel (defaults to [S,E]HZ)
	:param str deconv: ``'VEL'``, ``'ACC'``, ``'GRAV'``, ``'DISP'``, or ``'CHAN'``
	:param str fwaddr: Specify a forwarding address to send RSAM in a UDP packet
	:param str fwport: Specify a forwarding port to send RSAM in a UDP packet
	:param str fwformat: ``'LITE'``, ``'JSON'``, or ``'CSV'``
	"""

	def __init__(self, q=False, debug=False, interval=5, cha='HZ', deconv=False,
				 fwaddr=False, fwport=False, fwformat='LITE', *args, **kwargs):
		"""
		Initializes the RSAM analysis thread.
		"""
		super().__init__()
		self.sender = 'RSAM'
		self.alive = True
		self.debug = debug
		self.stn = rs.stn
		self.fwaddr = fwaddr
		self.fwport = fwport
		self.fwformat = fwformat.upper()
		self.sock = False
		self.interval = interval
		self.default_ch = 'HZ'
		self.args = args
		self.kwargs = kwargs
		self.raw = rs.Stream()
		self.stream = rs.Stream()
		self.units = 'counts'

		self._set_deconv(deconv)

		self._set_channel(cha)

		self.rsam = [1, 1, 1]

		if q:
			self.queue = q
		else:
			printE('no queue passed to the consumer thread! We will exit now!',
				   self.sender)
			sys.stdout.flush()
			self.alive = False
			sys.exit()

		printM('Starting.', self.sender)


	def _set_deconv(self, deconv):

		"""
		This function sets the deconvolution units. Allowed values are as follows:

		.. |ms2| replace:: m/s\ :sup:`2`\

		- ``'VEL'`` - velocity (m/s)
		- ``'ACC'`` - acceleration (|ms2|)
		- ``'GRAV'`` - fraction of acceleration due to gravity (g, or 9.81 |ms2|)
		- ``'DISP'`` - displacement (m)
		- ``'CHAN'`` - channel-specific unit calculation, i.e. ``'VEL'`` for geophone channels and ``'ACC'`` for accelerometer channels

		:param str deconv: ``'VEL'``, ``'ACC'``, ``'GRAV'``, ``'DISP'``, or ``'CHAN'``
		"""
		deconv = deconv.upper() if deconv else False
		self.deconv = deconv if (deconv in rs.UNITS) else False
		if self.deconv and rs.inv:
			self.units = '%s (%s)' % (rs.UNITS[self.deconv][0], rs.UNITS[self.deconv][1]) if (self.deconv in rs.UNITS) else self.units
			printM('Signal deconvolution set to %s' % (self.deconv), self.sender)
		else:
			self.units = rs.UNITS['CHAN'][1]
			self.deconv = False
		printM('RSAM stream units are %s' % (self.units.strip(' ').lower()), self.sender)


	def _find_chn(self):
		"""
		Finds channel match in list of channels.
		"""
		for chn in rs.chns:
			if self.cha in chn:
				self.cha = chn


	def _set_channel(self, cha):

		"""
		This function sets the channel to listen to. Allowed values are as follows:

		- "SHZ"``, ``"EHZ"``, ``"EHN"`` or ``"EHE"`` - velocity channels
		- ``"ENZ"``, ``"ENN"``, ``"ENE"`` - acceleration channels
		- ``"HDF"`` - pressure transducer channel
		- ``"all"`` - resolves to either ``"EHZ"`` or ``"SHZ"`` if available

		:param cha: the channel to listen to
		:type cha: str
		"""
		cha = self.default_ch if (cha == 'all') else cha
		self.cha = cha if isinstance(cha, str) else cha[0]

		if self.cha in str(rs.chns):
			self._find_chn()
		else:
			printE('Could not find channel %s in list of channels! Please correct and restart.' % self.cha, self.sender)
			sys.exit(2)


	def _getq(self):

		"""
		Reads data from the queue and updates the stream.

		:rtype: bool
		:return: Returns ``True`` if stream is updated, otherwise ``False``.
		"""
		d = self.queue.get(True, timeout=None)
		self.queue.task_done()
		if self.cha in str(d):
			self.raw = rs.update_stream(stream=self.raw, d=d, fill_value='latest')
			return True
		elif 'TERM' in str(d):
			self.alive = False
			printM('Exiting.', self.sender)
			sys.exit()
		else:
			return False


	def _deconvolve(self):
		"""
		Deconvolves the stream associated with this class.
		"""
		if self.deconv:
			helpers.deconvolve(self)


	def _subloop(self):
		"""
		Gets the queue and figures out whether or not the specified channel is in the packet.
		"""
		while True:
			if self.queue.qsize() > 0:
				self._getq()			# get recent packets
			else:
				if self._getq():		# is this the specified channel? if so break
					break


	def _rsam(self):
		"""
		Run the RSAM analysis
		"""
		arr = [abs(el) for el in self.stream[0].data]
		meanv = statistics.mean(arr)
		medianv = statistics.median(arr)
		minv = min(arr)
		maxv = max(arr)
		self.rsam = [meanv, medianv, minv, maxv]


	def _print_rsam(self):
		"""
		Print the current RSAM analysis
		"""
		if self.debug:
			msg = '%s [%s] Current RSAM: mean %s median %s min %s max %s' % (
				(self.stream[0].stats.starttime + timedelta(seconds=
															len(self.stream[0].data) * self.stream[0].stats.delta)).strftime('%Y-%m-%d %H:%M:%S'),
				self.sender,
				self.rsam[0],
				self.rsam[1],
				self.rsam[2],
				self.rsam[3]
			)
			printM(msg, self.sender)

	def _forward_rsam(self):
		"""
		Send the RSAM analysis via UDP to another destination in a lightweight format
		"""
		if self.sock:
			msg = 'stn:%s|ch:%s|mean:%s|med:%s|min:%s|max:%s' % (self.stn, self.cha, self.rsam[0], self.rsam[1], self.rsam[2], self.rsam[3])
			if self.fwformat is 'JSON':
				msg = '{"station":"%s","channel":"%s","mean":%s,"median":%s,"min":%s,"max":%s}' \
					  % (self.stn, self.cha, self.rsam[0], self.rsam[1], self.rsam[2], self.rsam[3])
			elif self.fwformat is 'CSV':
				msg = '%s,%s,%s,%s,%s,%s' \
					  % (self.stn, self.cha, self.rsam[0], self.rsam[1], self.rsam[2], self.rsam[3])
			packet = bytes(msg, 'utf-8')
			self.sock.sendto(packet, (self.fwaddr, self.fwport))


	def run(self):
		"""
		Reads data from the queue and executes self.codefile if it sees an ``ALARM`` message.
		Quits if it sees a ``TERM`` message.
		"""
		if self.fwaddr and self.fwport:
			printM('Opening socket...', sender=self.sender)
			socket_type = s.SOCK_DGRAM if os.name in 'nt' else s.SOCK_DGRAM | s.SO_REUSEADDR
			self.sock = s.socket(s.AF_INET, socket_type)

		n = 0
		next_int = time.time() + self.interval

		wait_pkts = self.interval / (rs.tf / 1000)

		while n > 3:
			self.getq()
			n += 1

		n = 0
		while True:
			self._subloop()

			self.raw = rs.copy(self.raw)	# necessary to avoid memory leak
			self.stream = self.raw.copy()
			self._deconvolve()

			if n > wait_pkts:
				# if the trigger is activated
				obstart = self.stream[0].stats.endtime - timedelta(seconds=self.interval)	# obspy time
				self.raw = self.raw.slice(starttime=obstart)		# slice the stream to the specified length (seconds variable)
				self.stream = self.stream.slice(starttime=obstart)	# slice the stream to the specified length (seconds variable)

				# run rsam analysis
				if time.time() > next_int:
					self._rsam()
					self.stream = rs.copy(self.stream)  # prevent mem leak
					self._forward_rsam()
					self._print_rsam()
					next_int = time.time() + self.interval

			elif n == 0:
				printM('Starting RSAM analysis with interval=%s on station=%s channel=%s forward=%s' %
					   (self.interval, self.stn, self.cha, self.fwaddr),
					   self.sender)
			elif n == wait_pkts:
				printM('RSAM analysis up and running normally.', self.sender)
			else:
				pass

			n += 1
			sys.stdout.flush()


1		import sys, os, time
2		import socket as s
3		from datetime import datetime, timedelta
4		import statistics
5		from rsudp import printM, printW, printE
6		from rsudp import helpers
7		import rsudp.raspberryshake as rs
8		COLOR = {}
9		from rsudp import COLOR
10
11		# set the terminal text color to green
12		COLOR['current'] = COLOR['green']
13
14		class RSAM(rs.ConsumerThread):
15		"""
16		.. versionadded:: 1.0.0
17
18		A consumer class that runs an Real-time Seismic Amplitude Measurement (RSAM).
19
20		:param queue.Queue q: queue of data and messages sent by :class:`rsudp.c_consumer.Consumer`.
21		:param bool debug: whether or not to display RSAM analysis live to the console.
22		:param float interval: window of time in seconds to apply RSAM analysis.
23		:param str cha: listening channel (defaults to [S,E]HZ)
24		:param str deconv: ``'VEL'``, ``'ACC'``, ``'GRAV'``, ``'DISP'``, or ``'CHAN'``
25		:param str fwaddr: Specify a forwarding address to send RSAM in a UDP packet
26		:param str fwport: Specify a forwarding port to send RSAM in a UDP packet
27		:param str fwformat: ``'LITE'``, ``'JSON'``, or ``'CSV'``
28		"""
29
30		def __init__(self, q=False, debug=False, interval=5, cha='HZ', deconv=False,
31		fwaddr=False, fwport=False, fwformat='LITE', args, *kwargs):
32		"""
33		Initializes the RSAM analysis thread.
34		"""
35		super().__init__()
36		self.sender = 'RSAM'
37		self.alive = True
38		self.debug = debug
39		self.stn = rs.stn
40		self.fwaddr = fwaddr
41		self.fwport = fwport
42		self.fwformat = fwformat.upper()
43		self.sock = False
44		self.interval = interval
45		self.default_ch = 'HZ'
46		self.args = args
47		self.kwargs = kwargs
48		self.raw = rs.Stream()
49		self.stream = rs.Stream()
50		self.units = 'counts'
51
52		self._set_deconv(deconv)
53
54		self._set_channel(cha)
55
56		self.rsam = [1, 1, 1]
57
58		if q:
59		self.queue = q
60		else:
61		printE('no queue passed to the consumer thread! We will exit now!',
62		self.sender)
63		sys.stdout.flush()
64		self.alive = False
65		sys.exit()
66
67		printM('Starting.', self.sender)
68
69
70	View Code Duplication	def _set_deconv(self, deconv):
		0 ignored issues – show Duplication introduced 2020-04-18 03:47 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
71		"""
72		This function sets the deconvolution units. Allowed values are as follows:
73
74		.. \|ms2\| replace:: m/s\ :sup:`2`\
75
76		- ``'VEL'`` - velocity (m/s)
77		- ``'ACC'`` - acceleration (\|ms2\|)
78		- ``'GRAV'`` - fraction of acceleration due to gravity (g, or 9.81 \|ms2\|)
79		- ``'DISP'`` - displacement (m)
80		- ``'CHAN'`` - channel-specific unit calculation, i.e. ``'VEL'`` for geophone channels and ``'ACC'`` for accelerometer channels
81
82		:param str deconv: ``'VEL'``, ``'ACC'``, ``'GRAV'``, ``'DISP'``, or ``'CHAN'``
83		"""
84		deconv = deconv.upper() if deconv else False
85		self.deconv = deconv if (deconv in rs.UNITS) else False
86		if self.deconv and rs.inv:
87		self.units = '%s (%s)' % (rs.UNITS[self.deconv][0], rs.UNITS[self.deconv][1]) if (self.deconv in rs.UNITS) else self.units
88		printM('Signal deconvolution set to %s' % (self.deconv), self.sender)
89		else:
90		self.units = rs.UNITS['CHAN'][1]
91		self.deconv = False
92		printM('RSAM stream units are %s' % (self.units.strip(' ').lower()), self.sender)
93
94
95		def _find_chn(self):
96		"""
97		Finds channel match in list of channels.
98		"""
99		for chn in rs.chns:
100		if self.cha in chn:
101		self.cha = chn
102
103
104	View Code Duplication	def _set_channel(self, cha):
		0 ignored issues – show Duplication introduced 2020-04-18 03:47 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
105		"""
106		This function sets the channel to listen to. Allowed values are as follows:
107
108		- "SHZ"``, ``"EHZ"``, ``"EHN"`` or ``"EHE"`` - velocity channels
109		- ``"ENZ"``, ``"ENN"``, ``"ENE"`` - acceleration channels
110		- ``"HDF"`` - pressure transducer channel
111		- ``"all"`` - resolves to either ``"EHZ"`` or ``"SHZ"`` if available
112
113		:param cha: the channel to listen to
114		:type cha: str
115		"""
116		cha = self.default_ch if (cha == 'all') else cha
117		self.cha = cha if isinstance(cha, str) else cha[0]
118
119		if self.cha in str(rs.chns):
120		self._find_chn()
121		else:
122		printE('Could not find channel %s in list of channels! Please correct and restart.' % self.cha, self.sender)
123		sys.exit(2)
124
125
126	View Code Duplication	def _getq(self):
		0 ignored issues – show Duplication introduced 2020-04-18 03:47 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
127		"""
128		Reads data from the queue and updates the stream.
129
130		:rtype: bool
131		:return: Returns ``True`` if stream is updated, otherwise ``False``.
132		"""
133		d = self.queue.get(True, timeout=None)
134		self.queue.task_done()
135		if self.cha in str(d):
136		self.raw = rs.update_stream(stream=self.raw, d=d, fill_value='latest')
137		return True
138		elif 'TERM' in str(d):
139		self.alive = False
140		printM('Exiting.', self.sender)
141		sys.exit()
142		else:
143		return False
144
145
146		def _deconvolve(self):
147		"""
148		Deconvolves the stream associated with this class.
149		"""
150		if self.deconv:
151		helpers.deconvolve(self)
152
153
154		def _subloop(self):
155		"""
156		Gets the queue and figures out whether or not the specified channel is in the packet.
157		"""
158		while True:
159		if self.queue.qsize() > 0:
160		self._getq() # get recent packets
161		else:
162		if self._getq(): # is this the specified channel? if so break
163		break
164
165
166		def _rsam(self):
167		"""
168		Run the RSAM analysis
169		"""
170		arr = [abs(el) for el in self.stream[0].data]
171		meanv = statistics.mean(arr)
172		medianv = statistics.median(arr)
173		minv = min(arr)
174		maxv = max(arr)
175		self.rsam = [meanv, medianv, minv, maxv]
176
177
178		def _print_rsam(self):
179		"""
180		Print the current RSAM analysis
181		"""
182		if self.debug:
183		msg = '%s [%s] Current RSAM: mean %s median %s min %s max %s' % (
184		(self.stream[0].stats.starttime + timedelta(seconds=
185		len(self.stream[0].data) * self.stream[0].stats.delta)).strftime('%Y-%m-%d %H:%M:%S'),
186		self.sender,
187		self.rsam[0],
188		self.rsam[1],
189		self.rsam[2],
190		self.rsam[3]
191		)
192		printM(msg, self.sender)
193
194		def _forward_rsam(self):
195		"""
196		Send the RSAM analysis via UDP to another destination in a lightweight format
197		"""
198		if self.sock:
199		msg = 'stn:%s\|ch:%s\|mean:%s\|med:%s\|min:%s\|max:%s' % (self.stn, self.cha, self.rsam[0], self.rsam[1], self.rsam[2], self.rsam[3])
200		if self.fwformat is 'JSON':
201		msg = '{"station":"%s","channel":"%s","mean":%s,"median":%s,"min":%s,"max":%s}' \
202		% (self.stn, self.cha, self.rsam[0], self.rsam[1], self.rsam[2], self.rsam[3])
203		elif self.fwformat is 'CSV':
204		msg = '%s,%s,%s,%s,%s,%s' \
205		% (self.stn, self.cha, self.rsam[0], self.rsam[1], self.rsam[2], self.rsam[3])
206		packet = bytes(msg, 'utf-8')
207		self.sock.sendto(packet, (self.fwaddr, self.fwport))
208
209
210		def run(self):
211		"""
212		Reads data from the queue and executes self.codefile if it sees an ``ALARM`` message.
213		Quits if it sees a ``TERM`` message.
214		"""
215		if self.fwaddr and self.fwport:
216		printM('Opening socket...', sender=self.sender)
217		socket_type = s.SOCK_DGRAM if os.name in 'nt' else s.SOCK_DGRAM \| s.SO_REUSEADDR
218		self.sock = s.socket(s.AF_INET, socket_type)
219
220		n = 0
221		next_int = time.time() + self.interval
222
223		wait_pkts = self.interval / (rs.tf / 1000)
224
225		while n > 3:
226		self.getq()
227		n += 1
228
229		n = 0
230		while True:
231		self._subloop()
232
233		self.raw = rs.copy(self.raw) # necessary to avoid memory leak
234		self.stream = self.raw.copy()
235		self._deconvolve()
236
237		if n > wait_pkts:
238		# if the trigger is activated
239		obstart = self.stream[0].stats.endtime - timedelta(seconds=self.interval) # obspy time
240		self.raw = self.raw.slice(starttime=obstart) # slice the stream to the specified length (seconds variable)
241		self.stream = self.stream.slice(starttime=obstart) # slice the stream to the specified length (seconds variable)
242
243		# run rsam analysis
244		if time.time() > next_int:
245		self._rsam()
246		self.stream = rs.copy(self.stream) # prevent mem leak
247		self._forward_rsam()
248		self._print_rsam()
249		next_int = time.time() + self.interval
250
251		elif n == 0:
252		printM('Starting RSAM analysis with interval=%s on station=%s channel=%s forward=%s' %
253		(self.interval, self.stn, self.cha, self.fwaddr),
254		self.sender)
255		elif n == wait_pkts:
256		printM('RSAM analysis up and running normally.', self.sender)
257		else:
258		pass
259
260		n += 1
261		sys.stdout.flush()
262

raspishake / rsudp

Pull Request — master (#6)

build.rsudp.c_rsam.RSAM._set_deconv() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like