build.rsudp.helpers.deconv_acc_inst() - Code Metrics - Inspection of "refactoring again" - raspishake/rsudp - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( fe5590...f40c33 )

by Ian

created 2020-03-25 04:25 UTC

build.rsudp.helpers.deconv_acc_inst() B

↳ Parent: build.rsudp.helpers

Complexity

Conditions

Size

Total Lines	30
Code Lines	18

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	18
dl	0
loc	30
rs	8
c	0
b	0
f	0
cc	7
nop	3

import rsudp.raspberryshake as rs
from rsudp import COLOR, printM, printW
import os
import json


def dump_default(settings_loc, default_settings):
	'''
	Dumps a default settings file to a specified location.

	:param str settings_loc: The location to create the new settings JSON.
	:param str default_settings: The default settings to dump to file.
	'''
	print('Creating a default settings file at %s' % settings_loc)
	with open(settings_loc, 'w+') as f:
		f.write(default_settings)
		f.write('\n')


def default_settings(output_dir='%s/rsudp' % os.path.expanduser('~').replace('\\', '/'), verbose=True):
	'''
	Returns a formatted json string of default settings.

	:param str output_dir: the user's specified output location. defaults to ``~/rsudp``.
	:param bool verbose: if ``True``, displays some information as the string is created.
	:return: default settings string in formatted json
	:rtype: str
	'''
	def_settings = r"""{
"settings": {
    "port": 8888,
    "station": "Z0000",
    "output_dir": "%s",
    "debug": true},
"printdata": {
    "enabled": false},
"write": {
    "enabled": false,
    "channels": ["all"]},
"plot": {
    "enabled": true,
    "duration": 30,
    "spectrogram": true,
    "fullscreen": false,
    "kiosk": false,
    "eq_screenshots": false,
    "channels": ["HZ", "HDF"],
    "deconvolve": false,
    "units": "CHAN"},
"forward": {
    "enabled": false,
    "address": "192.168.1.254",
    "port": 8888,
    "channels": ["all"]},
"alert": {
    "enabled": true,
    "channel": "HZ",
    "sta": 6,
    "lta": 30,
    "threshold": 1.7,
    "reset": 1.6,
    "highpass": 0,
    "lowpass": 50,
    "deconvolve": false,
    "units": "VEL"},
"alertsound": {
    "enabled": false,
    "mp3file": "doorbell"},
"custom": {
    "enabled": false,
    "codefile": "n/a",
    "win_override": false},
"tweets": {
    "enabled": false,
    "tweet_images": true,
    "api_key": "n/a",
    "api_secret": "n/a",
    "access_token": "n/a",
    "access_secret": "n/a"},
"telegram": {
    "enabled": false,
    "send_images": true,
    "token": "n/a",
    "chat_id": "n/a"}
}

""" % (output_dir)
	if verbose:
		print('By default output_dir is set to %s' % output_dir)
	return def_settings


def read_settings(loc):
	'''
	Reads settings from a specific location.

	:param str loc: location on disk to read json settings file from
	:return: settings dictionary read from JSON, or ``None``
	:rtype: dict or NoneType
	'''
	settings_loc = os.path.abspath(os.path.expanduser(loc)).replace('\\', '/')
	settings = None
	with open(settings_loc, 'r') as f:
		try:
			data = f.read().replace('\\', '/')
			settings = json.loads(data)
		except Exception as e:
			print(COLOR['red'] + 'ERROR: Could not load settings file. Perhaps the JSON is malformed?' + COLOR['white'])
			print(COLOR['red'] + '       detail: %s' % e + COLOR['white'])
			print(COLOR['red'] + '       If you would like to overwrite and rebuild the file, you can enter the command below:' + COLOR['white'])
			print(COLOR['bold'] + '       shake_client -d %s' % loc + COLOR['white'])
			exit(2)
	return settings


def set_channels(self, cha):
	'''
	This function sets the channels available for plotting. Allowed units are as follows:

	- ``["SHZ", "EHZ", "EHN", "EHE"]`` - velocity channels
	- ``["ENZ", "ENN", "ENE"]`` - acceleration channels
	- ``["HDF"]`` - pressure transducer channel
	- ``["all"]`` - all available channels

	So for example, if you wanted to display the two vertical channels of a Shake 4D,
	(geophone and vertical accelerometer) you could specify:

	``["EHZ", "ENZ"]``

	You can also specify partial channel names.
	So for example, the following will display at least one channel from any
	Raspberry Shake instrument:

	``["HZ", "HDF"]``

	Or if you wanted to display only vertical channels from a RS4D,
	you could specify

	``["Z"]``

	which would match both ``"EHZ"`` and ``"ENZ"``.

	:param self self: self object of the class calling this function
	:param cha: the channel or list of channels to plot
	:type cha: list or str
	'''
	cha = rs.chns if ('all' in cha) else cha
	cha = list(cha) if isinstance(cha, str) else cha
	for c in rs.chns:
		n = 0
		for uch in cha:
			if (uch.upper() in c) and (c not in str(self.chans)):
				self.chans.append(c)
			n += 1
	if len(self.chans) < 1:
			self.chans = rs.chns


