125 lines
4.0 KiB
Python
125 lines
4.0 KiB
Python
#!/usr/bin/env python
|
|
# -*- coding: utf-8 -*-
|
|
|
|
import matplotlib
|
|
|
|
|
|
def get_bg_color(check_key, status, dt_thresh=None, hex=False):
|
|
message = status.message
|
|
if check_key == 'last active':
|
|
bg_color = get_time_delay_color(message, dt_thresh)
|
|
elif check_key == 'temp':
|
|
bg_color = get_temp_color(message)
|
|
else:
|
|
if status.is_warn:
|
|
bg_color = get_color('WARNX')(status.count)
|
|
elif status.is_error:
|
|
bg_color = get_color('FAIL')
|
|
else:
|
|
bg_color = get_color(message)
|
|
if not bg_color:
|
|
bg_color = get_color('undefined')
|
|
|
|
if hex:
|
|
bg_color = '#{:02x}{:02x}{:02x}'.format(*bg_color[:3])
|
|
return bg_color
|
|
|
|
|
|
def get_color(key):
|
|
# some GUI default colors
|
|
colors_dict = {'FAIL': (255, 50, 0, 255),
|
|
'NO DATA': (255, 255, 125, 255),
|
|
'WARN': (255, 255, 80, 255),
|
|
'WARNX': lambda x: (min([255, 200 + x ** 2]), 255, 80, 255),
|
|
'OK': (125, 255, 125, 255),
|
|
'undefined': (230, 230, 230, 255)}
|
|
return colors_dict.get(key)
|
|
|
|
|
|
def get_time_delay_color(dt, dt_thresh):
|
|
""" Set color of time delay after thresholds specified in self.dt_thresh """
|
|
if dt < dt_thresh[0]:
|
|
return get_color('OK')
|
|
elif dt_thresh[0] <= dt < dt_thresh[1]:
|
|
return get_color('WARN')
|
|
return get_color('FAIL')
|
|
|
|
|
|
def get_temp_color(temp, vmin=-10, vmax=60, cmap='coolwarm'):
|
|
""" Get an rgba temperature value back from specified cmap, linearly interpolated between vmin and vmax. """
|
|
if type(temp) in [str]:
|
|
return get_color('undefined')
|
|
cmap = matplotlib.cm.get_cmap(cmap)
|
|
val = (temp - vmin) / (vmax - vmin)
|
|
rgba = [int(255 * c) for c in cmap(val)]
|
|
return rgba
|
|
|
|
|
|
def modify_stream_for_plot(st, parameters):
|
|
""" copy (if necessary) and modify stream for plotting """
|
|
ch_units = parameters.get('CHANNEL_UNITS')
|
|
ch_transf = parameters.get('CHANNEL_TRANSFORM')
|
|
|
|
# if either of both are defined make copy
|
|
if ch_units or ch_transf:
|
|
st = st.copy()
|
|
|
|
# modify trace for plotting by multiplying unit factor (e.g. 1e-3 mV to V)
|
|
if ch_units:
|
|
for tr in st:
|
|
channel = tr.stats.channel
|
|
unit_factor = ch_units.get(channel)
|
|
if unit_factor:
|
|
tr.data = tr.data * float(unit_factor)
|
|
# modify trace for plotting by other arithmetic expressions
|
|
if ch_transf:
|
|
for tr in st:
|
|
channel = tr.stats.channel
|
|
transf = ch_transf.get(channel)
|
|
if transf:
|
|
tr.data = transform_trace(tr.data, transf)
|
|
|
|
return st
|
|
|
|
|
|
def transform_trace(data, transf):
|
|
"""
|
|
Transform trace with arithmetic operations in order, specified in transf
|
|
@param data: numpy array
|
|
@param transf: list of lists with arithmetic operations (e.g. [['*', '20'], ] -> multiply data by 20
|
|
"""
|
|
# This looks a little bit hardcoded, however it is safer than using e.g. "eval"
|
|
for operator_str, val in transf:
|
|
if operator_str == '+':
|
|
data = data + val
|
|
elif operator_str == '-':
|
|
data = data - val
|
|
elif operator_str == '*':
|
|
data = data * val
|
|
elif operator_str == '/':
|
|
data = data / val
|
|
else:
|
|
raise IOError(f'Unknown arithmethic operator string: {operator_str}')
|
|
|
|
return data
|
|
|
|
|
|
def annotate_trace_axes(fig, parameters, verbosity=0):
|
|
"""
|
|
Adds channel names to y-axis if defined in parameters.
|
|
Can get mixed up if channel order in stream and channel names defined in parameters.yaml differ, but it is
|
|
difficult to assess the correct order from Obspy plotting routing.
|
|
"""
|
|
names = parameters.get('channel_names')
|
|
if not names: # or not len(st.traces):
|
|
return
|
|
if not len(names) == len(fig.axes):
|
|
if verbosity:
|
|
print('Mismatch in axis and label lengths. Not adding plot labels')
|
|
return
|
|
for channel_name, ax in zip(names, fig.axes):
|
|
if channel_name:
|
|
ax.set_ylabel(channel_name)
|
|
|
|
|