from cmath import polar import os import obspy import pickle import datetime import h5py import numpy as np from obspy.clients.fdsn.header import FDSNNoDataException #from obspy.clients.fdsn import Client from obspy import UTCDateTime import time import multiprocessing import multiprocessing import h5py import datetime import numpy as np import matplotlib.pyplot as plt from obspy.signal.filter import bandpass class DataForPnSnTest(): def __init__(self, file_name="models/h5test/all-gzip4.h5", n_length=10240, stride=1, padlen=256, max_dist=2000): self.file_name = file_name self.length = n_length self.stride = stride self.max_dist = max_dist self.padlen = padlen self.n_thread = 2 self.phase_dict = { "Pg":0, "Sg":1, #"P":0, #"S":1, "Pn":2, "Sn":3, } self.ploar1 = { "C":0, "U":0, "R":1, "D":1, } self.ploar2 = { "I":0, "M":1, "E":2, } self.etype_dict = { "eq":0, "ve":1, "ss":2, "sp":3, "ep":4, "ot":5, "se":6 } fqueue = multiprocessing.Queue(100) self.dqueue = multiprocessing.Queue(100) for year in [2020+i for i in range(3)]: multiprocessing.Process(target=self.feed_data, args=(fqueue, year)).start() for _ in range(self.n_thread): multiprocessing.Process(target=self.process, args=(fqueue, self.dqueue)).start() #multiprocessing.Process(target=self.batch_data, args=(dqueue, )).start() def feed_data(self, fqueue, year): file_name = f"ayrdata/csndata/{year}.h5" h5keys = np.load(f"ayrdata/keys/{year}.npy") with open("large/distaz.pkl", "rb") as f: distaz = pickle.load(f) while True: h5file = h5py.File(file_name, "r") print(f"{file_name}数据加载完成") for ekey in h5keys: #print(ekey) event = h5file[ekey] #事件ID etype = event.attrs["type"] #事件类型 if etype in self.etype_dict: typeid = self.etype_dict[etype] #是否为可识别的事件 else: typeid = 7 #不是可识别的6种，则ID设为7 for skey in event: station = event[skey] #获取事件的台站信息 data = [0, 0, 0] #初始化一个长度为3的列表，用于存储台站的数据 #print(skey) for dkey in station: if "HZ" in dkey: idx = 2 elif "HE" in dkey: idx = 0 elif "HN" in dkey: idx = 1 else: print(dkey, "Not exist") continue data[idx] = station[dkey][:] #data.append(station[dkey][:]) btime = datetime.datetime.strptime(station[dkey].attrs['btime'], "%Y/%m/%d %H:%M:%S.%f") #当前数据通道的属性中提取台站数据的起始时间 if len(data)!=3:continue #data[HE数据，HN数据，HZ数据] cnt = 0 for d in data: if type(d)==int: cnt+= 1 if cnt!=0:continue phases = {} phase_count = {"P":0, "S":0, "Pn":0, "Sn":0} dist = -1 if "POLARITY.Pg.UPDOWN" in station.attrs and "POLARITY.Pg.CLARITY" in station.attrs: ptype1 = station.attrs["POLARITY.Pg.UPDOWN"] #获取台站极性 ptype2 = station.attrs["POLARITY.Pg.CLARITY" ] #获取台站极性的清晰度 if ptype1 not in self.ploar1 or ptype2 not in self.ploar2: polars = [] else: polars = [self.ploar1[ptype1], self.ploar2[ptype2]]#获取极性信息和清晰度对应的数值 else: polars = [] ptypes = [i.split(".")[-1] for i in station.attrs["types"].split(",")] #print(ptypes) #if "Pg" not in ptypes :continue #if "Sg" not in ptypes :continue #if "Pn" not in ptypes :continue #if "Sn" not in ptypes :continue for akey in station.attrs: pkey = akey.split(".")[-1] if f"{pkey}.dist" in station.attrs:dist = float(station.attrs[f"{pkey}.dist"]) if pkey not in ptypes:continue pname = pkey.split("+i")[-1].split("-i")[-1].split("+")[-1].split("-")[-1].split("i")[-1].split("2")[0].split("*")[-1] if "Pg" in pname: pname = akey.split(".")[-1] if pname in self.phase_dict: phase_count["P"] += 1 #print(akey, ptypes) phases[pname] = datetime.datetime.strptime(station.attrs[akey], "%Y/%m/%d %H:%M:%S.%f") else: if pname in self.phase_dict: phases[pname] = datetime.datetime.strptime(station.attrs[akey], "%Y/%m/%d %H:%M:%S.%f") if akey == "P" or akey == "Pg": phase_count["P"] += 1 if akey == "S" or akey == "Sg": phase_count["S"] += 1 if akey in ["Pn", "Sn"]: phase_count[akey] += 1 if dist > self.max_dist:continue #if phase_count["Pn"] !=0: # if phase_count["P"] ==0: # continue #if phase_count["Sn"] !=0: # if phase_count["S"] ==0: # continue #print(data) if len(phases)==0:continue fqueue.put([data, btime, phases, polars, typeid, [dist, ekey, skey]]) def process(self, fqueue, dqueue): count = 0 llen = self.length//self.stride while True: data, btime, phases, polars, etype, infos = fqueue.get() pidx = {} plist = [] for pkey in phases: ptime = phases[pkey] delta = (ptime-btime).total_seconds() delta_idx = int(delta * 100) pidx[pkey] = delta_idx plist.append(delta_idx) wz = data[-1] we = data[-2] wn = data[-3] wz_filter = bandpass(wz, 0.5, 20, 100) we_filter = bandpass(we, 0.5, 20, 100) wn_filter = bandpass(wn, 0.5, 20, 100) snrs = [-10000.0, -10000.0, -10000.0, -10000.0] for pkey, pdet in pidx.items(): pi = self.phase_dict[pkey] if pi == 0 or pi==2:# Pg or Pn pre = wz_filter[pdet-50:pdet] aft = wz_filter[pdet:pdet+50] if len(pre)==0 or len(aft)==0: continue snrs[pi] = 10 * np.log10((np.std(aft)+1e-6)/(np.std(pre)+1e-6)) else: pre1 = we_filter[pdet-150:pdet] aft1 = we_filter[pdet:pdet+150] pre2 = wn_filter[pdet-150:pdet] aft2 = wn_filter[pdet:pdet+150] if len(pre1)==0 or len(pre2)==0 or len(aft1)==0 or len(aft2)==0: continue snr1 = 10 * np.log10((np.std(aft1)+1e-6)/(np.std(pre1)+1e-6)) snr2 = 10 * np.log10((np.std(aft2)+1e-6)/(np.std(pre2)+1e-6)) snrs[pi] = snr1 * 0.5 + snr2 * 0.5 cidx = np.random.choice(plist) - np.random.randint(self.padlen, self.length-self.padlen) rdata = [] flen = False for d in data: w = d[cidx:cidx+self.length] wlen = len(w) if wlen!=self.length: flen = True break w = w - np.mean(w) w = w / (np.std(w)+1e-6) w = w / (np.max(np.abs(w))+1e-6) rdata.append(w[np.newaxis, :, np.newaxis]) if flen: continue rdata = np.concatenate(rdata, axis=2) label1 = np.zeros([1, llen, 2]) label2 = np.zeros([1, self.length, 5]) label_polar = np.zeros([1, self.length]) label_quali = np.zeros([1, self.length]) label_weigh = np.zeros([1, self.length]) for pkey in pidx: pid = self.phase_dict[pkey] idx = (pidx[pkey] - cidx)//self.stride if idx-1>0: label1[0, idx-1:idx+2] = -1 if idx > 0 and idx < llen: label1[0, idx, 0] = pid + 1 label1[0, idx, 1] = (pidx[pkey] - cidx)%self.stride phase_intv = {"P":0, "S":0} def norm(t, mu, std=0.1): p = np.exp(-(t-mu)**2/std**2/2) p /= (np.max(p)+1e-6) return p t = np.arange(self.length) * 0.01 phase_time = {0:-1, 1:-1, 2:-1, 3:-1} for pkey in pidx: pid = self.phase_dict[pkey] idx = (pidx[pkey] - cidx) if idx > 0 and idx < self.length: phase_time[pid] = idx dqueue.put([rdata, [phase_time[0], phase_time[1], phase_time[2], phase_time[3]], snrs, infos]) count += 1 def batch_data(self, batch_size=50): x1, x2, x3, x4 = [], [], [], [] for _ in range(batch_size): data, label1, snrs, infos = self.dqueue.get() x1.append(data) x2.append(label1) x3.append(snrs) x4.append(infos) x1 = np.concatenate(x1, axis=0) return x1, x2, x3, x4 class DataTestLargeForIPolar(): def __init__(self, file_name="models/h5test/all-gzip4.h5", n_length=10240, stride=1, padlen=256): self.file_name = file_name self.length = n_length self.stride = stride self.padlen = padlen self.n_thread = 2 self.phase_dict = { "Pg":0, "Sg":1, "P":0, "S":1, "Pn":2, "Sn":3, } self.ploar1 = { "C":0, "U":0, "R":1, "D":1, } self.ploar2 = { #I平缓，M锐利， E既不平缓也不锐利，无标签 "I":0, "M":1, "E":2, } self.etype_dict = { "eq":0, "ve":1, "ss":2, "sp":3, "ep":4, "ot":5, "se":6 } fqueue = multiprocessing.Queue(100) self.dqueue = multiprocessing.Queue(100) for year in [2020+i for i in range(3)]: multiprocessing.Process(target=self.feed_data, args=(fqueue, year)).start() for _ in range(self.n_thread): multiprocessing.Process(target=self.process, args=(fqueue, self.dqueue)).start() #multiprocessing.Process(target=self.batch_data, args=(dqueue, )).start() def feed_data(self, fqueue, year): file_name = f"ayrdata/csndata/{year}.h5" h5keys = np.load(f"ayrdata/keys/{year}.npy") with open("large/distaz.pkl", "rb") as f: distaz = pickle.load(f) while True: h5file = h5py.File(file_name, "r") print(f"{file_name}数据加载完成") for ekey in h5keys: #print(ekey) event = h5file[ekey] etype = event.attrs["type"] if etype in self.etype_dict: typeid = self.etype_dict[etype] else: typeid = 7 for skey in event: station = event[skey] #print(station) data = [0, 0, 0] #print(skey) for dkey in station: if "HZ" in dkey: idx = 2 elif "HE" in dkey: idx = 0 elif "HN" in dkey: idx = 1 else: print(dkey, "Not exist") continue data[idx] = station[dkey][:] #data.append(station[dkey][:]) btime = datetime.datetime.strptime(station[dkey].attrs['btime'], "%Y/%m/%d %H:%M:%S.%f") if len(data)!=3:continue cnt = 0 for d in data: if type(d)==int: cnt+= 1 if cnt!=0:continue phases = {} phase_count = {"P":0, "S":0, "Pn":0, "Sn":0} dist = -1 if "POLARITY.Pg.UPDOWN" in station.attrs and "POLARITY.Pg.CLARITY" not in station.attrs:print("有初动方向，无清晰度标识") if "POLARITY.Pg.UPDOWN" not in station.attrs and "POLARITY.Pg.CLARITY" in station.attrs:print("无初动方向，有清晰度标识") if "POLARITY.Pg.UPDOWN" in station.attrs and "POLARITY.Pg.CLARITY" in station.attrs: ptype1 = station.attrs["POLARITY.Pg.UPDOWN"] ptype2 = station.attrs["POLARITY.Pg.CLARITY"] if ptype2 != "I": continue #print(ptype1,ptype2) if ptype1 not in self.ploar1 or ptype2 not in self.ploar2: polars = [] else: polars = [self.ploar1[ptype1], self.ploar2[ptype2]] else: polars = [] ptypes = [i.split(".")[-1] for i in station.attrs["types"].split(",")] for akey in station.attrs: pkey = akey.split(".")[-1] if pkey not in ptypes:continue pname = pkey.split("+i")[-1].split("-i")[-1].split("+")[-1].split("-")[-1].split("i")[-1].split("2")[0].split("*")[-1] if "Pg" in pname: pname = akey.split(".")[-1] if pname in self.phase_dict: phase_count["P"] += 1 #print(akey, ptypes) phases[pname] = datetime.datetime.strptime(station.attrs[akey], "%Y/%m/%d %H:%M:%S.%f") else: if pname in self.phase_dict: phases[pname] = datetime.datetime.strptime(station.attrs[akey], "%Y/%m/%d %H:%M:%S.%f") if akey == "P" or akey == "Pg": phase_count["P"] += 1 if akey == "S" or akey == "Sg": phase_count["S"] += 1 if akey in ["Pn", "Sn"]: phase_count[akey] += 1 #if dist > 2000:continue #if phase_count["Pn"] !=0: # if phase_count["P"] ==0: # continue #if phase_count["Sn"] !=0: # if phase_count["S"] ==0: # continue #print(data) if len(phases)==0:continue if len(polars)==0:continue #if polars[1] == 2: #print(polars) #print("无标签") #continue #print(polars) #print(ptype1,ptype2) fqueue.put([data, btime, phases, polars, typeid]) def process(self, fqueue, dqueue): count = 0 llen = self.length//self.stride while True: data, btime, phases, polars, etype = fqueue.get() #print(polars) pidx = {} plist = [] for pkey in phases: ptime = phases[pkey] delta = (ptime-btime).total_seconds() delta_idx = int(delta * 100) pidx[pkey] = delta_idx plist.append(delta_idx) #print(pidx) #cidx设置数据的随机起始点 cidx = np.random.choice(plist) - np.random.randint(self.padlen, self.length-self.padlen) rdata = [] flen = False for d in data: w = d[cidx:cidx+self.length] wlen = len(w) if wlen!=self.length: flen = True break #mean去均值 #std归一化 w = w - np.mean(w) w = w / (np.std(w)+1e-6) rdata.append(w[np.newaxis, :, np.newaxis]) if flen: continue rdata = np.concatenate(rdata, axis=2) label1 = np.zeros([1, llen, 2]) label2 = np.zeros([1, self.length, 5]) label_polar = np.zeros([1, self.length]) label_quali = np.zeros([1, self.length]) label_weigh = np.zeros([1, self.length]) #pidx 各个震相在rdata里面的时间点 for pkey in pidx: pid = self.phase_dict[pkey] idx = (pidx[pkey] - cidx)//self.stride if idx-1>0: label1[0, idx-1:idx+2] = -1 if idx > 0 and idx < llen: label1[0, idx, 0] = pid + 1 label1[0, idx, 1] = (pidx[pkey] - cidx)%self.stride phase_intv = {"P":0, "S":0} def norm(t, mu, std=0.1): p = np.exp(-(t-mu)**2/std**2/2) p /= (np.max(p)+1e-6) return p t = np.arange(self.length) * 0.01 #print(pidx) {'Pg': 50000, 'Sg': 51760, 'Sn': 51557} #print(polars) for pkey in pidx: pid = self.phase_dict[pkey] # idx 为rdata的起始事件到该震相的时间长度 idx = (pidx[pkey] - cidx) if idx > 0 and idx < self.length: label2[0, :, pid+1] = norm(t, idx*0.01, 0.1) if pid == 0 and len(polars)>0: #只取pg波的时间 begin = np.clip(idx-50, 0, self.length-60) #print(begin) label_polar[0, begin:begin+100] = polars[0] label_quali[0, begin:begin+100] = polars[1] label_weigh[0, begin:begin+100] = 1 label2[0, :, 0] = np.clip(1-label2[0, :, 1]-label2[0, :, 2]-label2[0, :, 3]-label2[0, :, 4], 0, 1) dqueue.put([rdata, [label_polar, label_quali, label_weigh]]) count += 1 def batch_data(self, batch_size=32): x1, x2, x3 = [], [], [] p1, p2, p3 = [], [], [] x4 = [] for _ in range(batch_size): data,(label_polar, label_quali, label_weigh) = self.dqueue.get() x1.append(data) #x2.append(label1) #x3.append(label2) #x4.append(etype) p1.append(label_polar) p2.append(label_quali) p3.append(label_weigh) x1 = np.concatenate(x1, axis=0) #x2 = np.concatenate(x2, axis=0) #x3 = np.concatenate(x3, axis=0) p1 = np.concatenate(p1, axis=0) p2 = np.concatenate(p2, axis=0) p3 = np.concatenate(p3, axis=0) #x4 = np.array(x4) #print(p1) return x1, p1, p2, p3 class DataTestLargeForAllPolar(): def __init__(self, file_name="models/h5test/all-gzip4.h5", n_length=10240, stride=1, padlen=256): self.file_name = file_name self.length = n_length self.stride = stride self.padlen = padlen self.n_thread = 2 self.phase_dict = { "Pg":0, "Sg":1, #"P":0, #"S":1, "Pn":2, "Sn":3, } self.ploar1 = { "C":0, "U":0, "R":1, "D":1, } self.ploar2 = { #I平缓，M锐利， E既不平缓也不锐利，无标签 "I":0, "M":1, "E":2, } self.etype_dict = { "eq":0, "ve":1, "ss":2, "sp":3, "ep":4, "ot":5, "se":6 } fqueue = multiprocessing.Queue(100) self.dqueue = multiprocessing.Queue(100) for year in [2020+i for i in range(3)]: multiprocessing.Process(target=self.feed_data, args=(fqueue, year)).start() for _ in range(self.n_thread): multiprocessing.Process(target=self.process, args=(fqueue, self.dqueue)).start() #multiprocessing.Process(target=self.batch_data, args=(dqueue, )).start() def feed_data(self, fqueue, year): file_name = f"ayrdata/csndata/{year}.h5" h5keys = np.load(f"ayrdata/keys/{year}.npy") with open("large/distaz.pkl", "rb") as f: distaz = pickle.load(f) while True: h5file = h5py.File(file_name, "r") print(f"{file_name}数据加载完成") for ekey in h5keys: #print(ekey) event = h5file[ekey] etype = event.attrs["type"] if etype in self.etype_dict: typeid = self.etype_dict[etype] else: typeid = 7 for skey in event: station = event[skey] #print(station) data = [0, 0, 0] #print(skey) for dkey in station: if "HZ" in dkey: idx = 2 elif "HE" in dkey: idx = 0 elif "HN" in dkey: idx = 1 else: print(dkey, "Not exist") continue data[idx] = station[dkey][:] #data.append(station[dkey][:]) btime = datetime.datetime.strptime(station[dkey].attrs['btime'], "%Y/%m/%d %H:%M:%S.%f") if len(data)!=3:continue cnt = 0 for d in data: if type(d)==int: cnt+= 1 if cnt!=0:continue phases = {} phase_count = {"P":0, "S":0, "Pn":0, "Sn":0} dist = -1 if "POLARITY.Pg.UPDOWN" in station.attrs and "POLARITY.Pg.CLARITY" in station.attrs: ptype1 = station.attrs["POLARITY.Pg.UPDOWN"] ptype2 = station.attrs["POLARITY.Pg.CLARITY" ] #if ptype2 != "M": continue if ptype1 not in self.ploar1 or ptype2 not in self.ploar2: polars = [] else: polars = [self.ploar1[ptype1], self.ploar2[ptype2]] else: polars = [] ptypes = [i.split(".")[-1] for i in station.attrs["types"].split(",")] for akey in station.attrs: pkey = akey.split(".")[-1] if pkey not in ptypes:continue pname = pkey.split("+i")[-1].split("-i")[-1].split("+")[-1].split("-")[-1].split("i")[-1].split("2")[0].split("*")[-1] if "Pg" in pname: pname = akey.split(".")[-1] if pname in self.phase_dict: phase_count["P"] += 1 #print(akey, ptypes) phases[pname] = datetime.datetime.strptime(station.attrs[akey], "%Y/%m/%d %H:%M:%S.%f") else: if pname in self.phase_dict: phases[pname] = datetime.datetime.strptime(station.attrs[akey], "%Y/%m/%d %H:%M:%S.%f") if akey == "P" or akey == "Pg": phase_count["P"] += 1 if akey == "S" or akey == "Sg": phase_count["S"] += 1 if akey in ["Pn", "Sn"]: phase_count[akey] += 1 #if dist > 2000:continue #if phase_count["Pn"] !=0: # if phase_count["P"] ==0: # continue #if phase_count["Sn"] !=0: # if phase_count["S"] ==0: # continue #print(data) if len(phases)==0:continue if len(polars)==0:continue #if polars[1] == 2: # print("无标签") # continue #print(polars) fqueue.put([data, btime, phases, polars, typeid]) def process(self, fqueue, dqueue): count = 0 llen = self.length//self.stride while True: data, btime, phases, polars, etype = fqueue.get() #print(polars) pidx = {} plist = [] for pkey in phases: ptime = phases[pkey] delta = (ptime-btime).total_seconds() delta_idx = int(delta * 100) pidx[pkey] = delta_idx plist.append(delta_idx) #print(pidx) #cidx设置数据的随机起始点 cidx = np.random.choice(plist) - np.random.randint(self.padlen, self.length-self.padlen) rdata = [] flen = False for d in data: w = d[cidx:cidx+self.length] wlen = len(w) if wlen!=self.length: flen = True break #mean去均值 #std归一化 w = w - np.mean(w) w = w / (np.std(w)+1e-6) rdata.append(w[np.newaxis, :, np.newaxis]) if flen: continue rdata = np.concatenate(rdata, axis=2) label1 = np.zeros([1, llen, 2]) label2 = np.zeros([1, self.length, 5]) label_polar = np.zeros([1, self.length]) label_quali = np.zeros([1, self.length]) label_weigh = np.zeros([1, self.length]) #pidx 各个震相在rdata里面的时间点 for pkey in pidx: pid = self.phase_dict[pkey] idx = (pidx[pkey] - cidx)//self.stride if idx-1>0: label1[0, idx-1:idx+2] = -1 if idx > 0 and idx < llen: label1[0, idx, 0] = pid + 1 label1[0, idx, 1] = (pidx[pkey] - cidx)%self.stride phase_intv = {"P":0, "S":0} def norm(t, mu, std=0.1): p = np.exp(-(t-mu)**2/std**2/2) p /= (np.max(p)+1e-6) return p t = np.arange(self.length) * 0.01 #print(pidx) {'Pg': 50000, 'Sg': 51760, 'Sn': 51557} #print(polars) for pkey in pidx: pid = self.phase_dict[pkey] # idx 为rdata的起始事件到该震相的时间长度 idx = (pidx[pkey] - cidx) if idx > 0 and idx < self.length: label2[0, :, pid+1] = norm(t, idx*0.01, 0.1) if pid == 0 and len(polars)>0: #只取pg波的时间 begin = np.clip(idx-50, 0, self.length-60) #print(begin) label_polar[0, begin:begin+100] = polars[0] label_quali[0, begin:begin+100] = polars[1] label_weigh[0, begin:begin+100] = 1 label2[0, :, 0] = np.clip(1-label2[0, :, 1]-label2[0, :, 2]-label2[0, :, 3]-label2[0, :, 4], 0, 1) dqueue.put([rdata, [label_polar, label_quali, label_weigh]]) count += 1 def batch_data(self, batch_size=32): x1, x2, x3 = [], [], [] p1, p2, p3 = [], [], [] x4 = [] for _ in range(batch_size): data,(label_polar, label_quali, label_weigh) = self.dqueue.get() x1.append(data) #x2.append(label1) #x3.append(label2) #x4.append(etype) p1.append(label_polar) p2.append(label_quali) p3.append(label_weigh) x1 = np.concatenate(x1, axis=0) #x2 = np.concatenate(x2, axis=0) #x3 = np.concatenate(x3, axis=0) p1 = np.concatenate(p1, axis=0) p2 = np.concatenate(p2, axis=0) p3 = np.concatenate(p3, axis=0) #x4 = np.array(x4) #print(p1) return x1, p1, p2, p3 class DataTestLargeForEPolar(): def __init__(self, file_name="models/h5test/all-gzip4.h5", n_length=10240, stride=1, padlen=256): self.file_name = file_name self.length = n_length self.stride = stride self.padlen = padlen self.n_thread = 2 self.phase_dict = { "Pg":0, "Sg":1, "P":0, "S":1, "Pn":2, "Sn":3, } self.ploar1 = { "C":0, "U":0, "R":1, "D":1, } self.ploar2 = { #I平缓，M锐利， E既不平缓也不锐利，无标签 "I":0, "M":1, "E":2, } self.etype_dict = { "eq":0, "ve":1, "ss":2, "sp":3, "ep":4, "ot":5, "se":6 } fqueue = multiprocessing.Queue(100) self.dqueue = multiprocessing.Queue(100) for year in [2020]: multiprocessing.Process(target=self.feed_data, args=(fqueue, year)).start() for _ in range(self.n_thread): multiprocessing.Process(target=self.process, args=(fqueue, self.dqueue)).start() #multiprocessing.Process(target=self.batch_data, args=(dqueue, )).start() def feed_data(self, fqueue, year): file_name = f"ayrdata/csndata/{year}.h5" h5keys = np.load(f"ayrdata/keys/{year}.npy") with open("large/distaz.pkl", "rb") as f: distaz = pickle.load(f) while True: h5file = h5py.File(file_name, "r") print(f"{file_name}数据加载完成") for ekey in h5keys: #print(ekey) event = h5file[ekey] etype = event.attrs["type"] if etype in self.etype_dict: typeid = self.etype_dict[etype] else: typeid = 7 for skey in event: station = event[skey] #print(station) data = [0, 0, 0] #print(skey) for dkey in station: if "HZ" in dkey: idx = 2 elif "HE" in dkey: idx = 0 elif "HN" in dkey: idx = 1 else: print(dkey, "Not exist") continue data[idx] = station[dkey][:] #data.append(station[dkey][:]) btime = datetime.datetime.strptime(station[dkey].attrs['btime'], "%Y/%m/%d %H:%M:%S.%f") if len(data)!=3:continue cnt = 0 for d in data: if type(d)==int: cnt+= 1 if cnt!=0:continue phases = {} phase_count = {"P":0, "S":0, "Pn":0, "Sn":0} dist = -1 if "POLARITY.Pg.UPDOWN" in station.attrs and "POLARITY.Pg.CLARITY" not in station.attrs:print("有初动方向，无清晰度标识") if "POLARITY.Pg.UPDOWN" not in station.attrs and "POLARITY.Pg.CLARITY" in station.attrs:print("无初动方向，有清晰度标识") if "POLARITY.Pg.UPDOWN" in station.attrs and "POLARITY.Pg.CLARITY" in station.attrs: ptype1 = station.attrs["POLARITY.Pg.UPDOWN"] ptype2 = station.attrs["POLARITY.Pg.CLARITY"] if ptype2 != "E": continue #if ptype2 == "E": print(ptype1) if ptype1 not in self.ploar1 or ptype2 not in self.ploar2: polars = [] else: polars = [self.ploar1[ptype1], self.ploar2[ptype2]] else: polars = [] ptypes = [i.split(".")[-1] for i in station.attrs["types"].split(",")] #if station.attrs["POLARITY.Pg.CLARITY" ] != "E":continue for akey in station.attrs: pkey = akey.split(".")[-1] if pkey not in ptypes:continue pname = pkey.split("+i")[-1].split("-i")[-1].split("+")[-1].split("-")[-1].split("i")[-1].split("2")[0].split("*")[-1] if "Pg" in pname: pname = akey.split(".")[-1] if pname in self.phase_dict: phase_count["P"] += 1 #print(akey, ptypes) phases[pname] = datetime.datetime.strptime(station.attrs[akey], "%Y/%m/%d %H:%M:%S.%f") else: if pname in self.phase_dict: phases[pname] = datetime.datetime.strptime(station.attrs[akey], "%Y/%m/%d %H:%M:%S.%f") if akey == "P" or akey == "Pg": phase_count["P"] += 1 if akey == "S" or akey == "Sg": phase_count["S"] += 1 if akey in ["Pn", "Sn"]: phase_count[akey] += 1 #if dist > 2000:continue #if phase_count["Pn"] !=0: # if phase_count["P"] ==0: # continue #if phase_count["Sn"] !=0: # if phase_count["S"] ==0: # continue #print(data) if len(phases)==0:continue if len(polars)==0:continue #if polars[1] == 2: # print(polars) # #print("无标签") # continue #print(polars) fqueue.put([data, btime, phases, polars, typeid]) def process(self, fqueue, dqueue): count = 0 llen = self.length//self.stride while True: data, btime, phases, polars, etype = fqueue.get() print(polars) pidx = {} plist = [] for pkey in phases: ptime = phases[pkey] delta = (ptime-btime).total_seconds() delta_idx = int(delta * 100) pidx[pkey] = delta_idx plist.append(delta_idx) #print(pidx) #cidx设置数据的随机起始点 cidx = np.random.choice(plist) - np.random.randint(self.padlen, self.length-self.padlen) rdata = [] flen = False for d in data: w = d[cidx:cidx+self.length] wlen = len(w) if wlen!=self.length: flen = True break #mean去均值 #std归一化 w = w - np.mean(w) w = w / (np.std(w)+1e-6) rdata.append(w[np.newaxis, :, np.newaxis]) if flen: continue rdata = np.concatenate(rdata, axis=2) label1 = np.zeros([1, llen, 2]) label2 = np.zeros([1, self.length, 5]) label_polar = np.zeros([1, self.length]) label_quali = np.zeros([1, self.length]) label_weigh = np.zeros([1, self.length]) #pidx 各个震相在rdata里面的时间点 for pkey in pidx: pid = self.phase_dict[pkey] idx = (pidx[pkey] - cidx)//self.stride if idx-1>0: label1[0, idx-1:idx+2] = -1 if idx > 0 and idx < llen: label1[0, idx, 0] = pid + 1 label1[0, idx, 1] = (pidx[pkey] - cidx)%self.stride phase_intv = {"P":0, "S":0} def norm(t, mu, std=0.1): p = np.exp(-(t-mu)**2/std**2/2) p /= (np.max(p)+1e-6) return p t = np.arange(self.length) * 0.01 #print(pidx) {'Pg': 50000, 'Sg': 51760, 'Sn': 51557} #print(polars) for pkey in pidx: pid = self.phase_dict[pkey] # idx 为rdata的起始事件到该震相的时间长度 idx = (pidx[pkey] - cidx) if idx > 0 and idx < self.length: label2[0, :, pid+1] = norm(t, idx*0.01, 0.1) if pid == 0 and len(polars)>0: #只取pg波的时间 begin = np.clip(idx-50, 0, self.length-60) #print(begin) label_polar[0, begin:begin+100] = polars[0] label_quali[0, begin:begin+100] = polars[1] label_weigh[0, begin:begin+100] = 1 label2[0, :, 0] = np.clip(1-label2[0, :, 1]-label2[0, :, 2]-label2[0, :, 3]-label2[0, :, 4], 0, 1) dqueue.put([rdata, [label_polar, label_quali, label_weigh]]) count += 1 def batch_data(self, batch_size=32): x1, x2, x3 = [], [], [] p1, p2, p3 = [], [], [] x4 = [] for _ in range(batch_size): data,(label_polar, label_quali, label_weigh) = self.dqueue.get() x1.append(data) #x2.append(label1) #x3.append(label2) #x4.append(etype) p1.append(label_polar) p2.append(label_quali) p3.append(label_weigh) x1 = np.concatenate(x1, axis=0) #x2 = np.concatenate(x2, axis=0) #x3 = np.concatenate(x3, axis=0) p1 = np.concatenate(p1, axis=0) p2 = np.concatenate(p2, axis=0) p3 = np.concatenate(p3, axis=0) #x4 = np.array(x4) #print(p1) return x1, p1, p2, p3 class DataLargeSNRTest(): def __init__(self, file_name="models/h5test/all-gzip4.h5", n_length=10240, stride=1, padlen=256): self.file_name = file_name self.length = n_length self.stride = stride self.padlen = padlen self.n_thread = 2 self.phase_dict = { "Pg":0, "Sg":1, #"P":0, #"S":1, "Pn":2, "Sn":3, } self.ploar1 = { "C":0, "U":0, "R":1, "D":1, } self.ploar2 = { "I":0, "M":1, "E":2, } self.etype_dict = { "eq":0, "ve":1, "ss":2, "sp":3, "ep":4, "ot":5, "se":6 } fqueue = multiprocessing.Queue(100) self.dqueue = multiprocessing.Queue(100) #for year in [2020+i for i in range(3)]: for year in [2020]: multiprocessing.Process(target=self.feed_data, args=(fqueue, year)).start() for _ in range(self.n_thread): multiprocessing.Process(target=self.process, args=(fqueue, self.dqueue)).start() #multiprocessing.Process(target=self.batch_data, args=(dqueue, )).start() def feed_data(self, fqueue, year): file_name = f"ayrdata/csndata/{year}.h5" h5keys = np.load(f"ayrdata/keys/{year}.npy") with open("large/distaz.pkl", "rb") as f: distaz = pickle.load(f) while True: h5file = h5py.File(file_name, "r") print(f"{file_name}数据加载完成") for ekey in h5keys: #print(ekey) event = h5file[ekey] #事件ID etype = event.attrs["type"] #事件类型 if etype in self.etype_dict: typeid = self.etype_dict[etype] #是否为可识别的事件 else: typeid = 7 #不是可识别的6种，则ID设为7 for skey in event: station = event[skey] #获取事件的台站信息 data = [0, 0, 0] #初始化一个长度为3的列表，用于存储台站的数据 #print(skey) for dkey in station: if "HZ" in dkey: idx = 2 elif "HE" in dkey: idx = 0 elif "HN" in dkey: idx = 1 else: print(dkey, "Not exist") continue data[idx] = station[dkey][:] #data.append(station[dkey][:]) btime = datetime.datetime.strptime(station[dkey].attrs['btime'], "%Y/%m/%d %H:%M:%S.%f") #当前数据通道的属性中提取台站数据的起始时间 if len(data)!=3:continue #data[HE数据，HN数据，HZ数据] cnt = 0 for d in data: if type(d)==int: cnt+= 1 if cnt!=0:continue phases = {} phase_count = {"P":0, "S":0, "Pn":0, "Sn":0} dist = -1 if "POLARITY.Pg.UPDOWN" in station.attrs and "POLARITY.Pg.CLARITY" in station.attrs: ptype1 = station.attrs["POLARITY.Pg.UPDOWN"] #获取台站极性 ptype2 = station.attrs["POLARITY.Pg.CLARITY" ] #获取台站极性的清晰度 if ptype1 not in self.ploar1 or ptype2 not in self.ploar2: polars = [] else: polars = [self.ploar1[ptype1], self.ploar2[ptype2]]#获取极性信息和清晰度对应的数值 else: polars = [] ptypes = [i.split(".")[-1] for i in station.attrs["types"].split(",")] #print(ptypes) if "Pg" not in ptypes :continue if "Sg" not in ptypes :continue if "Pn" not in ptypes :continue if "Sn" not in ptypes :continue for akey in station.attrs: pkey = akey.split(".")[-1] if f"{pkey}.dist" in station.attrs:dist = float(station.attrs[f"{pkey}.dist"]) if pkey not in ptypes:continue pname = pkey.split("+i")[-1].split("-i")[-1].split("+")[-1].split("-")[-1].split("i")[-1].split("2")[0].split("*")[-1] if "Pg" in pname: pname = akey.split(".")[-1] if pname in self.phase_dict: phase_count["P"] += 1 #print(akey, ptypes) phases[pname] = datetime.datetime.strptime(station.attrs[akey], "%Y/%m/%d %H:%M:%S.%f") else: if pname in self.phase_dict: phases[pname] = datetime.datetime.strptime(station.attrs[akey], "%Y/%m/%d %H:%M:%S.%f") if akey == "P" or akey == "Pg": phase_count["P"] += 1 if akey == "S" or akey == "Sg": phase_count["S"] += 1 if akey in ["Pn", "Sn"]: phase_count[akey] += 1 if dist > 1000:continue #if phase_count["Pn"] !=0: # if phase_count["P"] ==0: # continue #if phase_count["Sn"] !=0: # if phase_count["S"] ==0: # continue #print(data) if len(phases)==0:continue fqueue.put([data, btime, phases, polars, typeid]) def process(self, fqueue, dqueue): count = 0 llen = self.length//self.stride while True: data, btime, phases, polars, etype = fqueue.get() pidx = {} plist = [] for pkey in phases: ptime = phases[pkey] delta = (ptime-btime).total_seconds() delta_idx = int(delta * 100) pidx[pkey] = delta_idx plist.append(delta_idx) cidx = np.random.choice(plist) - np.random.randint(self.padlen, self.length-self.padlen) rdata = [] flen = False for d in data: w = d[cidx:cidx+self.length] wlen = len(w) if wlen!=self.length: flen = True break w = w - np.mean(w) w = w / (np.std(w)+1e-6) w = w / (np.max(np.abs(w))+1e-6) rdata.append(w[np.newaxis, :, np.newaxis]) if flen: continue rdata = np.concatenate(rdata, axis=2) label1 = np.zeros([1, llen, 2]) label2 = np.zeros([1, self.length, 5]) label_polar = np.zeros([1, self.length]) label_quali = np.zeros([1, self.length]) label_weigh = np.zeros([1, self.length]) for pkey in pidx: pid = self.phase_dict[pkey] idx = (pidx[pkey] - cidx)//self.stride if idx-1>0: label1[0, idx-1:idx+2] = -1 if idx > 0 and idx < llen: label1[0, idx, 0] = pid + 1 label1[0, idx, 1] = (pidx[pkey] - cidx)%self.stride phase_intv = {"P":0, "S":0} def norm(t, mu, std=0.1): p = np.exp(-(t-mu)**2/std**2/2) p /= (np.max(p)+1e-6) return p t = np.arange(self.length) * 0.01 phase_time = {0:-1, 1:-1, 2:-1, 3:-1} for pkey in pidx: pid = self.phase_dict[pkey] idx = (pidx[pkey] - cidx) if idx > 0 and idx < self.length: phase_time[pid] = idx dqueue.put([rdata, [phase_time[0], phase_time[1], phase_time[2], phase_time[3]]]) count += 1 def batch_data(self, batch_size=50): x1, x2 = [], [] for _ in range(batch_size): data, label1 = self.dqueue.get() x1.append(data) x2.append(label1) x1 = np.concatenate(x1, axis=0) return x1, x2 class DataForOutput(): def __init__(self, file_name="models/h5test/all-gzip4.h5", n_length=10240, stride=1, padlen=256, max_dist=2000): self.file_name = file_name self.length = n_length self.stride = stride self.max_dist = max_dist self.padlen = padlen self.n_thread = 2 self.phase_dict = { "Pg":0, "Sg":1, #"P":0, #"S":1, "Pn":2, "Sn":3, } self.ploar1 = { "C":0, "U":0, "R":1, "D":1, } self.ploar2 = { "I":0, "M":1, "E":2, } self.etype_dict = { "eq":0, "ve":1, "ss":2, "sp":3, "ep":4, "ot":5, "se":6 } fqueue = multiprocessing.Queue(100) self.dqueue = multiprocessing.Queue(100) for year in [2009+i for i in range(14)]: multiprocessing.Process(target=self.feed_data, args=(fqueue, year)).start() for _ in range(self.n_thread): multiprocessing.Process(target=self.process, args=(fqueue, self.dqueue)).start() #multiprocessing.Process(target=self.batch_data, args=(dqueue, )).start() def feed_data(self, fqueue, year): file_name = f"ayrdata/csndata/{year}.h5" h5keys = np.load(f"ayrdata/keys/{year}.npy") while True: h5file = h5py.File(file_name, "r") print(f"{file_name}数据加载完成") for ekey in h5keys: #print(ekey) event = h5file[ekey] #事件ID etype = event.attrs["type"] #事件类型 if etype in self.etype_dict: typeid = self.etype_dict[etype] #是否为可识别的事件 else: typeid = 7 #不是可识别的6种，则ID设为7 for skey in event: station = event[skey] #获取事件的台站信息 data = [0, 0, 0] #初始化一个长度为3的列表，用于存储台站的数据 #print(skey) for dkey in station: if "HZ" in dkey: idx = 2 elif "HE" in dkey: idx = 0 elif "HN" in dkey: idx = 1 else: print(dkey, "Not exist") continue data[idx] = station[dkey][:] #data.append(station[dkey][:]) btime = datetime.datetime.strptime(station[dkey].attrs['btime'], "%Y/%m/%d %H:%M:%S.%f") #当前数据通道的属性中提取台站数据的起始时间 if len(data)!=3:continue #data[HE数据，HN数据，HZ数据] cnt = 0 for d in data: if type(d)==int: cnt+= 1 if cnt!=0:continue phases = {} phase_count = {"P":0, "S":0, "Pn":0, "Sn":0} dist = -1 if "POLARITY.Pg.UPDOWN" in station.attrs and "POLARITY.Pg.CLARITY" in station.attrs: ptype1 = station.attrs["POLARITY.Pg.UPDOWN"] #获取台站极性 ptype2 = station.attrs["POLARITY.Pg.CLARITY" ] #获取台站极性的清晰度 if ptype1 not in self.ploar1 or ptype2 not in self.ploar2: polars = [] else: polars = [self.ploar1[ptype1], self.ploar2[ptype2]]#获取极性信息和清晰度对应的数值 else: polars = [] ptypes = [i.split(".")[-1] for i in station.attrs["types"].split(",")] #print(ptypes) if "Pg" not in ptypes :continue if "Sg" not in ptypes :continue if "Pn" not in ptypes :continue if "Sn" not in ptypes :continue for akey in station.attrs: pkey = akey.split(".")[-1] if f"{pkey}.dist" in station.attrs:dist = float(station.attrs[f"{pkey}.dist"]) if pkey not in ptypes:continue pname = pkey.split("+i")[-1].split("-i")[-1].split("+")[-1].split("-")[-1].split("i")[-1].split("2")[0].split("*")[-1] if "Pg" in pname: pname = akey.split(".")[-1] if pname in self.phase_dict: phase_count["P"] += 1 #print(akey, ptypes) phases[pname] = datetime.datetime.strptime(station.attrs[akey], "%Y/%m/%d %H:%M:%S.%f") else: if pname in self.phase_dict: phases[pname] = datetime.datetime.strptime(station.attrs[akey], "%Y/%m/%d %H:%M:%S.%f") if akey == "P" or akey == "Pg": phase_count["P"] += 1 if akey == "S" or akey == "Sg": phase_count["S"] += 1 if akey in ["Pn", "Sn"]: phase_count[akey] += 1 if dist > self.max_dist:continue #if phase_count["Pn"] !=0: # if phase_count["P"] ==0: # continue #if phase_count["Sn"] !=0: # if phase_count["S"] ==0: # continue #print(data) if len(phases)==0:continue fqueue.put([data, btime, phases, polars, typeid, [dist, ekey, skey]]) def process(self, fqueue, dqueue): count = 0 llen = self.length//self.stride while True: data, btime, phases, polars, etype, infos = fqueue.get() pidx = {} plist = [] for pkey in phases: ptime = phases[pkey] delta = (ptime-btime).total_seconds() delta_idx = int(delta * 100) pidx[pkey] = delta_idx plist.append(delta_idx) wz = data[-1] we = data[-2] wn = data[-3] wz_filter = bandpass(wz, 0.5, 20, 100) we_filter = bandpass(we, 0.5, 20, 100) wn_filter = bandpass(wn, 0.5, 20, 100) snrs = [-10000.0, -10000.0, -10000.0, -10000.0] for pkey, pdet in pidx.items(): pi = self.phase_dict[pkey] if pi == 0 or pi==2:# Pg or Pn pre = wz_filter[pdet-50:pdet] aft = wz_filter[pdet:pdet+50] if len(pre)==0 or len(aft)==0: continue snrs[pi] = 10 * np.log10((np.std(aft)+1e-6)/(np.std(pre)+1e-6)) else: pre1 = we_filter[pdet-150:pdet] aft1 = we_filter[pdet:pdet+150] pre2 = wn_filter[pdet-150:pdet] aft2 = wn_filter[pdet:pdet+150] if len(pre1)==0 or len(pre2)==0 or len(aft1)==0 or len(aft2)==0: continue snr1 = 10 * np.log10((np.std(aft1)+1e-6)/(np.std(pre1)+1e-6)) snr2 = 10 * np.log10((np.std(aft2)+1e-6)/(np.std(pre2)+1e-6)) snrs[pi] = snr1 * 0.5 + snr2 * 0.5 cidx = np.random.choice(plist) - np.random.randint(self.padlen, self.length-self.padlen) rdata = [] flen = False for d in data: w = d[cidx:cidx+self.length] wlen = len(w) if wlen!=self.length: flen = True break w = w - np.mean(w) w = w / (np.std(w)+1e-6) w = w / (np.max(np.abs(w))+1e-6) rdata.append(w[np.newaxis, :, np.newaxis]) if flen: continue rdata = np.concatenate(rdata, axis=2) label1 = np.zeros([1, llen, 2]) label2 = np.zeros([1, self.length, 5]) label_polar = np.zeros([1, self.length]) label_quali = np.zeros([1, self.length]) label_weigh = np.zeros([1, self.length]) for pkey in pidx: pid = self.phase_dict[pkey] idx = (pidx[pkey] - cidx)//self.stride if idx-1>0: label1[0, idx-1:idx+2] = -1 if idx > 0 and idx < llen: label1[0, idx, 0] = pid + 1 label1[0, idx, 1] = (pidx[pkey] - cidx)%self.stride phase_intv = {"P":0, "S":0} def norm(t, mu, std=0.1): p = np.exp(-(t-mu)**2/std**2/2) p /= (np.max(p)+1e-6) return p t = np.arange(self.length) * 0.01 phase_time = {0:-1, 1:-1, 2:-1, 3:-1} for pkey in pidx: pid = self.phase_dict[pkey] idx = (pidx[pkey] - cidx) if idx > 0 and idx < self.length: phase_time[pid] = idx dqueue.put([rdata, [phase_time[0], phase_time[1], phase_time[2], phase_time[3]], snrs, infos]) count += 1 def batch_data(self, batch_size=50): x1, x2, x3, x4 = [], [], [], [] for _ in range(batch_size): data, label1, snrs, infos = self.dqueue.get() x1.append(data) x2.append(label1) x3.append(snrs) x4.append(infos) x1 = np.concatenate(x1, axis=0) return x1, x2, x3, x4