def fsec(ti):
	'''
	.. versionadded:: 0.4.3

	The Raspberry Shake records at hundredths-of-a-second precision.
	In order to report time at this precision, we need to do some time-fu.

	This function rounds the microsecond fraction of a
	:py:class:`obspy.core.utcdatetime.UTCDateTime`
	depending on its precision, so that it accurately reflects the Raspberry Shake's
	event measurement precision.

	This is necessary because datetime objects in Python are strange and confusing, and
	strftime doesn't support fractional returns, only the full integer microsecond field
	which is an integer right-padded with zeroes. This function uses the ``precision``
	of a datetime object.

	For example:

	.. code-block:: python

		>>> from obspy import UTCDateTime
		>>> ti = UTCDateTime(2020, 1, 1, 0, 0, 0, 599000, precision=3)
		>>> fsec(ti)
		UTCDateTime(2020, 1, 1, 0, 0, 0, 600000)

	:param ti: time object to convert microseconds for
	:type ti: obspy.core.utcdatetime.UTCDateTime
	:return: the hundredth-of-a-second rounded version of the time object passed (precision is 0.01 second)
	:rtype: obspy.core.utcdatetime.UTCDateTime
	'''
	# time in python is weird and confusing, but luckily obspy is better than Python
	# at dealing with datetimes. all we need to do is tell it what precision we want
	# and it handles the rounding for us.
	return rs.UTCDateTime(ti, precision=2)


def conn_stats(TESTING=False):
	'''
	Print some stats about the connection.

	Example:

	.. code-block:: python

		>>> conn_stats()
		2020-03-25 01:35:04 [conn_stats] Initialization stats:
		2020-03-25 01:35:04 [conn_stats]                 Port: 18069
		2020-03-25 01:35:04 [conn_stats]   Sending IP address: 192.168.0.4
		2020-03-25 01:35:04 [conn_stats]     Set station name: R24FA
		2020-03-25 01:35:04 [conn_stats]   Number of channels: 4
		2020-03-25 01:35:04 [conn_stats]   Transmission freq.: 250 ms/packet
		2020-03-25 01:35:04 [conn_stats]    Transmission rate: 4 packets/sec
		2020-03-25 01:35:04 [conn_stats]   Samples per second: 100 sps
		2020-03-25 01:35:04 [conn_stats]            Inventory: AM.R24FA (Raspberry Shake Citizen Science Station)
	'''
	s = 'conn_stats'
	pf = printW if TESTING else printM
	pf('Initialization stats:', sender=s, announce=False)
	pf('                Port: %s' % rs.port, sender=s, announce=False)
	pf('  Sending IP address: %s' % rs.firstaddr, sender=s, announce=False)
	pf('    Set station name: %s' % rs.stn, sender=s, announce=False)
	pf('  Number of channels: %s' % rs.numchns, sender=s, announce=False)
	pf('  Transmission freq.: %s ms/packet' % rs.tf, sender=s, announce=False)
	pf('   Transmission rate: %s packets/sec' % rs.tr, sender=s, announce=False)
	pf('  Samples per second: %s sps' % rs.sps, sender=s, announce=False)
	if rs.inv:
		pf('           Inventory: %s' % rs.inv.get_contents()['stations'][0],
			   sender=s, announce=False)


def msg_alarm(event_time):
	'''
	This function constructs the ``ALARM`` message as a bytes object.
	Currently this is only used by :py:class:`rsudp.p_producer.Producer`
	to construct alarm queue messages.

	For example:

	.. code-block:: python

		>>> from obspy import UTCDateTime
		>>> ti = UTCDateTime(2020, 1, 1, 0, 0, 0, 599000, precision=3)
		>>> msg_alarm(ti)
		b'ALARM 2020-01-01T00:00:00.599Z'

	:param obspy.core.utcdatetime.UTCDateTime event_time: the datetime object to serialize and convert to bytes
	:rtype: bytes
	:return: the ``ALARM`` message, ready to be put on the queue
	'''
	return b'ALARM %s' % bytes(str(event_time), 'utf-8')


