import os import obspy import pickle import datetime import h5py import numpy as np import time import multiprocessing import tqdm from obspy.geodetics.base import gps2dist_azimuth, kilometers2degrees class DataDist2(): def __init__(self, file_name="h5data", n_length=256, stride=16, padlen=64, mindist=0, maxdist=1000, istest=False, angle_only=False, filter_by_snr=False): self.file_name = file_name self.length = n_length self.stride = stride self.padlen = padlen self.n_thread = 1 self.maxdist = maxdist self.mindist = mindist self.angle_only = angle_only self.filter_by_snr = filter_by_snr self.phase_dict = { "Pg":0, "P":1, "Pn":2, #"PcP":3, #"PgP":4, #"PgPn":5, #"PgPcP":6, #"PnP":7, #"PnPcP":8, #"PcPP":9, #"PcPPn":10, #"PcPPcP":11, #"PcPnP":12 } self.ploar1 = { "C":0, "U":0, "R":1, "D":1, } self.ploar2 = { "I":0, "M":1, "E":2, } fqueue = multiprocessing.Queue(100) self.dqueue = multiprocessing.Queue(100) self.istest = istest if istest: for i in range(4): p1 = multiprocessing.Process(target=self.feed_data, args=(fqueue, i+2019),daemon=True) p1.start() else: for i in range(10): p1 = multiprocessing.Process(target=self.feed_data, args=(fqueue, i+2009),daemon=True) p1.start() for _ in range(self.n_thread): c1 = multiprocessing.Process(target=self.process, args=(fqueue, self.dqueue),daemon=True) c1.start() #multiprocessing.Process(target=self.batch_data, args=(dqueue, )).start() def feed_data(self, fqueue, year): f = open("ayrdata/china.loc", "r", encoding="utf8") sloc = {} for line in f.readlines(): sline = [i for i in line.split(" ") if len(i)>0] skey = ".".join(sline[:3]) loc = [float(sline[3]), float(sline[4])] sloc[skey] = loc while True: h5file = h5py.File(f"ayrdata/csndata/{year}.h5", "r") h5key = np.load(f"ayrdata/keys/{year}.npy") for ekey in h5key: if self.maxdist>=1000000: if "CB." not in ekey: if np.random.random()<0.95: continue event = h5file[ekey] #if "FJ." not in ekey:continue #for key in event.attrs: # print("ekey", key) if "depth" not in event.attrs: depth = 0.0 #continue mag = event.attrs["mag"] elon = event.attrs["lon"] elat = event.attrs["lat"] edep = event.attrs["depth"] #if self.istest: # if mag<0.0:continue keys = [key for key in event] if len(keys)<1:continue sidx = np.random.randint(len(keys)) for skey in keys[sidx:sidx+1]: station = event[skey] if skey not in sloc:continue slon, slat = sloc[skey] try: dist, abz, baz = gps2dist_azimuth(elat, elon, slat, slon) dist = dist/1000 except: continue if dist>self.maxdist or dist < self.mindist: continue #for key in station.attrs: # print(skey, key, dist, abz) #if "POLARITY.Pg" not in station.attrs:continue #if "POLARITY.Pg.UPDOWN" not in station.attrs:continue #if "POLARITY.Pg.CLARITY" not in station.attrs:continue #ptype1 = station.attrs["POLARITY.Pg.UPDOWN"] #ptype2 = station.attrs["POLARITY.Pg.CLARITY" ] #if ptype1 not in self.ploar1 or ptype2 not in self.ploar2:continue data = [0, 0, 0] for dkey in ["BHE", "BHN", "BHZ"]: if dkey not in station: dkey = dkey.replace("B", "S") if dkey not in station: continue if "E" in dkey: data[0] = station[dkey][:] elif "N" in dkey: data[1] = station[dkey][:] elif "Z" in dkey: data[2] = station[dkey][:] #data.append(station[dkey][:]) btime = datetime.datetime.strptime(station[dkey].attrs['btime'], "%Y/%m/%d %H:%M:%S.%f") if type(data[0])==int or type(data[1])==int or type(data[2])==int:continue #if len(data)!=3:continue phases = {} dists= -1 for akey in station.attrs: if "dist" in akey: dists = float(station.attrs[akey]) if "Pg" in akey: pname = akey.split(".")[-1] if pname in self.phase_dict: phases[pname] = datetime.datetime.strptime(station.attrs[akey], "%Y/%m/%d %H:%M:%S.%f") else: if akey in self.phase_dict: phases[akey] = datetime.datetime.strptime(station.attrs[akey], "%Y/%m/%d %H:%M:%S.%f") if len(phases)==0:continue if dist < 0: print("Dist skip", dist) continue fqueue.put([data, btime, phases, dist, mag, baz, edep, mag]) def process(self, fqueue, dqueue): count = 0 llen = self.length//self.stride while True: data, btime, phases, dist, mags, abz, edep, emag = fqueue.get() if dist>self.maxdist or dist < self.mindist: continue 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) pidx = np.random.randint(self.padlen, self.padlen*2) cidx = np.random.choice(plist) - pidx 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.max(w)+1e-6) rdata.append(w[np.newaxis, :, np.newaxis]) if flen: continue rdata = np.concatenate(rdata, axis=2) pre = rdata[0, pidx-100:pidx, :] aft = rdata[0, pidx:pidx+100, :] snr = np.std(aft) / (np.std(pre) + 1e-6) if self.filter_by_snr: if snr < 5.0: continue #if self.istest: # if snr < 5.0: # continue #print("数据+1") rdata /= (np.max(np.abs(rdata)) + 1e-6) #rdata *= np.random.uniform(0.5, 1.5) abzr = abz/180.*np.pi if self.maxdist>=1000000: #print("pre", dist) dist = kilometers2degrees(dist) #print("aft", dist) if self.angle_only: dist = 1.0 sinx = np.sin(abzr) * dist cosx = np.cos(abzr) * dist #print(dist) label = np.array([[cosx, sinx, edep, emag]]) dqueue.put([rdata, label, [pidx]]) count += 1 def batch_data(self, batch_size=32): x1, x2, x3 = [], [], [] for _ in range(batch_size): data, label, pidx = self.dqueue.get() #print(data.shape, label.shape) x1.append(data) x2.append(label) x3.append(pidx) x1 = np.concatenate(x1, axis=0) x2 = np.concatenate(x2, axis=0) return x1, x2, x3 class Data(): def __init__(self, file_name="h5data", n_length=256, stride=16, padlen=64, mindist=0, maxdist=1000, istest=False, angle_only=False, filter_by_snr=False): self.file_name = file_name self.length = n_length self.stride = stride self.padlen = padlen self.n_thread = 1 self.maxdist = maxdist self.mindist = mindist self.angle_only = angle_only self.filter_by_snr = filter_by_snr self.phase_dict = { "Pg":0, "Sg":1, #"P":0, #"S":1, "Pn":2, "Sn":3, #"PcP":3, #"PgP":4, #"PgPn":5, #"PgPcP":6, #"PnP":7, #"PnPcP":8, #"PcPP":9, #"PcPPn":10, #"PcPPcP":11, #"PcPnP":12 } self.ploar1 = { "C":0, "U":0, "R":1, "D":1, } self.ploar2 = { "I":0, "M":1, "E":2, } fqueue = multiprocessing.Queue(100) self.dqueue = multiprocessing.Queue(100) self.istest = istest if istest: for i in range(4): p1 = multiprocessing.Process(target=self.feed_data, args=(fqueue, i+2019),daemon=True) p1.start() else: for i in range(10): p1 = multiprocessing.Process(target=self.feed_data, args=(fqueue, i+2009),daemon=True) p1.start() for _ in range(self.n_thread): c1 = multiprocessing.Process(target=self.process, args=(fqueue, self.dqueue),daemon=True) c1.start() #multiprocessing.Process(target=self.batch_data, args=(dqueue, )).start() def feed_data(self, fqueue, year): f = open("ayrdata/china.loc", "r", encoding="utf8") sloc = {} for line in f.readlines(): sline = [i for i in line.split(" ") if len(i)>0] skey = ".".join(sline[:3]) loc = [float(sline[3]), float(sline[4])] sloc[skey] = loc while True: h5file = h5py.File(f"ayrdata/csndata/{year}.h5", "r") h5key = np.load(f"ayrdata/keys/{year}.npy") for ekey in h5key: if self.maxdist>=1000000: if "CB." not in ekey: if np.random.random()<0.5: continue event = h5file[ekey] #if "FJ." not in ekey:continue #for key in event.attrs: # print("ekey", key) if "depth" not in event.attrs: depth = 0.0 #continue mag = event.attrs["mag"] elon = event.attrs["lon"] elat = event.attrs["lat"] edep = event.attrs["depth"] #if self.istest: # if mag<0.0:continue keys = [key for key in event] if len(keys)<1:continue sidx = np.random.randint(len(keys)) for skey in keys[sidx:sidx+1]: station = event[skey] if skey not in sloc:continue slon, slat = sloc[skey] try: dist, abz, baz = gps2dist_azimuth(elat, elon, slat, slon) dist = dist/1000 except: continue if dist>self.maxdist or dist < self.mindist: continue #for key in station.attrs: # print(skey, key, dist, abz) #if "POLARITY.Pg" not in station.attrs:continue #if "POLARITY.Pg.UPDOWN" not in station.attrs:continue #if "POLARITY.Pg.CLARITY" not in station.attrs:continue #ptype1 = station.attrs["POLARITY.Pg.UPDOWN"] #ptype2 = station.attrs["POLARITY.Pg.CLARITY" ] #if ptype1 not in self.ploar1 or ptype2 not in self.ploar2:continue data = [0, 0, 0] for dkey in ["BHE", "BHN", "BHZ"]: if dkey not in station: dkey = dkey.replace("B", "S") if dkey not in station: continue if "E" in dkey: data[0] = station[dkey][:] elif "N" in dkey: data[1] = station[dkey][:] elif "Z" in dkey: data[2] = station[dkey][:] #data.append(station[dkey][:]) btime = datetime.datetime.strptime(station[dkey].attrs['btime'], "%Y/%m/%d %H:%M:%S.%f") if type(data[0])==int or type(data[1])==int or type(data[2])==int:continue #if len(data)!=3:continue phases = {} dists= -1 for akey in station.attrs: if "dist" in akey: dists = float(station.attrs[akey]) if "Pg" in akey: pname = akey.split(".")[-1] if pname in self.phase_dict: phases[pname] = datetime.datetime.strptime(station.attrs[akey], "%Y/%m/%d %H:%M:%S.%f") else: if akey in self.phase_dict: phases[akey] = datetime.datetime.strptime(station.attrs[akey], "%Y/%m/%d %H:%M:%S.%f") if len(phases)==0:continue #if dist < 0: # print("Dist skip", dist) # continue fqueue.put([data, btime, phases, dist, mag, baz, edep, mag]) def process(self, fqueue, dqueue): count = 0 llen = self.length//self.stride while True: data, btime, phases, dist, mags, abz, edep, emag = fqueue.get() if dist>self.maxdist or dist < self.mindist: continue 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) dpidx = np.random.randint(self.padlen, self.length-self.padlen) cidx = np.random.choice(plist) - dpidx bbidx = np.clip(np.min(plist)-cidx, 0, self.length) 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.max(w)+1e-6) rdata.append(w[np.newaxis, :, np.newaxis]) if flen: continue rdata = np.concatenate(rdata, axis=2) pre = rdata[0, dpidx-100:dpidx, :] aft = rdata[0, dpidx:dpidx+100, :] snr = np.std(aft) / (np.std(pre) + 1e-6) if self.filter_by_snr: if snr < 5.0: continue rdata /= (np.max(np.abs(rdata)) + 1e-6) #rdata *= np.random.uniform(0.5, 1.5) abzr = abz/180.*np.pi dist = kilometers2degrees(dist) #print("aft", dist) if self.angle_only: dist = 1.0 sinx = np.sin(abzr) * dist cosx = np.cos(abzr) * dist #print(dist) phase_label = np.zeros((1, self.length, 5)) 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 for pkey in pidx: if pkey not in self.phase_dict:continue pid = self.phase_dict[pkey] idx = (pidx[pkey] - cidx) if idx > 0 and idx < self.length: phase_label[0, :, pid+1] = norm(t, idx*0.01, 0.2) label = np.array([[cosx, sinx, edep, emag, snr]]) phase_label[:, :, 0] = np.clip(1-np.sum(phase_label[:, :, 1:], axis=2), 0, 1) dqueue.put([rdata, label, [bbidx], phase_label]) count += 1 def batch_data(self, batch_size=32): x1, x2, x3 = [], [], [] x4 = [] for _ in range(batch_size): data, label, pidx, pha = self.dqueue.get() #print(data.shape, label.shape) x1.append(data) x2.append(label) x3.append(pidx) x4.append(pha) x1 = np.concatenate(x1, axis=0) x2 = np.concatenate(x2, axis=0) x4 = np.concatenate(x4, axis=0) return x1, x2, x3, x4 import threading import queue import multiprocessing class DataThread(): def __init__(self, file_name="h5data", n_length=256, stride=16, padlen=64, mindist=0, maxdist=1000, istest=False, angle_only=False, filter_by_snr=False): self.file_name = file_name self.length = n_length self.stride = stride self.padlen = padlen self.n_thread = 1 self.maxdist = maxdist self.mindist = mindist self.angle_only = angle_only self.filter_by_snr = filter_by_snr self.phase_dict = { "Pg":0, "Sg":1, #"P":0, #"S":1, "Pn":2, "Sn":3, #"PcP":3, #"PgP":4, #"PgPn":5, #"PgPcP":6, #"PnP":7, #"PnPcP":8, #"PcPP":9, #"PcPPn":10, #"PcPPcP":11, #"PcPnP":12 } self.ploar1 = { "C":0, "U":0, "R":1, "D":1, } self.ploar2 = { "I":0, "M":1, "E":2, } fqueue = queue.Queue(100) self.dqueue = queue.Queue(100) self.istest = istest if istest: for i in range(4): p1 = threading.Thread(target=self.feed_data, args=(fqueue, i+2019),daemon=True) p1.start() else: for i in range(10): p1 = threading.Thread(target=self.feed_data, args=(fqueue, i+2009),daemon=True) p1.start() for _ in range(self.n_thread): c1 = threading.Thread(target=self.process, args=(fqueue, self.dqueue),daemon=True) c1.start() #multiprocessing.Process(target=self.batch_data, args=(dqueue, )).start() def feed_data(self, fqueue, year): f = open("ayrdata/china.loc", "r", encoding="utf8") sloc = {} for line in f.readlines(): sline = [i for i in line.split(" ") if len(i)>0] skey = ".".join(sline[:3]) loc = [float(sline[3]), float(sline[4])] sloc[skey] = loc while True: h5file = h5py.File(f"ayrdata/csndata/{year}.h5", "r") h5key = np.load(f"ayrdata/keys/{year}.npy") np.random.shuffle(h5key) for ekey in h5key: if self.maxdist>=1000000: if "CB." not in ekey: if np.random.random()<0.5: continue event = h5file[ekey] #if "FJ." not in ekey:continue #for key in event.attrs: # print("ekey", key) if "depth" not in event.attrs: depth = 0.0 #continue mag = event.attrs["mag"] elon = event.attrs["lon"] elat = event.attrs["lat"] edep = event.attrs["depth"] #if self.istest: # if mag<0.0:continue keys = [key for key in event] if len(keys)<1:continue sidx = np.random.randint(len(keys)) for skey in keys[sidx:sidx+1]: station = event[skey] if skey not in sloc:continue slon, slat = sloc[skey] try: dist, abz, baz = gps2dist_azimuth(elat, elon, slat, slon) dist = dist/1000 except: continue if dist>self.maxdist or dist < self.mindist: continue #for key in station.attrs: # print(skey, key, dist, abz) #if "POLARITY.Pg" not in station.attrs:continue #if "POLARITY.Pg.UPDOWN" not in station.attrs:continue #if "POLARITY.Pg.CLARITY" not in station.attrs:continue #ptype1 = station.attrs["POLARITY.Pg.UPDOWN"] #ptype2 = station.attrs["POLARITY.Pg.CLARITY" ] #if ptype1 not in self.ploar1 or ptype2 not in self.ploar2:continue data = [0, 0, 0] for dkey in ["BHE", "BHN", "BHZ"]: if dkey not in station: dkey = dkey.replace("B", "S") if dkey not in station: continue if "E" in dkey: data[0] = station[dkey][:] elif "N" in dkey: data[1] = station[dkey][:] elif "Z" in dkey: data[2] = station[dkey][:] #data.append(station[dkey][:]) btime = datetime.datetime.strptime(station[dkey].attrs['btime'], "%Y/%m/%d %H:%M:%S.%f") if type(data[0])==int or type(data[1])==int or type(data[2])==int:continue #if len(data)!=3:continue phases = {} dists= -1 for akey in station.attrs: if "dist" in akey: dists = float(station.attrs[akey]) if "Pg" in akey: pname = akey.split(".")[-1] if pname in self.phase_dict: phases[pname] = datetime.datetime.strptime(station.attrs[akey], "%Y/%m/%d %H:%M:%S.%f") else: if akey in self.phase_dict: phases[akey] = datetime.datetime.strptime(station.attrs[akey], "%Y/%m/%d %H:%M:%S.%f") if len(phases)==0:continue #if dist < 0: # print("Dist skip", dist) # continue fqueue.put([data, btime, phases, dist, mag, baz, edep, mag]) def process(self, fqueue, dqueue): count = 0 llen = self.length//self.stride while True: data, btime, phases, dist, mags, abz, edep, emag = fqueue.get() if dist>self.maxdist or dist < self.mindist: continue 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) dpidx = np.random.randint(self.padlen, self.length-self.padlen) cidx = np.random.choice(plist) - dpidx bbidx = np.clip(np.min(plist)-cidx, 0, self.length) 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.max(w)+1e-6) rdata.append(w[np.newaxis, :, np.newaxis]) if flen: continue rdata = np.concatenate(rdata, axis=2) pre = rdata[0, dpidx-100:dpidx, :] aft = rdata[0, dpidx:dpidx+100, :] snr = np.std(aft) / (np.std(pre) + 1e-6) if self.filter_by_snr: if snr < 5.0: continue rdata /= (np.max(np.abs(rdata)) + 1e-6) #rdata *= np.random.uniform(0.5, 1.5) abzr = abz/180.*np.pi dist = kilometers2degrees(dist) #print("aft", dist) if self.angle_only: dist = 1.0 sinx = np.sin(abzr) * dist cosx = np.cos(abzr) * dist #print(dist) phase_label = np.zeros((1, self.length, 5)) 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 for pkey in pidx: if pkey not in self.phase_dict:continue pid = self.phase_dict[pkey] idx = (pidx[pkey] - cidx) if idx > 0 and idx < self.length: phase_label[0, :, pid+1] = norm(t, idx*0.01, 0.15) llen = self.length//self.stride lppn_label = np.zeros([1, llen, 2]) phase_idx = {} for pkey in pidx: pid = self.phase_dict[pkey] idx = (pidx[pkey] - cidx)//self.stride phase_idx[pid] = pidx[pkey] - cidx if idx-1>0: lppn_label[0, idx-1:idx+2] = -1 if idx > 0 and idx < llen: lppn_label[0, idx, 0] = pid + 1 lppn_label[0, idx, 1] = (pidx[pkey] - cidx)%self.stride label = np.array([[cosx, sinx, edep, emag, snr]]) phase_label[:, :, 0] = np.clip(1-np.sum(phase_label[:, :, 1:], axis=2), 0, 1) dqueue.put([rdata, label, [bbidx, phase_idx], phase_label, lppn_label]) count += 1 def batch_data(self, batch_size=32): x1, x2, x3 = [], [], [] x4 = [] x5 = [] for _ in range(batch_size): data, label, pidx, pha, lppn = self.dqueue.get() #print(data.shape, label.shape) x1.append(data) x2.append(label) x3.append(pidx) x4.append(pha) x5.append(lppn) x1 = np.concatenate(x1, axis=0) x2 = np.concatenate(x2, axis=0) x4 = np.concatenate(x4, axis=0) x5 = np.concatenate(x5, axis=0) return x1, x2, x3, x4, x5 class DataThreadWithClass(): def __init__(self, file_name="h5data", n_length=256, stride=16, padlen=64, mindist=0, maxdist=1000, istest=False, angle_only=False, filter_by_snr=False, eq_filter_level=0.95, my_std=0.20, phase_balance=False): self.file_name = file_name self.length = n_length self.stride = stride self.padlen = padlen self.n_thread = 1 self.my_std = my_std self.maxdist = maxdist self.mindist = mindist self.angle_only = angle_only self.filter_by_snr = filter_by_snr self.eq_filter_level = eq_filter_level self.phase_balance = phase_balance self.phase_dict = { "Pg":0, "Sg":1, #"P":2, #"S":3, "Pn":2, "Sn":3, #"PcP":3, #"PgP":4, #"PgPn":5, #"PgPcP":6, #"PnP":7, #"PnPcP":8, #"PcPP":9, #"PcPPn":10, #"PcPPcP":11, #"PcPnP":12 } if istest: self.phase_dict = { "Pg":0, "Sg":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, 'ep': 1, 'ss': 2, 'sp': 3, 'ot': 4, 'se': 5, 've': 6} fqueue = queue.Queue(100) self.dqueue = queue.Queue(100) self.istest = istest if istest: for i in range(4): p1 = threading.Thread(target=self.feed_data, args=(fqueue, i+2019),daemon=True) p1.start() else: for i in range(50): p1 = threading.Thread(target=self.feed_data, args=(fqueue, i%11+2009),daemon=True) p1.start() for _ in range(self.n_thread): c1 = threading.Thread(target=self.process, args=(fqueue, self.dqueue),daemon=True) c1.start() #multiprocessing.Process(target=self.batch_data, args=(dqueue, )).start() def feed_data(self, fqueue, year): f = open("ayrdata/china.loc", "r", encoding="utf8") sloc = {} for line in f.readlines(): sline = [i for i in line.split(" ") if len(i)>0] skey = ".".join(sline[:3]) loc = [float(sline[3]), float(sline[4])] sloc[skey] = loc while True: h5file = h5py.File(f"ayrdata/csndata/{year}.h5", "r") h5key = np.load(f"ayrdata/keys/{year}.npy") np.random.shuffle(h5key) for ekey in h5key: if self.maxdist>=1000000: if "CB." not in ekey: if np.random.random()<0.5: continue event = h5file[ekey] #if "FJ." not in ekey:continue #for key in event.attrs: # print("ekey", key) if "depth" not in event.attrs: depth = 0.0 #continue mag = event.attrs["mag"] elon = event.attrs["lon"] elat = event.attrs["lat"] edep = event.attrs["depth"] if event.attrs["type"] not in self.etype_dict: continue etype_id = self.etype_dict[event.attrs["type"]] #if etype_id == 0: # if np.random.random()self.maxdist or dist < self.mindist: continue #for key in station.attrs: # print(skey, key, dist, abz) #if "POLARITY.Pg" not in station.attrs:continue #if "POLARITY.Pg.UPDOWN" not in station.attrs:continue #if "POLARITY.Pg.CLARITY" not in station.attrs:continue #ptype1 = station.attrs["POLARITY.Pg.UPDOWN"] #ptype2 = station.attrs["POLARITY.Pg.CLARITY" ] #if ptype1 not in self.ploar1 or ptype2 not in self.ploar2:continue data = [0, 0, 0] for dkey in ["BHE", "BHN", "BHZ"]: if dkey not in station: dkey = dkey.replace("B", "S") if dkey not in station: continue if "E" in dkey: data[0] = station[dkey][:] elif "N" in dkey: data[1] = station[dkey][:] elif "Z" in dkey: data[2] = station[dkey][:] #data.append(station[dkey][:]) btime = datetime.datetime.strptime(station[dkey].attrs['btime'], "%Y/%m/%d %H:%M:%S.%f") if type(data[0])==int or type(data[1])==int or type(data[2])==int:continue #if len(data)!=3:continue phases = {} dists= -1 for akey in station.attrs: if "dist" in akey: dists = float(station.attrs[akey]) if "Pg" in akey: pname = akey.split(".")[-1] if pname in self.phase_dict: phases[pname] = datetime.datetime.strptime(station.attrs[akey], "%Y/%m/%d %H:%M:%S.%f") else: if akey in self.phase_dict: phases[akey] = datetime.datetime.strptime(station.attrs[akey], "%Y/%m/%d %H:%M:%S.%f") if self.phase_balance: if "Pg" in phases or "Sg" in phases: if "Pn" not in phases and "Sn" not in phases: if np.random.random()<0.9: #print("skip data") continue #print("Not skip data") if len(phases)==0:continue #if dist < 0: # print("Dist skip", dist) # continue fqueue.put([data, btime, phases, dist, mag, baz, edep, mag, etype_id]) def process(self, fqueue, dqueue): count = 0 llen = self.length//self.stride while True: data, btime, phases, dist, mags, abz, edep, emag, etype_id = fqueue.get() if dist>self.maxdist or dist < self.mindist: continue 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) dpidx = np.random.randint(self.padlen, self.length-self.padlen) cidx = np.random.choice(plist) - dpidx bbidx = np.clip(np.min(plist)-cidx, 0, self.length) 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.max(w)+1e-6) rdata.append(w[np.newaxis, :, np.newaxis]) if flen: continue rdata = np.concatenate(rdata, axis=2) pre = rdata[0, dpidx-100:dpidx, :] aft = rdata[0, dpidx:dpidx+100, :] snr = np.std(aft) / (np.std(pre) + 1e-6) if self.filter_by_snr: if snr < 5.0: continue rdata /= (np.max(np.abs(rdata)) + 1e-6) #rdata *= np.random.uniform(0.5, 1.5) abzr = abz/180.*np.pi dist = kilometers2degrees(dist) #print("aft", dist) if self.angle_only: dist = 1.0 sinx = np.sin(abzr) * dist cosx = np.cos(abzr) * dist #print(dist) phase_label = np.zeros((1, self.length, 5)) 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 for pkey in pidx: if pkey not in self.phase_dict:continue pid = self.phase_dict[pkey] idx = (pidx[pkey] - cidx) if idx > 0 and idx < self.length: phase_label[0, :, pid+1] = norm(t, idx*0.01, self.my_std) llen = self.length//self.stride lppn_label = np.zeros([1, llen, 2]) phase_idx = {} for pkey in pidx: pid = self.phase_dict[pkey] idx = (pidx[pkey] - cidx)//self.stride phase_idx[pid] = pidx[pkey] - cidx if idx-1>0: lppn_label[0, idx-1:idx+2] = -1 if idx > 0 and idx < llen: lppn_label[0, idx, 0] = pid + 1 lppn_label[0, idx, 1] = (pidx[pkey] - cidx)%self.stride label = np.array([[cosx, sinx, edep, emag, snr]]) phase_label[:, :, 0] = np.clip(1-np.sum(phase_label[:, :, 1:], axis=2), 0, 1) dqueue.put([rdata, label, [bbidx, phase_idx, snr], phase_label, lppn_label, etype_id]) count += 1 def batch_data(self, batch_size=32): x1, x2, x3 = [], [], [] x4 = [] x5 = [] x6 = [] for _ in range(batch_size): data, label, pidx, pha, lppn, eid = self.dqueue.get() #print(data.shape, label.shape) x1.append(data) x2.append(label) x3.append(pidx) x4.append(pha) x5.append(lppn) x6.append(eid) x1 = np.concatenate(x1, axis=0) x2 = np.concatenate(x2, axis=0) x4 = np.concatenate(x4, axis=0) x5 = np.concatenate(x5, axis=0) x6 = np.array(x6) return x1, x2, x3, x4, x5, x6 class DataProcessWithClass(): def __init__(self, file_name="h5data", n_length=256, stride=16, padlen=64, mindist=0, maxdist=1000, istest=False, angle_only=False, filter_by_snr=False, eq_filter_level=0.95, my_std=0.20, phase_balance=False): self.file_name = file_name self.length = n_length self.stride = stride self.padlen = padlen self.n_thread = 64 self.my_std = my_std self.maxdist = maxdist self.mindist = mindist self.angle_only = angle_only self.filter_by_snr = filter_by_snr self.eq_filter_level = eq_filter_level self.phase_balance = phase_balance self.phase_dict = { "Pg":0, "Sg":1, #"P":2, #"S":3, "Pn":2, "Sn":3, #"PcP":3, #"PgP":4, #"PgPn":5, #"PgPcP":6, #"PnP":7, #"PnPcP":8, #"PcPP":9, #"PcPPn":10, #"PcPPcP":11, #"PcPnP":12 } if istest: self.phase_dict = { "Pg":0, "Sg":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, 'ep': 1, 'ss': 2, 'sp': 3, 'ot': 4, 'se': 5, 've': 6} fqueue = multiprocessing.Queue(100) self.dqueue = multiprocessing.Queue(100) self.istest = istest if istest: for i in range(30): p1 = multiprocessing.Process(target=self.feed_data, args=(fqueue, i%3+2020),daemon=True) p1.start() else: for i in range(60): p1 = multiprocessing.Process(target=self.feed_data, args=(fqueue, i%11+2009),daemon=True) p1.start() for _ in range(self.n_thread): c1 = multiprocessing.Process(target=self.process, args=(fqueue, self.dqueue),daemon=True) c1.start() #multiprocessing.Process(target=self.batch_data, args=(dqueue, )).start() def feed_data(self, fqueue, year): f = open("ayrdata/china.loc", "r", encoding="utf8") sloc = {} for line in f.readlines(): sline = [i for i in line.split(" ") if len(i)>0] skey = ".".join(sline[:3]) loc = [float(sline[3]), float(sline[4])] sloc[skey] = loc while True: h5file = h5py.File(f"ayrdata/csndata/{year}.h5", "r") h5key = np.load(f"ayrdata/keys/{year}.npy") np.random.shuffle(h5key) for ekey in h5key: if self.maxdist>=1000000: if "CB." not in ekey: if np.random.random()<0.5: continue event = h5file[ekey] #if "FJ." not in ekey:continue #for key in event.attrs: # print("ekey", key) if "depth" not in event.attrs: depth = 0.0 #continue mag = event.attrs["mag"] elon = event.attrs["lon"] elat = event.attrs["lat"] edep = event.attrs["depth"] if event.attrs["type"] not in self.etype_dict: continue etype_id = self.etype_dict[event.attrs["type"]] if etype_id == 0: if np.random.random()self.maxdist or dist < self.mindist: continue #for key in station.attrs: # print(skey, key, dist, abz) #if "POLARITY.Pg" not in station.attrs:continue #if "POLARITY.Pg.UPDOWN" not in station.attrs:continue #if "POLARITY.Pg.CLARITY" not in station.attrs:continue #ptype1 = station.attrs["POLARITY.Pg.UPDOWN"] #ptype2 = station.attrs["POLARITY.Pg.CLARITY" ] #if ptype1 not in self.ploar1 or ptype2 not in self.ploar2:continue data = [0, 0, 0] for dkey in ["BHE", "BHN", "BHZ"]: if dkey not in station: dkey = dkey.replace("B", "S") if dkey not in station: continue if "E" in dkey: data[0] = station[dkey][:] elif "N" in dkey: data[1] = station[dkey][:] elif "Z" in dkey: data[2] = station[dkey][:] #data.append(station[dkey][:]) btime = datetime.datetime.strptime(station[dkey].attrs['btime'], "%Y/%m/%d %H:%M:%S.%f") if type(data[0])==int or type(data[1])==int or type(data[2])==int:continue #if len(data)!=3:continue phases = {} dists= -1 for akey in station.attrs: if "dist" in akey: dists = float(station.attrs[akey]) if "Pg" in akey: pname = akey.split(".")[-1] if pname in self.phase_dict: phases[pname] = datetime.datetime.strptime(station.attrs[akey], "%Y/%m/%d %H:%M:%S.%f") else: if akey in self.phase_dict: phases[akey] = datetime.datetime.strptime(station.attrs[akey], "%Y/%m/%d %H:%M:%S.%f") if self.phase_balance: if "Pg" in phases or "Sg" in phases: if "Pn" not in phases and "Sn" not in phases: if np.random.random()<0.9: #print("skip data") continue #print("Not skip data") if len(phases)==0:continue #if dist < 0: # print("Dist skip", dist) # continue fqueue.put([data, btime, phases, dist, mag, baz, edep, mag, etype_id]) def process(self, fqueue, dqueue): count = 0 llen = self.length//self.stride while True: data, btime, phases, dist, mags, abz, edep, emag, etype_id = fqueue.get() if dist>self.maxdist or dist < self.mindist: continue 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) dpidx = np.random.randint(self.padlen, self.length-self.padlen) cidx = np.random.choice(plist) - dpidx bbidx = np.clip(np.min(plist)-cidx, 0, self.length) 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.max(w)+1e-6) rdata.append(w[np.newaxis, :, np.newaxis]) if flen: continue rdata = np.concatenate(rdata, axis=2) pre = rdata[0, dpidx-100:dpidx, :] aft = rdata[0, dpidx:dpidx+100, :] snr = np.std(aft) / (np.std(pre) + 1e-6) if self.filter_by_snr: if snr < 5.0: continue rdata /= (np.max(np.abs(rdata)) + 1e-6) #rdata *= np.random.uniform(0.5, 1.5) abzr = abz/180.*np.pi dist = kilometers2degrees(dist) #print("aft", dist) if self.angle_only: dist = 1.0 sinx = np.sin(abzr) * dist cosx = np.cos(abzr) * dist #print(dist) phase_label = np.zeros((1, self.length, 5)) 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 for pkey in pidx: if pkey not in self.phase_dict:continue pid = self.phase_dict[pkey] idx = (pidx[pkey] - cidx) if idx > 0 and idx < self.length: phase_label[0, :, pid+1] = norm(t, idx*0.01, self.my_std) llen = self.length//self.stride lppn_label = np.zeros([1, llen, 2]) phase_idx = {} for pkey in pidx: pid = self.phase_dict[pkey] idx = (pidx[pkey] - cidx)//self.stride phase_idx[pid] = pidx[pkey] - cidx if idx-1>0: lppn_label[0, idx-1:idx+2] = -1 if idx > 0 and idx < llen: lppn_label[0, idx, 0] = pid + 1 lppn_label[0, idx, 1] = (pidx[pkey] - cidx)%self.stride label = np.array([[cosx, sinx, edep, emag, snr]]) phase_label[:, :, 0] = np.clip(1-np.sum(phase_label[:, :, 1:], axis=2), 0, 1) dqueue.put([rdata, label, [bbidx, phase_idx, snr], phase_label, lppn_label, etype_id]) count += 1 def batch_data(self, batch_size=32): x1, x2, x3 = [], [], [] x4 = [] x5 = [] x6 = [] for _ in range(batch_size): data, label, pidx, pha, lppn, eid = self.dqueue.get() #print(data.shape, label.shape) x1.append(data) x2.append(label) x3.append(pidx) x4.append(pha) x5.append(lppn) x6.append(eid) x1 = np.concatenate(x1, axis=0) x2 = np.concatenate(x2, axis=0) x4 = np.concatenate(x4, axis=0) x5 = np.concatenate(x5, axis=0) x6 = np.array(x6) return x1, x2, x3, x4, x5, x6 class DataThreadWithClassV2(): def __init__(self, file_name="h5data", n_length=256, stride=16, padlen=64, mindist=0, maxdist=1000, istest=False, angle_only=False, filter_by_snr=False, eq_filter_level=0.95, my_std=0.20): self.file_name = file_name self.length = n_length self.stride = stride self.padlen = padlen self.n_thread = 1 self.my_std = my_std self.maxdist = maxdist self.mindist = mindist self.angle_only = angle_only self.filter_by_snr = filter_by_snr self.eq_filter_level = eq_filter_level self.phase_dict = { "Pg":0, "Sg":1, #"P":2, #"S":3, "Pn":2, "Sn":3, #"PcP":3, #"PgP":4, #"PgPn":5, #"PgPcP":6, #"PnP":7, #"PnPcP":8, #"PcPP":9, #"PcPPn":10, #"PcPPcP":11, #"PcPnP":12 } if istest: self.phase_dict = { "Pg":0, "Sg":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, 'ep': 1, 'ss': 2, 'sp': 3, 'ot': 4, 'se': 5, 've': 6} fqueue = queue.Queue(100) self.dqueue = queue.Queue(100) self.istest = istest if istest: for i in range(4): p1 = threading.Thread(target=self.feed_data, args=(fqueue, i+2019),daemon=True) p1.start() else: for i in range(10): p1 = threading.Thread(target=self.feed_data, args=(fqueue, i+2009),daemon=True) p1.start() for _ in range(self.n_thread): c1 = threading.Thread(target=self.process, args=(fqueue, self.dqueue),daemon=True) c1.start() #multiprocessing.Process(target=self.batch_data, args=(dqueue, )).start() def feed_data(self, fqueue, year): f = open("ayrdata/china.loc", "r", encoding="utf8") sloc = {} for line in f.readlines(): sline = [i for i in line.split(" ") if len(i)>0] skey = ".".join(sline[:3]) loc = [float(sline[3]), float(sline[4])] sloc[skey] = loc while True: h5file = h5py.File(f"ayrdata/csndata/{year}.h5", "r") h5key = np.load(f"ayrdata/keys/{year}.npy") np.random.shuffle(h5key) for ekey in h5key: if self.maxdist>=1000000: if "CB." not in ekey: if np.random.random()<0.5: continue event = h5file[ekey] #if "FJ." not in ekey:continue #for key in event.attrs: # print("ekey", key) if "depth" not in event.attrs: depth = 0.0 #continue mag = event.attrs["mag"] elon = event.attrs["lon"] elat = event.attrs["lat"] edep = event.attrs["depth"] if event.attrs["type"] not in self.etype_dict: continue etype_id = self.etype_dict[event.attrs["type"]] if etype_id == 0: if np.random.random()self.maxdist or dist < self.mindist: continue #for key in station.attrs: # print(skey, key, dist, abz) #if "POLARITY.Pg" not in station.attrs:continue #if "POLARITY.Pg.UPDOWN" not in station.attrs:continue #if "POLARITY.Pg.CLARITY" not in station.attrs:continue #ptype1 = station.attrs["POLARITY.Pg.UPDOWN"] #ptype2 = station.attrs["POLARITY.Pg.CLARITY" ] #if ptype1 not in self.ploar1 or ptype2 not in self.ploar2:continue data = [0, 0, 0] for dkey in ["BHE", "BHN", "BHZ"]: if dkey not in station: dkey = dkey.replace("B", "S") if dkey not in station: continue if "E" in dkey: data[0] = station[dkey][:] elif "N" in dkey: data[1] = station[dkey][:] elif "Z" in dkey: data[2] = station[dkey][:] #data.append(station[dkey][:]) btime = datetime.datetime.strptime(station[dkey].attrs['btime'], "%Y/%m/%d %H:%M:%S.%f") if type(data[0])==int or type(data[1])==int or type(data[2])==int:continue #if len(data)!=3:continue phases = {} dists= -1 for akey in station.attrs: if "dist" in akey: dists = float(station.attrs[akey]) if "Pg" in akey: pname = akey.split(".")[-1] if pname in self.phase_dict: phases[pname] = datetime.datetime.strptime(station.attrs[akey], "%Y/%m/%d %H:%M:%S.%f") else: if akey in self.phase_dict: phases[akey] = datetime.datetime.strptime(station.attrs[akey], "%Y/%m/%d %H:%M:%S.%f") if len(phases)==0:continue #if dist < 0: # print("Dist skip", dist) # continue fqueue.put([data, btime, phases, dist, mag, baz, edep, mag, etype_id]) def process(self, fqueue, dqueue): count = 0 llen = self.length//self.stride while True: data, btime, phases, dist, mags, abz, edep, emag, etype_id = fqueue.get() if dist>self.maxdist or dist < self.mindist: continue 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) dpidx = np.random.randint(-self.length * 2, self.length * 2) cidx = np.random.choice(plist) - dpidx bbidx = np.clip(np.min(plist)-cidx, 0, self.length) 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.max(w)+1e-6) rdata.append(w[np.newaxis, :, np.newaxis]) if flen: continue rdata = np.concatenate(rdata, axis=2) pre = rdata[0, dpidx-100:dpidx, :] aft = rdata[0, dpidx:dpidx+100, :] snr = np.std(aft) / (np.std(pre) + 1e-6) if self.filter_by_snr: if snr < 5.0: continue rdata /= (np.max(np.abs(rdata)) + 1e-6) #rdata *= np.random.uniform(0.5, 1.5) abzr = abz/180.*np.pi dist = kilometers2degrees(dist) #print("aft", dist) if self.angle_only: dist = 1.0 sinx = np.sin(abzr) * dist cosx = np.cos(abzr) * dist #print(dist) phase_label = np.zeros((1, self.length, 5)) 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 for pkey in pidx: if pkey not in self.phase_dict:continue pid = self.phase_dict[pkey] idx = (pidx[pkey] - cidx) if idx > 0 and idx < self.length: phase_label[0, :, pid+1] = norm(t, idx*0.01, self.my_std) llen = self.length//self.stride lppn_label = np.zeros([1, llen, 2]) phase_idx = {} for pkey in pidx: pid = self.phase_dict[pkey] idx = (pidx[pkey] - cidx)//self.stride phase_idx[pid] = pidx[pkey] - cidx if idx-1>0: lppn_label[0, idx-1:idx+2] = -1 if idx > 0 and idx < llen: lppn_label[0, idx, 0] = pid + 1 lppn_label[0, idx, 1] = (pidx[pkey] - cidx)%self.stride label = np.array([[cosx, sinx, edep, emag, snr]]) phase_label[:, :, 0] = np.clip(1-np.sum(phase_label[:, :, 1:], axis=2), 0, 1) dqueue.put([rdata, label, [bbidx, phase_idx], phase_label, lppn_label, etype_id]) count += 1 def batch_data(self, batch_size=32): x1, x2, x3 = [], [], [] x4 = [] x5 = [] x6 = [] for _ in range(batch_size): data, label, pidx, pha, lppn, eid = self.dqueue.get() #print(data.shape, label.shape) x1.append(data) x2.append(label) x3.append(pidx) x4.append(pha) x5.append(lppn) x6.append(eid) x1 = np.concatenate(x1, axis=0) x2 = np.concatenate(x2, axis=0) x4 = np.concatenate(x4, axis=0) x5 = np.concatenate(x5, axis=0) x6 = np.array(x6) return x1, x2, x3, x4, x5, x6 class DataThreadWithPrior(): def __init__(self, file_name="h5data", n_length=256, stride=16, padlen=64, mindist=0, maxdist=1000, istest=False, angle_only=False, filter_by_snr=False): self.file_name = file_name self.length = n_length self.stride = stride self.padlen = padlen self.n_thread = 1 self.maxdist = maxdist self.mindist = mindist self.angle_only = angle_only self.filter_by_snr = filter_by_snr self.phase_dict = { "PcP":0, "ScS":1, #"P":0, #"S":1, #"Pn":2, #"Sn":3, #"PcP":3, #"PgP":4, #"PgPn":5, #"PgPcP":6, #"PnP":7, #"PnPcP":8, #"PcPP":9, #"PcPPn":10, #"PcPPcP":11, #"PcPnP":12 } self.ploar1 = { "C":0, "U":0, "R":1, "D":1, } self.ploar2 = { "I":0, "M":1, "E":2, } fqueue = queue.Queue(100) self.dqueue = queue.Queue(100) self.istest = istest if istest: for i in range(4): p1 = threading.Thread(target=self.feed_data, args=(fqueue, i+2019),daemon=True) p1.start() else: for i in range(10): p1 = threading.Thread(target=self.feed_data, args=(fqueue, i+2009),daemon=True) p1.start() for _ in range(self.n_thread): c1 = threading.Thread(target=self.process, args=(fqueue, self.dqueue),daemon=True) c1.start() #multiprocessing.Process(target=self.batch_data, args=(dqueue, )).start() def feed_data(self, fqueue, year): f = open("ayrdata/china.loc", "r", encoding="utf8") sloc = {} for line in f.readlines(): sline = [i for i in line.split(" ") if len(i)>0] skey = ".".join(sline[:3]) loc = [float(sline[3]), float(sline[4])] sloc[skey] = loc while True: h5file = h5py.File(f"ayrdata/csndata/{year}.h5", "r") h5key = np.load(f"ayrdata/keys/{year}.npy") np.random.shuffle(h5key) for ekey in h5key: if self.maxdist>=1000000: if "CB." not in ekey: if np.random.random()<0.5: continue event = h5file[ekey] #if "FJ." not in ekey:continue #for key in event.attrs: # print("ekey", key) if "depth" not in event.attrs: depth = 0.0 #continue mag = event.attrs["mag"] elon = event.attrs["lon"] elat = event.attrs["lat"] edep = event.attrs["depth"] #if self.istest: # if mag<0.0:continue keys = [key for key in event] if len(keys)<1:continue sidx = np.random.randint(len(keys)) for skey in keys[sidx:sidx+1]: station = event[skey] if skey not in sloc:continue slon, slat = sloc[skey] try: dist, abz, baz = gps2dist_azimuth(elat, elon, slat, slon) dist = dist/1000 except: continue if dist>self.maxdist or dist < self.mindist: continue #for key in station.attrs: # print(skey, key, dist, abz) #if "POLARITY.Pg" not in station.attrs:continue #if "POLARITY.Pg.UPDOWN" not in station.attrs:continue #if "POLARITY.Pg.CLARITY" not in station.attrs:continue #ptype1 = station.attrs["POLARITY.Pg.UPDOWN"] #ptype2 = station.attrs["POLARITY.Pg.CLARITY" ] #if ptype1 not in self.ploar1 or ptype2 not in self.ploar2:continue data = [0, 0, 0] for dkey in ["BHE", "BHN", "BHZ"]: if dkey not in station: dkey = dkey.replace("B", "S") if dkey not in station: continue if "E" in dkey: data[0] = station[dkey][:] elif "N" in dkey: data[1] = station[dkey][:] elif "Z" in dkey: data[2] = station[dkey][:] #data.append(station[dkey][:]) btime = datetime.datetime.strptime(station[dkey].attrs['btime'], "%Y/%m/%d %H:%M:%S.%f") if type(data[0])==int or type(data[1])==int or type(data[2])==int:continue #if len(data)!=3:continue phases = {} dists= -1 for akey in station.attrs: if "dist" in akey: dists = float(station.attrs[akey]) if "Pg" in akey: pname = akey.split(".")[-1] if pname in self.phase_dict: phases[pname] = datetime.datetime.strptime(station.attrs[akey], "%Y/%m/%d %H:%M:%S.%f") else: if akey in self.phase_dict: phases[akey] = datetime.datetime.strptime(station.attrs[akey], "%Y/%m/%d %H:%M:%S.%f") if len(phases)==0:continue #if dist < 0: # print("Dist skip", dist) # continue fqueue.put([data, btime, phases, dist, mag, baz, edep, mag]) def process(self, fqueue, dqueue): count = 0 llen = self.length//self.stride while True: data, btime, phases, dist, mags, abz, edep, emag = fqueue.get() if dist>self.maxdist or dist < self.mindist: continue 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) dpidx = np.random.randint(self.padlen, self.length-self.padlen) cidx = np.random.choice(plist) - dpidx bbidx = np.clip(np.min(plist)-cidx, 0, self.length) 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.max(w)+1e-6) rdata.append(w[np.newaxis, :, np.newaxis]) if flen: continue rdata = np.concatenate(rdata, axis=2) pre = rdata[0, dpidx-100:dpidx, :] aft = rdata[0, dpidx:dpidx+100, :] snr = np.std(aft) / (np.std(pre) + 1e-6) if self.filter_by_snr: if snr < 5.0: continue rdata /= (np.max(np.abs(rdata)) + 1e-6) merge_data = np.zeros([1, self.length, 6]) merge_data[0, :, :3] = rdata[:, :, :3] merge_data[0, :, 3] = edep merge_data[0, :, 4] = kilometers2degrees(dist) #rdata *= np.random.uniform(0.5, 1.5) abzr = abz/180.*np.pi dist = kilometers2degrees(dist) #print("aft", dist) if self.angle_only: dist = 1.0 sinx = np.sin(abzr) * dist cosx = np.cos(abzr) * dist #print(dist) phase_label = np.zeros((1, self.length, 5)) 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 for pkey in pidx: if pkey not in self.phase_dict:continue pid = self.phase_dict[pkey] idx = (pidx[pkey] - cidx) if idx > 0 and idx < self.length: phase_label[0, :, pid+1] = norm(t, idx*0.01, 0.15) llen = self.length//self.stride lppn_label = np.zeros([1, llen, 2]) phase_idx = {} for pkey in pidx: pid = self.phase_dict[pkey] idx = (pidx[pkey] - cidx)//self.stride phase_idx[pid] = pidx[pkey] - cidx if idx-1>0: lppn_label[0, idx-1:idx+2] = -1 if idx > 0 and idx < llen: lppn_label[0, idx, 0] = pid + 1 lppn_label[0, idx, 1] = (pidx[pkey] - cidx)%self.stride label = np.array([[cosx, sinx, edep, emag, snr]]) phase_label[:, :, 0] = np.clip(1-np.sum(phase_label[:, :, 1:], axis=2), 0, 1) dqueue.put([rdata, label, [bbidx, phase_idx], phase_label, lppn_label]) count += 1 def batch_data(self, batch_size=32): x1, x2, x3 = [], [], [] x4 = [] x5 = [] for _ in range(batch_size): data, label, pidx, pha, lppn = self.dqueue.get() #print(data.shape, label.shape) x1.append(data) x2.append(label) x3.append(pidx) x4.append(pha) x5.append(lppn) x1 = np.concatenate(x1, axis=0) x2 = np.concatenate(x2, axis=0) x4 = np.concatenate(x4, axis=0) x5 = np.concatenate(x5, axis=0) return x1, x2, x3, x4, x5 class DataThreadWithClassFineTunning(): def __init__(self, file_name="h5data", n_length=256, stride=16, padlen=64, mindist=0, maxdist=1000, istest=False, angle_only=False, filter_by_snr=False, eq_filter_level=0.95, my_std=0.20, phase_dict={"Pg":0, "Sg":1, "Pn":2, "Sn":3}): self.file_name = file_name self.length = n_length self.stride = stride self.padlen = padlen self.n_thread = 1 self.my_std = my_std self.maxdist = maxdist self.mindist = mindist self.angle_only = angle_only self.filter_by_snr = filter_by_snr self.eq_filter_level = eq_filter_level self.phase_dict = phase_dict #{ # "Pg":0, # "Sg":1, # #"P":2, # #"S":3, # "Pn":2, # "Sn":3, # #"PcP":3, # #"PgP":4, # #"PgPn":5, # #"PgPcP":6, # #"PnP":7, # #"PnPcP":8, # #"PcPP":9, # #"PcPPn":10, # #"PcPPcP":11, # #"PcPnP":12 #} if istest: self.phase_dict = { "Pg":0, "Sg":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, 'ep': 1, 'ss': 2, 'sp': 3, 'ot': 4, 'se': 5, 've': 6} fqueue = queue.Queue(100) self.dqueue = queue.Queue(100) self.istest = istest if istest: for i in range(4): p1 = threading.Thread(target=self.feed_data, args=(fqueue, i+2019),daemon=True) p1.start() else: for i in range(10): p1 = threading.Thread(target=self.feed_data, args=(fqueue, i+2009),daemon=True) p1.start() for _ in range(self.n_thread): c1 = threading.Thread(target=self.process, args=(fqueue, self.dqueue),daemon=True) c1.start() #multiprocessing.Process(target=self.batch_data, args=(dqueue, )).start() def feed_data(self, fqueue, year): f = open("ayrdata/china.loc", "r", encoding="utf8") sloc = {} for line in f.readlines(): sline = [i for i in line.split(" ") if len(i)>0] skey = ".".join(sline[:3]) loc = [float(sline[3]), float(sline[4]), float(sline[5])] sloc[skey] = loc while True: h5file = h5py.File(f"ayrdata/csndata/{year}.h5", "r") h5key = np.load(f"ayrdata/keys/{year}.npy") np.random.shuffle(h5key) for ekey in h5key: if self.maxdist>=1000000: if "CB." not in ekey: if np.random.random()<0.5: continue event = h5file[ekey] #if "FJ." not in ekey:continue #for key in event.attrs: # print("ekey", key) if "depth" not in event.attrs: depth = 0.0 #continue mag = event.attrs["mag"] elon = event.attrs["lon"] elat = event.attrs["lat"] edep = event.attrs["depth"] etime = datetime.datetime.strptime(event.attrs["time"], "%Y/%m/%d %H:%M:%S.%f") if event.attrs["type"] not in self.etype_dict: continue etype_id = self.etype_dict[event.attrs["type"]] if etype_id == 0: if np.random.random()self.maxdist or dist < self.mindist: continue #for key in station.attrs: # print(skey, key, dist, abz) #if "POLARITY.Pg" not in station.attrs:continue #if "POLARITY.Pg.UPDOWN" not in station.attrs:continue #if "POLARITY.Pg.CLARITY" not in station.attrs:continue #ptype1 = station.attrs["POLARITY.Pg.UPDOWN"] #ptype2 = station.attrs["POLARITY.Pg.CLARITY" ] #if ptype1 not in self.ploar1 or ptype2 not in self.ploar2:continue data = [0, 0, 0] for dkey in ["BHE", "BHN", "BHZ"]: if dkey not in station: dkey = dkey.replace("B", "S") if dkey not in station: continue if "E" in dkey: data[0] = station[dkey][:] elif "N" in dkey: data[1] = station[dkey][:] elif "Z" in dkey: data[2] = station[dkey][:] #data.append(station[dkey][:]) btime = datetime.datetime.strptime(station[dkey].attrs['btime'], "%Y/%m/%d %H:%M:%S.%f") if type(data[0])==int or type(data[1])==int or type(data[2])==int:continue #if len(data)!=3:continue phases = {} dists= -1 for akey in station.attrs: if "dist" in akey: dists = float(station.attrs[akey]) if "Pg" in akey: pname = akey.split(".")[-1] if pname in self.phase_dict: phases[pname] = datetime.datetime.strptime(station.attrs[akey], "%Y/%m/%d %H:%M:%S.%f") else: if akey in self.phase_dict: phases[akey] = datetime.datetime.strptime(station.attrs[akey], "%Y/%m/%d %H:%M:%S.%f") #print(phases) if len(phases)==0:continue #if dist < 0: # print("Dist skip", dist) # continue fqueue.put([data, btime, phases, dist, mag, baz, edep, mag, etype_id, etime, [elon, elat, edep], [slon, slat, sdep]]) def process(self, fqueue, dqueue): count = 0 llen = self.length//self.stride while True: data, btime, phases, dist, mags, abz, edep, emag, etype_id, etime, eloc, sloc = fqueue.get() if dist>self.maxdist or dist < self.mindist: continue 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) dpidx = np.random.randint(self.padlen, self.length-self.padlen) cidx = np.random.choice(plist) - dpidx bbidx = np.clip(np.min(plist)-cidx, 0, self.length) delta = (etime-btime).total_seconds() - cidx / 100.0 reftime = np.arange(self.length) / 100.0 + delta 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.max(w)+1e-6) rdata.append(w[np.newaxis, :, np.newaxis]) if flen: continue rdata = np.concatenate(rdata, axis=2) pre = rdata[0, dpidx-100:dpidx, :] aft = rdata[0, dpidx:dpidx+100, :] snr = np.std(aft) / (np.std(pre) + 1e-6) if self.filter_by_snr: if snr < 5.0: continue rdata /= (np.max(np.abs(rdata)) + 1e-6) info = np.zeros([1, self.length, 11]) info[:, :, 0] = reftime / 3600 info[:, :, 1] = np.cos(eloc[0]/180.0 * np.pi) info[:, :, 2] = np.sin(eloc[0]/180.0 * np.pi) info[:, :, 3] = np.cos(eloc[1]/180.0 * np.pi) info[:, :, 4] = np.sin(eloc[1]/180.0 * np.pi) info[:, :, 5] = eloc[2]/100 info[:, :, 1+5] = np.cos(sloc[0]/180.0 * np.pi) info[:, :, 2+5] = np.sin(sloc[0]/180.0 * np.pi) info[:, :, 3+5] = np.cos(sloc[1]/180.0 * np.pi) info[:, :, 4+5] = np.sin(sloc[1]/180.0 * np.pi) info[:, :, 5+5] = sloc[2]/100 rdata = np.concatenate([rdata, info], axis=2) #rdata *= np.random.uniform(0.5, 1.5) abzr = abz/180.*np.pi dist = kilometers2degrees(dist) #print("aft", dist) if self.angle_only: dist = 1.0 sinx = np.sin(abzr) * dist cosx = np.cos(abzr) * dist #print(dist) phase_label = np.zeros((1, self.length, 5)) 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 for pkey in pidx: if pkey not in self.phase_dict:continue pid = self.phase_dict[pkey] idx = (pidx[pkey] - cidx) if idx > 0 and idx < self.length: phase_label[0, :, pid+1] = norm(t, idx*0.01, self.my_std) llen = self.length//self.stride lppn_label = np.zeros([1, llen, 2]) phase_idx = {} for pkey in pidx: pid = self.phase_dict[pkey] idx = (pidx[pkey] - cidx)//self.stride phase_idx[pid] = pidx[pkey] - cidx if idx-1>0: lppn_label[0, idx-1:idx+2] = -1 if idx > 0 and idx < llen: lppn_label[0, idx, 0] = pid + 1 lppn_label[0, idx, 1] = (pidx[pkey] - cidx)%self.stride label = np.array([[cosx, sinx, edep, emag, snr]]) phase_label[:, :, 0] = np.clip(1-np.sum(phase_label[:, :, 1:], axis=2), 0, 1) #print(phase_idx) dqueue.put([rdata, label, [bbidx, phase_idx, snr], phase_label, lppn_label, etype_id]) count += 1 def batch_data(self, batch_size=32): x1, x2, x3 = [], [], [] x4 = [] x5 = [] x6 = [] for _ in range(batch_size): data, label, pidx, pha, lppn, eid = self.dqueue.get() #print(data.shape, label.shape) x1.append(data) x2.append(label) x3.append(pidx) x4.append(pha) x5.append(lppn) x6.append(eid) x1 = np.concatenate(x1, axis=0) x2 = np.concatenate(x2, axis=0) x4 = np.concatenate(x4, axis=0) x5 = np.concatenate(x5, axis=0) x6 = np.array(x6) return x1, x2, x3, x4, x5, x6