build.rsudp.helpers.dump_default() - Code Metrics - Inspection of "Update and rename CONTRIBUTORS.md to CONTRIBUTORS...." - raspishake/rsudp - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( e96c3d...44474a )

by Ian

created 2020-05-26 20:36 UTC

build.rsudp.helpers.dump_default() A

↳ Parent: build.rsudp.helpers

Complexity

Conditions

Size

Total Lines	11
Code Lines	5

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	5
dl	0
loc	11
rs	10
c	0
b	0
f	0
cc	2
nop	2

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": true,
    "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"},
"rsam": {
    "enabled": false,
    "quiet": true,
    "fwaddr": false,
    "fwport": false,
    "fwformat": "LITE",
    "channel": "HZ",
    "interval": 10,
    "deconvolve": false,
    "units": "VEL"}
}

""" % (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)

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

raspishake / rsudp

Push — master ( e96c3d...44474a )

build.rsudp.helpers.dump_default() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like