def msg_reset(reset_time):
	'''
	This function constructs the ``RESET`` message as a bytes object.
	Currently this is only used by :py:class:`rsudp.p_producer.Producer`
	to construct reset queue messages.

	For example:

	.. code-block:: python

		>>> from obspy import UTCDateTime
		>>> ti = UTCDateTime(2020, 1, 1, 0, 0, 0, 599000, precision=3)
		>>> msg_reset(ti)
		b'RESET 2020-01-01T00:00:00.599Z'

	:param obspy.core.utcdatetime.UTCDateTime reset_time: the datetime object to serialize and convert to bytes
	:rtype: bytes
	:return: the ``RESET`` message, ready to be put on the queue
	'''
	return b'RESET %s' % bytes(str(reset_time), 'utf-8')


def msg_imgpath(event_time, figname):
	'''
	This function constructs the ``IMGPATH`` message as a bytes object.
	Currently this is only used by :py:class:`rsudp.c_plot.Plot`
	to construct queue messages containing timestamp and saved image path.

	For example:

	.. code-block:: python

		>>> from obspy import UTCDateTime
		>>> ti = UTCDateTime(2020, 1, 1, 0, 0, 0, 599000, precision=3)
		>>> path = '/home/pi/rsudp/screenshots/test.png'
		>>> msg_imgpath(ti, path)
		b'IMGPATH 2020-01-01T00:00:00.599Z /home/pi/rsudp/screenshots/test.png'

	:param obspy.core.utcdatetime.UTCDateTime event_time: the datetime object to serialize and convert to bytes
	:param str figname: the figure path as a string
	:rtype: bytes
	:return: the ``IMGPATH`` message, ready to be put on the queue
	'''
	return b'IMGPATH %s %s' % (bytes(str(event_time), 'utf-8'), bytes(str(figname), 'utf-8'))


def msg_term():
	'''
	This function constructs the simple ``TERM`` message as a bytes object.

	.. code-block:: python

		>>> msg_term()
		b'TERM'


	:rtype: bytes
	:return: the ``TERM`` message
	'''
	return b'TERM'


def get_msg_time(msg):
	'''
	This function gets the time from ``ALARM``, ``RESET``,
	and ``IMGPATH`` messages as a UTCDateTime object.

	For example:

	.. code-block:: python

		>>> from obspy import UTCDateTime
		>>> ti = UTCDateTime(2020, 1, 1, 0, 0, 0, 599000, precision=3)
		>>> path = '/home/pi/rsudp/screenshots/test.png'
		>>> msg = msg_imgpath(ti, path)
		>>> msg
		b'IMGPATH 2020-01-01T00:00:00.599Z /home/pi/rsudp/screenshots/test.png'
		>>> get_msg_time(msg)
		UTCDateTime(2020, 1, 1, 0, 0, 0, 599000)

	:param bytes msg: the bytes-formatted queue message to decode
	:rtype: obspy.core.utcdatetime.UTCDateTime
	:return: the time embedded in the message
	'''
	return rs.UTCDateTime.strptime(msg.decode('utf-8').split(' ')[1], '%Y-%m-%dT%H:%M:%S.%fZ')


def get_msg_path(msg):
	'''
	This function gets the path from ``IMGPATH`` messages as a string.

	For example:

	.. code-block:: python

		>>> from obspy import UTCDateTime
		>>> ti = UTCDateTime(2020, 1, 1, 0, 0, 0, 599000, precision=3)
		>>> path = '/home/pi/rsudp/screenshots/test.png'
		>>> msg = msg_imgpath(ti, path)
		>>> msg
		b'IMGPATH 2020-01-01T00:00:00.599Z /home/pi/rsudp/screenshots/test.png'
		>>> get_msg_path(msg)
		'/home/pi/rsudp/screenshots/test.png'

	:param bytes msg: the bytes-formatted queue message to decode
	:rtype: str
	:return: the path embedded in the message
	'''
	return msg.decode('utf-8').split(' ')[2]


def deconv_vel_inst(self, trace, output):
	'''
	.. role:: pycode(code)
		:language: python
	
	A helper function for :py:func:`rsudp.raspberryshake.deconvolve`
	for velocity channels.

	:param self self: The self object of the sub-consumer class calling this function.
	:param osbpy.core.trace.Trace trace: the trace object instance to deconvolve
	'''
	if self.deconv not in 'CHAN':
		trace.remove_response(inventory=rs.inv, pre_filt=[0.1, 0.6, 0.95*self.sps, self.sps],
								output=output, water_level=4.5, taper=False)
	else:
		trace.remove_response(inventory=rs.inv, pre_filt=[0.1, 0.6, 0.95*self.sps, self.sps],
								output='VEL', water_level=4.5, taper=False)
	if 'ACC' in self.deconv:
		trace.data = rs.np.gradient(trace.data, 1)
	elif 'GRAV' in self.deconv:
		trace.data = rs.np.gradient(trace.data, 1) / rs.g
		trace.stats.units = 'Earth gravity'
	elif 'DISP' in self.deconv:
		trace.data = rs.np.cumsum(trace.data)
		trace.taper(max_percentage=0.1, side='left', max_length=1)
		trace.detrend(type='demean')
	else:
		trace.stats.units = 'Velocity'


