Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def sm_generator(scnls=None, amplitudes=None):
""" Function to create station magntiudes for testing."""
counter = 1
sms = []
scnls = scnls or []
params = {
"origin_id": ev.ResourceIdentifier(),
"station_magnitude_type": "M",
"method_id": "mag_calculator",
}
for scnl in scnls:
sm = ev.StationMagnitude(
mag=counter,
mag_errors=ev.QuantityError(uncertainty=counter * 0.1, confidence_level=95),
waveform_id=ev.WaveformStreamID(seed_string=scnl),
creation_info=ev.CreationInfo(
agency_id="dummy_agency", author="dummy", creation_time=UTCDateTime()
),
**params,
)
sms.append(sm)
counter += 1
origin = event.origins[0]
evid = event.resource_id.id.split('/')[-1]
waveform_id = WaveformStreamID()
waveform_id.station_code = station
# network_code is required for QuakeML validation
waveform_id.network_code = ' '
station_string = \
waveform_id.get_seed_string()\
.replace(' ', '-').replace('.', '_').lower()
prefix = '/'.join((res_id_prefix, 'waveformstream',
evid, station_string))
waveform_id.resource_uri = ResourceIdentifier(prefix=prefix)
pick = Pick()
prefix = '/'.join((res_id_prefix, 'pick', evid, station_string))
pick.resource_id = ResourceIdentifier(prefix=prefix)
date = origin.time.strftime('%Y%m%d')
pick.time = UTCDateTime(date + arrival_time)
# Check if pick is on the next day:
if pick.time < origin.time:
pick.time += timedelta(days=1)
pick.waveform_id = waveform_id
pick.backazimuth = backazimuth
onset = phase[0]
if onset == 'e':
pick.onset = 'emergent'
phase = phase[1:]
elif onset == 'i':
pick.onset = 'impulsive'
phase = phase[1:]
elif onset == 'q':
pick.onset = 'questionable'
:param parent_res_id:
:class:`~obspy.core.event.resourceid.ResourceIdentifier`
:rtype: :class:`~obspy.core.event.resourceid.ResourceIdentifier`
:return:Â ResourceIdentifier object.
"""
prefix = self.res_id_prefix
# Put the parent id as prefix
# Example: smi:local/origin/351412/arrival/6389611
# | prefix | ident |
if parent:
prefix = parent.resource_id.id
elif parent_res_id:
prefix = parent_res_id.id
public_id = "%s/%s" % (prefix, ident)
return ResourceIdentifier(public_id)
def plot_data(event, station, show=True, legend=False):
val = data[obspy.core.event.ResourceIdentifier(event)]
obs = val["processed_observed_stream"].select(station=station)[0]
syn = val["processed_synthetic_stream"].select(station=station)[0]
plt.plot(obs.times(), obs.data * 1E6, color="0.3", linestyle="--", linewidth=5,
label="Observed Data")
plt.plot(syn.times(), syn.data * 1E6, color="0.1", linestyle=":", label="Synthetic Data")
ts= original_cs[(original_cs.event == validate) &
(original_cs.station == "TIN")].time_shift
plt.plot(syn.times() - float(ts), syn.data * 1E6, color="k", label="Shifted Synthetic Data")
plt.ylabel("Velocity [$\mu$m/s]")
plt.xlabel("Relative Time [s]")
plt.xlim(20, 65)
plt.xticks([30, 40, 50, 60], ["30", "40", "50", "60"])
def _set_resource_id(self, value):
if type(value) == dict:
value = ResourceIdentifier(**value)
elif type(value) != ResourceIdentifier:
value = ResourceIdentifier(value)
self.__dict__['resource_id'] = value
if z_amplitude is not None:
amplitude = Amplitude()
prefix = '/'.join((res_id_prefix, 'amp', evid, station_string))
amplitude.resource_id = ResourceIdentifier(prefix=prefix)
amplitude.generic_amplitude = z_amplitude * 1E-6
amplitude.unit = 'm'
amplitude.period = z_period
amplitude.type = 'AS'
amplitude.magnitude_hint = 'Ms'
amplitude.pick_id = pick.resource_id
event.amplitudes.append(amplitude)
if msz_mag is not None:
station_magnitude = StationMagnitude()
prefix = '/'.join((res_id_prefix, 'stationmagntiude',
evid, station_string))
station_magnitude.resource_id = ResourceIdentifier(prefix=prefix)
station_magnitude.origin_id = event.origins[0].resource_id
station_magnitude.mag = ms_mag
station_magnitude.station_magnitude_type = 'Ms'
if amplitude is not None:
station_magnitude.amplitude_id = amplitude.resource_id
event.station_magnitudes.append(station_magnitude)
UserWarning("Found focal-mechanism info: reading amplitude-ratio fit,"
"number of bad polarities and number of bad amplitude ratios"
"is not implemented.")
for line, line_num in tagged_lines['F']:
nodal_p = NodalPlane(strike=float(line[0:10]), dip=float(line[10:20]),
rake=float(line[20:30]))
try:
# Apparently these don't have to be filled.
nodal_p.strike_errors = QuantityError(float(line[30:35]))
nodal_p.dip_errors = QuantityError(float(line[35:40]))
nodal_p.rake_errors = QuantityError(float(line[40:45]))
except ValueError:
pass
fm = FocalMechanism(nodal_planes=NodalPlanes(nodal_plane_1=nodal_p))
try:
fm.method_id = ResourceIdentifier(
"smi:nc.anss.org/focalMehcanism/" + line[70:77].strip())
fm.creation_info = CreationInfo(agency_id=line[66:69].strip)
fm.misfit = float(line[45:50])
fm.station_distribution_ratio = float(line[50:55])
except ValueError:
pass
event.focal_mechanisms.append(fm)
return event
creation_info=CreationInfo(agency_id=mt_line_1[45:48].strip())))
event.magnitudes.append(Magnitude(
mag=float(mt_line_1[55:59]),
magnitude_type=_nortoevmag(mt_line_1[59]),
creation_info=CreationInfo(agency_id=mt_line_1[60:63].strip()),
origin_id=event.origins[-1].resource_id))
event.focal_mechanisms.append(FocalMechanism(
moment_tensor=MomentTensor(
derived_origin_id=event.origins[-1].resource_id,
moment_magnitude_id=event.magnitudes[-1].resource_id,
scalar_moment=float(mt_line_2[52:62]), tensor=Tensor(
m_rr=float(mt_line_2[3:9]), m_tt=float(mt_line_2[10:16]),
m_pp=float(mt_line_2[17:23]), m_rt=float(mt_line_2[24:30]),
m_rp=float(mt_line_2[31:37]),
m_tp=float(mt_line_2[38:44])),
method_id=ResourceIdentifier(
"smi:nc.anss.org/momentTensor/" + mt_line_1[70:77].strip()
))))
return event
depth_usage_flag = phase[7]
except IndexError:
# usage flag is not defined
depth_usage_flag = None
# FIXME: I'm not sure that 'X' actually
# means 'used'
if depth_usage_flag == 'X':
# FIXME: is this enough to say that
# the event is constrained by depth phases?
origin.depth_type = 'constrained by depth phases'
origin.quality.depth_phase_count += 1
else:
pick = Pick()
prefix = '/'.join((res_id_prefix, 'pick',
evid, station_string))
pick.resource_id = ResourceIdentifier(prefix=prefix)
date = origin.time.strftime('%Y%m%d')
pick.time = UTCDateTime(date + arrival_time)
# Check if pick is on the next day:
if pick.time < origin.time:
pick.time += timedelta(days=1)
pick.waveform_id = p_pick.waveform_id
pick.backazimuth = p_pick.backazimuth
onset = phase[0]
if onset == 'e':
pick.onset = 'emergent'
phase = phase[1:]
elif onset == 'i':
pick.onset = 'impulsive'
phase = phase[1:]
elif onset == 'q':
pick.onset = 'questionable'
station_magnitude.waveform_id = pick.waveform_id
res_id = '/'.join(
(res_id_prefix, 'magnitude/generic/body_wave_magnitude'))
station_magnitude.method_id = \
ResourceIdentifier(id=res_id)
event.station_magnitudes.append(station_magnitude)
arrival = Arrival()
prefix = '/'.join((res_id_prefix, 'arrival', evid, station_string))
arrival.resource_id = ResourceIdentifier(prefix=prefix)
arrival.pick_id = pick.resource_id
arrival.phase = pick.phase_hint
arrival.azimuth = azimuth
arrival.distance = distance
arrival.time_residual = residual
res_id = '/'.join((res_id_prefix, 'earthmodel/ak135'))
arrival.earth_model_id = ResourceIdentifier(id=res_id)
origin.arrivals.append(arrival)
origin.quality.minimum_distance = min(
d for d in (arrival.distance, origin.quality.minimum_distance)
if d is not None)
origin.quality.maximum_distance = \
max(arrival.distance, origin.quality.minimum_distance)
origin.quality.associated_phase_count += 1
return pick, arrival