def deconv_acc_inst(self, trace, output):
	'''
	.. role:: pycode(code)
		:language: python
	
	A helper function for :py:func:`rsudp.raspberryshake.deconvolve`
	for acceleration channels.

	:param self self: The self object of the sub-consumer class calling this function.
	:param osbpy.core.trace.Trace trace: the trace object instance to deconvolve
	'''
	if self.deconv not in 'CHAN':
		trace.remove_response(inventory=rs.inv, pre_filt=[0.1, 0.6, 0.95*self.sps, self.sps],
								output=output, water_level=4.5, taper=False)
	else:
		trace.remove_response(inventory=rs.inv, pre_filt=[0.1, 0.6, 0.95*self.sps, self.sps],
								output='ACC', water_level=4.5, taper=False)
	if 'VEL' in self.deconv:
		trace.data = rs.np.cumsum(trace.data)
		trace.detrend(type='demean')
	elif 'DISP' in self.deconv:
		trace.data = rs.np.cumsum(rs.np.cumsum(trace.data))
		trace.detrend(type='linear')
	elif 'GRAV' in self.deconv:
		trace.data = trace.data / rs.g
		trace.stats.units = 'Earth gravity'
	else:
		trace.stats.units = 'Acceleration'
	if ('ACC' not in self.deconv) and ('CHAN' not in self.deconv):
		trace.taper(max_percentage=0.1, side='left', max_length=1)


def deconv_rbm_inst(self, trace, output):
	'''
	.. role:: pycode(code)
		:language: python
	
	A helper function for :py:func:`rsudp.raspberryshake.deconvolve`
	for Raspberry Boom pressure transducer channels.

	.. note::

		The Raspberry Boom pressure transducer does not currently have a
		deconvolution function. The Raspberry Shake team is working on a
		calibration for the Boom, but until then Boom units are given in
		counts.

	:param self self: The self object of the sub-consumer class calling this function.
	:param osbpy.core.trace.Trace trace: the trace object instance to deconvolve
	'''
	trace.stats.units = ' counts'


def deconvolve(self):
	'''
	.. role:: pycode(code)
		:language: python
	
	A central helper function for sub-consumers (i.e. :py:class:`rsudp.c_plot.Plot` or :py:class:`rsudp.c_alert.Alert`)
	that need to deconvolve their raw data to metric units.
	Consumers with :py:class:`obspy.core.stream.Stream` objects in :pycode:`self.stream` can use this to deconvolve data
	if this library's :pycode:`rsudp.raspberryshake.inv` variable
	contains a valid :py:class:`obspy.core.inventory.inventory.Inventory` object.

	:param self self: The self object of the sub-consumer class calling this function. Must contain :pycode:`self.stream` as a :py:class:`obspy.core.stream.Stream` object.
	'''
	acc_channels = ['ENE', 'ENN', 'ENZ']
	vel_channels = ['EHE', 'EHN', 'EHZ', 'SHZ']
	rbm_channels = ['HDF']

	self.stream = self.raw.copy()
	for trace in self.stream:
		trace.stats.units = self.units
		output = 'ACC' if self.deconv == 'GRAV' else self.deconv	# if conversion is to gravity
		if self.deconv:
			if trace.stats.channel in vel_channels:
				deconv_vel_inst(self, trace, output)	# geophone channels

			elif trace.stats.channel in acc_channels:
				deconv_acc_inst(self, trace, output)	# accelerometer channels

			elif trace.stats.channel in rbm_channels:
				deconv_rbm_inst(self, trace, output)	# this is the Boom channel

			else:
				trace.stats.units = ' counts'	# this is a new one

		else:
			trace.stats.units = ' counts'		# this is not being deconvolved




1			import rsudp.raspberryshake as rs
2			from rsudp import COLOR, printM, printW
3			import os
4			import json
5
6
7			def dump_default(settings_loc, default_settings):
8			'''
9			Dumps a default settings file to a specified location.
10
11			:param str settings_loc: The location to create the new settings JSON.
12			:param str default_settings: The default settings to dump to file.
13			'''
14			print('Creating a default settings file at %s' % settings_loc)
15			with open(settings_loc, 'w+') as f:
16			f.write(default_settings)
17			f.write('\n')
18
19
20			def default_settings(output_dir='%s/rsudp' % os.path.expanduser('~').replace('\\', '/'), verbose=True):
21			'''
22			Returns a formatted json string of default settings.
23
24			:param str output_dir: the user's specified output location. defaults to ``~/rsudp``.
25			:param bool verbose: if ``True``, displays some information as the string is created.
26			:return: default settings string in formatted json
27			:rtype: str
28			'''
29			def_settings = r"""{
30			"settings": {
31			"port": 8888,
32			"station": "Z0000",
33			"output_dir": "%s",
34			"debug": true},
35			"printdata": {
36			"enabled": false},
37			"write": {
38			"enabled": false,
39			"channels": ["all"]},
40			"plot": {
41			"enabled": true,
42			"duration": 30,
43			"spectrogram": true,
44			"fullscreen": false,
45			"kiosk": false,
46			"eq_screenshots": false,
47			"channels": ["HZ", "HDF"],
48			"deconvolve": false,
49			"units": "CHAN"},
50			"forward": {
51			"enabled": false,
52			"address": "192.168.1.254",
53			"port": 8888,
54			"channels": ["all"]},
55			"alert": {
56			"enabled": true,
57			"channel": "HZ",
58			"sta": 6,
59			"lta": 30,
60			"threshold": 1.7,
61			"reset": 1.6,
62			"highpass": 0,
63			"lowpass": 50,
64			"deconvolve": false,
65			"units": "VEL"},
66			"alertsound": {
67			"enabled": false,
68			"mp3file": "doorbell"},
69			"custom": {
70			"enabled": false,
71			"codefile": "n/a",
72			"win_override": false},
73			"tweets": {
74			"enabled": false,
75			"tweet_images": true,
76			"api_key": "n/a",
77			"api_secret": "n/a",
78			"access_token": "n/a",
79			"access_secret": "n/a"},
80			"telegram": {
81			"enabled": false,
82			"send_images": true,
83			"token": "n/a",
84			"chat_id": "n/a"}
85			}
86
87			""" % (output_dir)
88			if verbose:
89			print('By default output_dir is set to %s' % output_dir)
90			return def_settings
91
92
93			def read_settings(loc):
94			'''
95			Reads settings from a specific location.
96
97			:param str loc: location on disk to read json settings file from
98			:return: settings dictionary read from JSON, or ``None``
99			:rtype: dict or NoneType
100			'''
101			settings_loc = os.path.abspath(os.path.expanduser(loc)).replace('\\', '/')
102			settings = None
103			with open(settings_loc, 'r') as f:
104			try:
105			data = f.read().replace('\\', '/')
106			settings = json.loads(data)
107			except Exception as e:
108			print(COLOR['red'] + 'ERROR: Could not load settings file. Perhaps the JSON is malformed?' + COLOR['white'])
109			print(COLOR['red'] + ' detail: %s' % e + COLOR['white'])
110			print(COLOR['red'] + ' If you would like to overwrite and rebuild the file, you can enter the command below:' + COLOR['white'])
111			print(COLOR['bold'] + ' shake_client -d %s' % loc + COLOR['white'])
112			exit(2)
113			return settings
114
115
116			def set_channels(self, cha):
117			'''
118			This function sets the channels available for plotting. Allowed units are as follows:
119
120			- ``["SHZ", "EHZ", "EHN", "EHE"]`` - velocity channels
121			- ``["ENZ", "ENN", "ENE"]`` - acceleration channels
122			- ``["HDF"]`` - pressure transducer channel
123			- ``["all"]`` - all available channels
124
125			So for example, if you wanted to display the two vertical channels of a Shake 4D,
126			(geophone and vertical accelerometer) you could specify:
127
128			``["EHZ", "ENZ"]``
129
130			You can also specify partial channel names.
131			So for example, the following will display at least one channel from any
132			Raspberry Shake instrument:
133
134			``["HZ", "HDF"]``
135
136			Or if you wanted to display only vertical channels from a RS4D,
137			you could specify
138
139			``["Z"]``
140
141			which would match both ``"EHZ"`` and ``"ENZ"``.
142
143			:param self self: self object of the class calling this function
144			:param cha: the channel or list of channels to plot
145			:type cha: list or str
146			'''
147			cha = rs.chns if ('all' in cha) else cha
148			cha = list(cha) if isinstance(cha, str) else cha
149			for c in rs.chns:
150			n = 0
151			for uch in cha:
152			if (uch.upper() in c) and (c not in str(self.chans)):
153			self.chans.append(c)
154			n += 1
155			if len(self.chans) < 1:
156			self.chans = rs.chns
157
158
159			def fsec(ti):
160			'''
161			.. versionadded:: 0.4.3
162
163			The Raspberry Shake records at hundredths-of-a-second precision.
164			In order to report time at this precision, we need to do some time-fu.
165
166			This function rounds the microsecond fraction of a
167			:py:class:`obspy.core.utcdatetime.UTCDateTime`
168			depending on its precision, so that it accurately reflects the Raspberry Shake's
169			event measurement precision.
170
171			This is necessary because datetime objects in Python are strange and confusing, and
172			strftime doesn't support fractional returns, only the full integer microsecond field
173			which is an integer right-padded with zeroes. This function uses the ``precision``
174			of a datetime object.
175
176			For example:
177
178			.. code-block:: python
179
180			>>> from obspy import UTCDateTime
181			>>> ti = UTCDateTime(2020, 1, 1, 0, 0, 0, 599000, precision=3)
182			>>> fsec(ti)
183			UTCDateTime(2020, 1, 1, 0, 0, 0, 600000)
184
185			:param ti: time object to convert microseconds for
186			:type ti: obspy.core.utcdatetime.UTCDateTime
187			:return: the hundredth-of-a-second rounded version of the time object passed (precision is 0.01 second)
188			:rtype: obspy.core.utcdatetime.UTCDateTime
189			'''
190			# time in python is weird and confusing, but luckily obspy is better than Python
191			# at dealing with datetimes. all we need to do is tell it what precision we want
192			# and it handles the rounding for us.
193			return rs.UTCDateTime(ti, precision=2)
194
195
196			def conn_stats(TESTING=False):
197			'''
198			Print some stats about the connection.
199
200			Example:
201
202			.. code-block:: python
203
204			>>> conn_stats()
205			2020-03-25 01:35:04 [conn_stats] Initialization stats:
206			2020-03-25 01:35:04 [conn_stats] Port: 18069
207			2020-03-25 01:35:04 [conn_stats] Sending IP address: 192.168.0.4
208			2020-03-25 01:35:04 [conn_stats] Set station name: R24FA
209			2020-03-25 01:35:04 [conn_stats] Number of channels: 4
210			2020-03-25 01:35:04 [conn_stats] Transmission freq.: 250 ms/packet
211			2020-03-25 01:35:04 [conn_stats] Transmission rate: 4 packets/sec
212			2020-03-25 01:35:04 [conn_stats] Samples per second: 100 sps
213			2020-03-25 01:35:04 [conn_stats] Inventory: AM.R24FA (Raspberry Shake Citizen Science Station)
214			'''
215			s = 'conn_stats'
216			pf = printW if TESTING else printM
217			pf('Initialization stats:', sender=s, announce=False)
218			pf(' Port: %s' % rs.port, sender=s, announce=False)
219			pf(' Sending IP address: %s' % rs.firstaddr, sender=s, announce=False)
220			pf(' Set station name: %s' % rs.stn, sender=s, announce=False)
221			pf(' Number of channels: %s' % rs.numchns, sender=s, announce=False)
222			pf(' Transmission freq.: %s ms/packet' % rs.tf, sender=s, announce=False)
223			pf(' Transmission rate: %s packets/sec' % rs.tr, sender=s, announce=False)
224			pf(' Samples per second: %s sps' % rs.sps, sender=s, announce=False)
225			if rs.inv:
226			pf(' Inventory: %s' % rs.inv.get_contents()['stations'][0],
227			sender=s, announce=False)
228
229
230			def msg_alarm(event_time):
231			'''
232			This function constructs the ``ALARM`` message as a bytes object.
233			Currently this is only used by :py:class:`rsudp.p_producer.Producer`
234			to construct alarm queue messages.
235
236			For example:
237
238			.. code-block:: python
239
240			>>> from obspy import UTCDateTime
241			>>> ti = UTCDateTime(2020, 1, 1, 0, 0, 0, 599000, precision=3)
242			>>> msg_alarm(ti)
243			b'ALARM 2020-01-01T00:00:00.599Z'
244
245			:param obspy.core.utcdatetime.UTCDateTime event_time: the datetime object to serialize and convert to bytes
246			:rtype: bytes
247			:return: the ``ALARM`` message, ready to be put on the queue
248			'''
249			return b'ALARM %s' % bytes(str(event_time), 'utf-8')
250
251
252			def msg_reset(reset_time):
253			'''
254			This function constructs the ``RESET`` message as a bytes object.
255			Currently this is only used by :py:class:`rsudp.p_producer.Producer`
256			to construct reset queue messages.
257
258			For example:
259
260			.. code-block:: python
261
262			>>> from obspy import UTCDateTime
263			>>> ti = UTCDateTime(2020, 1, 1, 0, 0, 0, 599000, precision=3)
264			>>> msg_reset(ti)
265			b'RESET 2020-01-01T00:00:00.599Z'
266
267			:param obspy.core.utcdatetime.UTCDateTime reset_time: the datetime object to serialize and convert to bytes
268			:rtype: bytes
269			:return: the ``RESET`` message, ready to be put on the queue
270			'''
271			return b'RESET %s' % bytes(str(reset_time), 'utf-8')
272
273
274			def msg_imgpath(event_time, figname):
275			'''
276			This function constructs the ``IMGPATH`` message as a bytes object.
277			Currently this is only used by :py:class:`rsudp.c_plot.Plot`
278			to construct queue messages containing timestamp and saved image path.
279
280			For example:
281
282			.. code-block:: python
283
284			>>> from obspy import UTCDateTime
285			>>> ti = UTCDateTime(2020, 1, 1, 0, 0, 0, 599000, precision=3)
286			>>> path = '/home/pi/rsudp/screenshots/test.png'
287			>>> msg_imgpath(ti, path)
288			b'IMGPATH 2020-01-01T00:00:00.599Z /home/pi/rsudp/screenshots/test.png'
289
290			:param obspy.core.utcdatetime.UTCDateTime event_time: the datetime object to serialize and convert to bytes
291			:param str figname: the figure path as a string
292			:rtype: bytes
293			:return: the ``IMGPATH`` message, ready to be put on the queue
294			'''
295			return b'IMGPATH %s %s' % (bytes(str(event_time), 'utf-8'), bytes(str(figname), 'utf-8'))
296
297
298			def msg_term():
299			'''
300			This function constructs the simple ``TERM`` message as a bytes object.
301
302			.. code-block:: python
303
304			>>> msg_term()
305			b'TERM'
306
307
308			:rtype: bytes
309			:return: the ``TERM`` message
310			'''
311			return b'TERM'
312
313
314			def get_msg_time(msg):
315			'''
316			This function gets the time from ``ALARM``, ``RESET``,
317			and ``IMGPATH`` messages as a UTCDateTime object.
318
319			For example:
320
321			.. code-block:: python
322
323			>>> from obspy import UTCDateTime
324			>>> ti = UTCDateTime(2020, 1, 1, 0, 0, 0, 599000, precision=3)
325			>>> path = '/home/pi/rsudp/screenshots/test.png'
326			>>> msg = msg_imgpath(ti, path)
327			>>> msg
328			b'IMGPATH 2020-01-01T00:00:00.599Z /home/pi/rsudp/screenshots/test.png'
329			>>> get_msg_time(msg)
330			UTCDateTime(2020, 1, 1, 0, 0, 0, 599000)
331
332			:param bytes msg: the bytes-formatted queue message to decode
333			:rtype: obspy.core.utcdatetime.UTCDateTime
334			:return: the time embedded in the message
335			'''
336			return rs.UTCDateTime.strptime(msg.decode('utf-8').split(' ')[1], '%Y-%m-%dT%H:%M:%S.%fZ')
337
338
339			def get_msg_path(msg):
340			'''
341			This function gets the path from ``IMGPATH`` messages as a string.
342
343			For example:
344
345			.. code-block:: python
346
347			>>> from obspy import UTCDateTime
348			>>> ti = UTCDateTime(2020, 1, 1, 0, 0, 0, 599000, precision=3)
349			>>> path = '/home/pi/rsudp/screenshots/test.png'
350			>>> msg = msg_imgpath(ti, path)
351			>>> msg
352			b'IMGPATH 2020-01-01T00:00:00.599Z /home/pi/rsudp/screenshots/test.png'
353			>>> get_msg_path(msg)
354			'/home/pi/rsudp/screenshots/test.png'
355
356			:param bytes msg: the bytes-formatted queue message to decode
357			:rtype: str
358			:return: the path embedded in the message
359			'''
360			return msg.decode('utf-8').split(' ')[2]
361
362
363			def deconv_vel_inst(self, trace, output):
364			'''
365			.. role:: pycode(code)
366			:language: python
367
368			A helper function for :py:func:`rsudp.raspberryshake.deconvolve`
369			for velocity channels.
370
371			:param self self: The self object of the sub-consumer class calling this function.
372			:param osbpy.core.trace.Trace trace: the trace object instance to deconvolve
373			'''
374			if self.deconv not in 'CHAN':
375			trace.remove_response(inventory=rs.inv, pre_filt=[0.1, 0.6, 0.95*self.sps, self.sps],
376			output=output, water_level=4.5, taper=False)
377			else:
378			trace.remove_response(inventory=rs.inv, pre_filt=[0.1, 0.6, 0.95*self.sps, self.sps],
379			output='VEL', water_level=4.5, taper=False)
380			if 'ACC' in self.deconv:
381			trace.data = rs.np.gradient(trace.data, 1)
382			elif 'GRAV' in self.deconv:
383			trace.data = rs.np.gradient(trace.data, 1) / rs.g
384			trace.stats.units = 'Earth gravity'
385			elif 'DISP' in self.deconv:
386			trace.data = rs.np.cumsum(trace.data)
387			trace.taper(max_percentage=0.1, side='left', max_length=1)
388			trace.detrend(type='demean')
389			else:
390			trace.stats.units = 'Velocity'
391
392
393			def deconv_acc_inst(self, trace, output):
394			'''
395			.. role:: pycode(code)
396			:language: python
397
398			A helper function for :py:func:`rsudp.raspberryshake.deconvolve`
399			for acceleration channels.
400
401			:param self self: The self object of the sub-consumer class calling this function.
402			:param osbpy.core.trace.Trace trace: the trace object instance to deconvolve
403			'''
404			if self.deconv not in 'CHAN':
405			trace.remove_response(inventory=rs.inv, pre_filt=[0.1, 0.6, 0.95*self.sps, self.sps],
406			output=output, water_level=4.5, taper=False)
407			else:
408			trace.remove_response(inventory=rs.inv, pre_filt=[0.1, 0.6, 0.95*self.sps, self.sps],
409			output='ACC', water_level=4.5, taper=False)
410			if 'VEL' in self.deconv:
411			trace.data = rs.np.cumsum(trace.data)
412			trace.detrend(type='demean')
413			elif 'DISP' in self.deconv:
414			trace.data = rs.np.cumsum(rs.np.cumsum(trace.data))
415			trace.detrend(type='linear')
416			elif 'GRAV' in self.deconv:
417			trace.data = trace.data / rs.g
418			trace.stats.units = 'Earth gravity'
419			else:
420			trace.stats.units = 'Acceleration'
421			if ('ACC' not in self.deconv) and ('CHAN' not in self.deconv):
422			trace.taper(max_percentage=0.1, side='left', max_length=1)
423
424
425			def deconv_rbm_inst(self, trace, output):
426			'''
427			.. role:: pycode(code)
428			:language: python
429
430			A helper function for :py:func:`rsudp.raspberryshake.deconvolve`
431			for Raspberry Boom pressure transducer channels.
432
433			.. note::
434
435			The Raspberry Boom pressure transducer does not currently have a
436			deconvolution function. The Raspberry Shake team is working on a
437			calibration for the Boom, but until then Boom units are given in
438			counts.
439
440			:param self self: The self object of the sub-consumer class calling this function.
441			:param osbpy.core.trace.Trace trace: the trace object instance to deconvolve
442			'''
443			trace.stats.units = ' counts'
444
445
446			def deconvolve(self):
447			'''
448			.. role:: pycode(code)
449			:language: python
450
451			A central helper function for sub-consumers (i.e. :py:class:`rsudp.c_plot.Plot` or :py:class:`rsudp.c_alert.Alert`)
452			that need to deconvolve their raw data to metric units.
453			Consumers with :py:class:`obspy.core.stream.Stream` objects in :pycode:`self.stream` can use this to deconvolve data
454			if this library's :pycode:`rsudp.raspberryshake.inv` variable
455			contains a valid :py:class:`obspy.core.inventory.inventory.Inventory` object.
456
457			:param self self: The self object of the sub-consumer class calling this function. Must contain :pycode:`self.stream` as a :py:class:`obspy.core.stream.Stream` object.
458			'''
459			acc_channels = ['ENE', 'ENN', 'ENZ']
460			vel_channels = ['EHE', 'EHN', 'EHZ', 'SHZ']
461			rbm_channels = ['HDF']
462
463			self.stream = self.raw.copy()
464			for trace in self.stream:
465			trace.stats.units = self.units
466			output = 'ACC' if self.deconv == 'GRAV' else self.deconv # if conversion is to gravity
467			if self.deconv:
468			if trace.stats.channel in vel_channels:
469			deconv_vel_inst(self, trace, output) # geophone channels
470
471			elif trace.stats.channel in acc_channels:
472			deconv_acc_inst(self, trace, output) # accelerometer channels
473
474			elif trace.stats.channel in rbm_channels:
475			deconv_rbm_inst(self, trace, output) # this is the Boom channel
476
477			else:
478			trace.stats.units = ' counts' # this is a new one
479
480			else:
481			trace.stats.units = ' counts' # this is not being deconvolved
482
483
484

raspishake / rsudp

Push — master ( fe5590...f40c33 )

build.rsudp.helpers.deconv_acc_inst() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like