import os import obspy import pickle import datetime from datetime import datetime as dt import h5py import numpy as np from obspy.clients.fdsn.header import FDSNNoDataException #from obspy.clients.fdsn import Client from obspy import UTCDateTime import multiprocessing from datetime import timedelta import numpy as np import matplotlib.pyplot as plt from obspy.signal.filter import bandpass from obspy.geodetics.base import gps2dist_azimuth from sqlalchemy import Column, Integer, String, Float, TIMESTAMP, ForeignKey from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import relationship import datetime from sqlalchemy.orm import joinedload from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker Base = declarative_base() class Event(Base): __tablename__ = 'events' event_id = Column(String, primary_key=True)# 不能重复 # 这是地震数据库，里面you包含的信息 origin_time = Column(TIMESTAMP, index=True)#建立索引，提升读取速度 lat = Column(Float) lon = Column(Float) dep = Column(Float) mag = Column(Float) # 关联的震相 phases = relationship("Phase", back_populates="event") class Phase(Base): __tablename__ = 'phases' phase_id = Column(Integer, primary_key=True, autoincrement=True) # 这是外键，即链接到地震数据库 event_id = Column(String, ForeignKey('events.event_id')) # 这个实际上就是地震震相的一些信息，可以再添加其他的 station_id = Column(String, index=True) # 台站编号 network = Column(String) # 网络名（可选） phase_type = Column(String) # 震相类型（如 P, S） pick_time = Column(TIMESTAMP, index=True) # 到时时间 error = Column(Float) # 到时误差（单位秒，可选） station_lat = Column(Float) # 台站经纬度 station_lon = Column(Float) # 关联事件 event = relationship("Event", back_populates="phases") class DataWithNoisyAndGauss(): def __init__(self, h5_file_name="scdata/sc.refineps.h5", key_file_name= "scdata/sc3.npz", n_length=10240, stride=16, padlen=256, noise_prob = 0.1, std = 0.1): self.h5_file_name = h5_file_name self.key_file_name = key_file_name self.length = n_length self.stride = stride self.padlen = padlen self.n_thread = 2 self.std = std self.noise_prob = noise_prob self.phase_dict = { "Pg":0, "Sg":1, "Pn":2 # "Sn":3 } self.phase_dict2 = { "Pg":0, "Sg":1, # "Pn":2 # "Sn":3 } self.engine = create_engine('sqlite:///ayrdata/phases/phaseCSNCD.db') self.Session = sessionmaker(bind=self.engine) fqueue = multiprocessing.Queue(100) self.dqueue = multiprocessing.Queue(100) for year in [2009+i for i in range(11)]: 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] for skey in event: #print(skey) station = event[skey] data = [0, 0, 0] 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 #print(data) phases = {} phase_count = {"P":0, "S":0} dist = -1 for akey in station.attrs: if "dist" in akey: dist = float(station.attrs[akey]) #print(dist) if "Pg" in akey: pname = akey.split(".")[-1] if pname in self.phase_dict: phase_count["P"] += 1 phases[pname] = datetime.datetime.strptime(station.attrs[akey], "%Y/%m/%d %H:%M:%S.%f") #phases["Noisy"] = datetime.datetime.strptime(station.attrs[akey], "%Y/%m/%d %H:%M:%S.%f") - datetime.timedelta(seconds=5) else: if akey in self.phase_dict: #print(type(station.attrs[akey])) #print(station.attrs[akey]) phases[akey] = 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 dist > 800:continue #if phase_count["P"] == 0 or phase_count["S"] == 0:continue #if "Pn" not in phases:continue if len(phases)==0 or (len(phases) == 1 and 'Pn' in phases):continue fqueue.put([data, btime, phases, skey]) def process(self, fqueue, dqueue): count = 0 llen = self.length//self.stride session = self.Session() while True: data, btime, phases, skey = fqueue.get() pidx = {} plist = [] if "Pn" in phases: noisytime = phases["Pn"] - datetime.timedelta(seconds=5) noisydelta = (noisytime-btime).total_seconds() noisydelta_idx = int(noisydelta * 100) elif "Pn" not in phases and "Pg" in phases: noisytime = phases["Pg"] - datetime.timedelta(seconds=5) noisydelta = (noisytime-btime).total_seconds() noisydelta_idx = int(noisydelta * 100) else: noisydelta_idx = 0 if 'Pn' in phases: del phases['Pn'] if noisydelta_idx < self.length:continue start_time = btime end_time = btime + timedelta(seconds=(self.length/100)) data4noisy = session.query(Phase).filter( Phase.station_id == skey, Phase.pick_time >= start_time, Phase.pick_time <= end_time ).order_by(Phase.pick_time).all() if len(data4noisy) != 0: #for phase in data4noisy: # print(f"{phase.pick_time} | {phase.station_id} | {phase.phase_type} ") continue 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 if np.random.random() < self.noise_prob: for d in data: w = d[0: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) #label1 = np.zeros([1, llen, 2]) label1 = np.zeros([1, self.length, 3]) label1[0, :, 0] = 1 else: rdatabtime = btime + timedelta(seconds=(cidx/100)) rdataendtime = rdatabtime + timedelta(seconds=(self.length/100)) data4label = session.query(Phase).filter( Phase.station_id == skey, Phase.pick_time >= rdatabtime, Phase.pick_time <= rdataendtime )\ .group_by(Phase.station_id, Phase.pick_time, Phase.phase_type)\ .order_by(Phase.pick_time)\ .all() phases2 = {} for phase in data4label: if phase.phase_type in self.phase_dict2: pdelta = (phase.pick_time - rdatabtime).total_seconds() if pdelta < 0:continue if phase.phase_type not in phases2: phases2[phase.phase_type] = [] phases2[phase.phase_type].append(int(pdelta*100)) #print(phases2) #print(pidx) #print("phase index:", pidx2) #print(phases) #print(f"{phase.pick_time} | {phase.station_id} | {phase.phase_type} ") #print(len(data4label),skey, phases) 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) label1 = np.zeros([1, self.length, 3]) 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 #flen = False for pkey in phases2: pid = self.phase_dict[pkey] idxs = phases2[pkey] #if len(idxs) != 2: # flen = True # break for idx in idxs: label1[0, :, pid+1] += norm(t, idx*0.01, std=self.std) label1[0, :, pid+1] /= np.max(label1[0, :, pid+1]) if flen:continue label1[0, :, 0] = np.clip(1-label1[0, :, 1]-label1[0, :, 2], 0, 1) #label2[0, phase_intv["P"]:phase_intv["S"], 3] = 1 dqueue.put([rdata, label1]) count += 1 def batch_data(self, batch_size=32): x1, x2, x3 = [], [], [] for _ in range(batch_size): data, label1 = self.dqueue.get() x1.append(data) x2.append(label1) x1 = np.concatenate(x1, axis=0) x2 = np.concatenate(x2, axis=0) return x1, x2 class DataWithNoisy(): def __init__(self, h5_file_name="scdata/sc.refineps.h5", key_file_name= "scdata/sc3.npz", n_length=10240, stride=16, padlen=256, noise_prob = 0.1, std=0.1): self.h5_file_name = h5_file_name self.key_file_name = key_file_name self.length = n_length self.stride = stride self.padlen = padlen self.n_thread = 2 self.std = std self.noise_prob = noise_prob self.phase_dict = { "Pg":0, "Sg":1, "Pn":2 # "Sn":3 } self.phase_dict2 = { "Pg":0, "Sg":1, # "Pn":2 # "Sn":3 } self.engine = create_engine('sqlite:///ayrdata/phases/phaseCSNCD.db') self.Session = sessionmaker(bind=self.engine) fqueue = multiprocessing.Queue(100) self.dqueue = multiprocessing.Queue(100) for year in [2009+i for i in range(11)]: 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] for skey in event: #print(skey) station = event[skey] data = [0, 0, 0] 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 #print(data) phases = {} phase_count = {"P":0, "S":0} dist = -1 for akey in station.attrs: if "dist" in akey: dist = float(station.attrs[akey]) #print(dist) if "Pg" in akey: pname = akey.split(".")[-1] if pname in self.phase_dict: phase_count["P"] += 1 phases[pname] = datetime.datetime.strptime(station.attrs[akey], "%Y/%m/%d %H:%M:%S.%f") #phases["Noisy"] = datetime.datetime.strptime(station.attrs[akey], "%Y/%m/%d %H:%M:%S.%f") - datetime.timedelta(seconds=5) else: if akey in self.phase_dict: #print(type(station.attrs[akey])) #print(station.attrs[akey]) phases[akey] = 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 dist > 800:continue #if phase_count["P"] == 0 or phase_count["S"] == 0:continue #if "Pn" not in phases:continue if len(phases)==0 or (len(phases) == 1 and 'Pn' in phases):continue fqueue.put([data, btime, phases, skey]) def process(self, fqueue, dqueue): count = 0 llen = self.length//self.stride session = self.Session() while True: data, btime, phases, skey = fqueue.get() pidx = {} plist = [] if "Pn" in phases: noisytime = phases["Pn"] - datetime.timedelta(seconds=5) noisydelta = (noisytime-btime).total_seconds() noisydelta_idx = int(noisydelta * 100) elif "Pn" not in phases and "Pg" in phases: noisytime = phases["Pg"] - datetime.timedelta(seconds=5) noisydelta = (noisytime-btime).total_seconds() noisydelta_idx = int(noisydelta * 100) else: noisydelta_idx = 0 if 'Pn' in phases: del phases['Pn'] if noisydelta_idx < self.length:continue start_time = btime end_time = btime + timedelta(seconds=(self.length/100)) data4noisy = session.query(Phase).filter( Phase.station_id == skey, Phase.pick_time >= start_time, Phase.pick_time <= end_time ).order_by(Phase.pick_time).all() if len(data4noisy) != 0: #for phase in data4noisy: # print(f"{phase.pick_time} | {phase.station_id} | {phase.phase_type} ") continue 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 if np.random.random() < self.noise_prob: for d in data: w = d[0: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) #label1 = np.zeros([1, llen, 2]) label1 = np.zeros([1, self.length, 3]) label1[0, :, 0] = 1 else: rdatabtime = btime + timedelta(seconds=(cidx/100)) rdataendtime = rdatabtime + timedelta(seconds=(self.length/100)) data4label = session.query(Phase).filter( Phase.station_id == skey, Phase.pick_time >= rdatabtime, Phase.pick_time <= rdataendtime )\ .group_by(Phase.station_id, Phase.pick_time, Phase.phase_type)\ .order_by(Phase.pick_time)\ .all() phases2 = {} for phase in data4label: if phase.phase_type in self.phase_dict2: pdelta = (phase.pick_time - rdatabtime).total_seconds() if pdelta < 0:continue if phase.phase_type not in phases2: phases2[phase.phase_type] = [] phases2[phase.phase_type].append(int(pdelta*100)) #print(phases2) #print(pidx) #print("phase index:", pidx2) #print(phases) #print(f"{phase.pick_time} | {phase.station_id} | {phase.phase_type} ") #print(len(data4label),skey, phases) 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) label1 = np.zeros([1, self.length, 3]) 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 #flen = False for pkey in phases2: pid = self.phase_dict[pkey] idxs = phases2[pkey] #if len(idxs) != 2: # flen = True # break for idx in idxs: label1[0, :, pid+1] += norm(t, idx*0.01, self.std) label1[0, :, pid+1] /= np.max(label1[0, :, pid+1]) if flen:continue label1[0, :, 0] = np.clip(1-label1[0, :, 1]-label1[0, :, 2], 0, 1) #label2[0, phase_intv["P"]:phase_intv["S"], 3] = 1 dqueue.put([rdata, label1]) count += 1 def batch_data(self, batch_size=32): x1, x2, x3 = [], [], [] for _ in range(batch_size): data, label1 = self.dqueue.get() x1.append(data) x2.append(label1) x1 = np.concatenate(x1, axis=0) x2 = np.concatenate(x2, axis=0) return x1, x2 class DataWithNoisyTest(): def __init__(self, h5_file_name="scdata/sc.refineps.h5", key_file_name= "scdata/sc3.npz", n_length=10240, stride=16, padlen=256, noise_prob = 0.1): self.h5_file_name = h5_file_name self.key_file_name = key_file_name self.length = n_length self.stride = stride self.padlen = padlen self.n_thread = 2 self.noise_prob = noise_prob self.phase_dict = { "Pg":0, "Sg":1, "Pn":2, # "Sn":3 } self.phase_dict = { "Pg":0, "Sg":1, # "Pn":2, # "Sn":3 } self.engine = create_engine('sqlite:///ayrdata/phases/phaseCSNCD.db') self.Session = sessionmaker(bind=self.engine) 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") 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[:2]) loc = [float(sline[3]), float(sline[4])] sloc[skey] = loc #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] elon = event.attrs["lon"] elat = event.attrs["lat"] for skey in event: sskey = ".".join(skey.split(".")[:2]) if sskey not in sloc: print("Station not found!") continue slon, slat = sloc[sskey]#台站位置 try: dist, abz, baz = gps2dist_azimuth(elat, elon, slat, slon) dist = dist/1000 except: continue #print(skey) station = event[skey] data = [0, 0, 0] 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 #print(data) phases = {} phase_count = {"P":0, "S":0} for akey in station.attrs: if "Pg" in akey: pname = akey.split(".")[-1] if pname in self.phase_dict: phase_count["P"] += 1 phases[pname] = datetime.datetime.strptime(station.attrs[akey], "%Y/%m/%d %H:%M:%S.%f") #phases["Noisy"] = datetime.datetime.strptime(station.attrs[akey], "%Y/%m/%d %H:%M:%S.%f") - datetime.timedelta(seconds=5) else: if akey in self.phase_dict: #print(type(station.attrs[akey])) #print(station.attrs[akey]) phases[akey] = 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 dist > 800:continue #if phase_count["P"] == 0 or phase_count["S"] == 0:continue #if "Pn" not in phases:continue if len(phases)==0 or (len(phases) == 1 and 'Pn' in phases):continue fqueue.put([data, btime, phases, ekey, skey, dist]) def process(self, fqueue, dqueue): count = 0 llen = self.length//self.stride session = self.Session() while True: data, btime, phases, ekey, skey, dist = fqueue.get() pidx = {} plist = [] if "Pn" in phases: noisytime = phases["Pn"] - datetime.timedelta(seconds=5) noisydelta = (noisytime-btime).total_seconds() noisydelta_idx = int(noisydelta * 100) elif "Pn" not in phases and "Pg" in phases: noisytime = phases["Pg"] - datetime.timedelta(seconds=5) noisydelta = (noisytime-btime).total_seconds() noisydelta_idx = int(noisydelta * 100) else: noisydelta_idx = 0 if 'Pn' in phases: del phases['Pn'] if noisydelta_idx < self.length:continue start_time = btime end_time = btime + timedelta(seconds=(self.length/100)) data4noisy = session.query(Phase).filter( Phase.station_id == skey, Phase.pick_time >= start_time, Phase.pick_time <= end_time ).order_by(Phase.pick_time).all() if len(data4noisy) != 0:continue 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 if np.random.random() < self.noise_prob: for d in data: w = d[0: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) label1 = np.zeros([1, llen, 2]) phase_time = {0:-1, 1:-1} snrs = [-10000, -10000] else: rdatabtime = btime + timedelta(seconds=(cidx/100)) rdataendtime = rdatabtime + timedelta(seconds=(self.length/100)) data4label = session.query(Phase).filter( Phase.station_id == skey, Phase.pick_time >= rdatabtime, Phase.pick_time <= rdataendtime )\ .group_by(Phase.station_id, Phase.pick_time, Phase.phase_type)\ .order_by(Phase.pick_time)\ .all() phases2 = {} 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) phase_time = {0:-1, 1:-1} snrs = [-10000, -10000] def squ(x): x -= np.mean(x) return np.sum(x**2) for pkey in pidx: pid = self.phase_dict[pkey] idx = (pidx[pkey] - cidx) if idx > 200 and idx < self.length-200: phase_time[pid] = idx w = rdata[0] epre = w[idx-150:idx, 0] eaft = w[idx:idx+150, 0] npre = w[idx-150:idx, 1] naft = w[idx:idx+150, 1] zpre = w[idx-50:idx, 2] zaft = w[idx:idx+50, 2] esnr = 10 * np.log10((squ(eaft))/(squ(epre)+1e-6)+1e-6) nsnr = 10 * np.log10((squ(naft))/(squ(npre)+1e-6)+1e-6) zsnr = 10 * np.log10((squ(zaft))/(squ(zpre)+1e-6)+1e-6) if pkey in ["Pg", "P"]: snrs[pid] = zsnr elif pkey in ["Sg", "S"]: snrs[pid] = (esnr + nsnr) / 2 dqueue.put([rdata, [phase_time[0], phase_time[1]], ekey, skey, snrs, dist]) count += 1 def batch_data(self, batch_size=50): x1, x2, x3, x4, x5,x6 = [], [], [], [], [],[] for _ in range(batch_size): data, label1, ekey ,skey, snrs, dist= self.dqueue.get() x1.append(data) x2.append(label1) x3.append(ekey) x4.append(skey) x5.append(snrs) x6.append(dist) x1 = np.concatenate(x1, axis=0) return x1, x2, x3, x4, x5, x6 class DataWithNoisyAndMaxDist(): def __init__(self, h5_file_name="scdata/sc.refineps.h5", key_file_name= "scdata/sc3.npz", n_length=10240, stride=16, padlen=256, noise_prob = 0.1, std=0.1): self.h5_file_name = h5_file_name self.key_file_name = key_file_name self.length = n_length self.stride = stride self.padlen = padlen self.n_thread = 2 self.std = std self.noise_prob = noise_prob self.phase_dict = { "Pg":0, "Sg":1, "Pn":2 # "Sn":3 } self.phase_dict2 = { "Pg":0, "Sg":1, # "Pn":2 # "Sn":3 } self.engine = create_engine('sqlite:///ayrdata/phases/phaseCSNCD.db') self.Session = sessionmaker(bind=self.engine) fqueue = multiprocessing.Queue(100) self.dqueue = multiprocessing.Queue(100) for year in [2009+i for i in range(11)]: 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] for skey in event: #print(skey) station = event[skey] data = [0, 0, 0] 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 #print(data) phases = {} phase_count = {"P":0, "S":0} dist = -1 for akey in station.attrs: if "dist" in akey: dist = float(station.attrs[akey]) #print(dist) if "Pg" in akey: pname = akey.split(".")[-1] if pname in self.phase_dict: phase_count["P"] += 1 phases[pname] = datetime.datetime.strptime(station.attrs[akey], "%Y/%m/%d %H:%M:%S.%f") #phases["Noisy"] = datetime.datetime.strptime(station.attrs[akey], "%Y/%m/%d %H:%M:%S.%f") - datetime.timedelta(seconds=5) else: if akey in self.phase_dict: #print(type(station.attrs[akey])) #print(station.attrs[akey]) phases[akey] = 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 dist > 800:continue #if phase_count["P"] == 0 or phase_count["S"] == 0:continue #if "Pn" not in phases:continue if dist == -1: continue #if dist < 300:continue #print(dist) if len(phases)==0 or (len(phases) == 1 and 'Pn' in phases):continue fqueue.put([data, btime, phases, skey, dist]) def process(self, fqueue, dqueue): count = 0 llen = self.length//self.stride session = self.Session() while True: data, btime, phases, skey, dist = fqueue.get() pidx = {} plist = [] if "Pn" in phases: noisytime = phases["Pn"] - datetime.timedelta(seconds=5) noisydelta = (noisytime-btime).total_seconds() noisydelta_idx = int(noisydelta * 100) elif "Pn" not in phases and "Pg" in phases: noisytime = phases["Pg"] - datetime.timedelta(seconds=5) noisydelta = (noisytime-btime).total_seconds() noisydelta_idx = int(noisydelta * 100) else: noisydelta_idx = 0 if 'Pn' in phases: del phases['Pn'] if noisydelta_idx < self.length:continue start_time = btime end_time = btime + timedelta(seconds=(self.length/100)) data4noisy = session.query(Phase).filter( Phase.station_id == skey, Phase.pick_time >= start_time, Phase.pick_time <= end_time ).order_by(Phase.pick_time).all() if len(data4noisy) != 0: #for phase in data4noisy: # print(f"{phase.pick_time} | {phase.station_id} | {phase.phase_type} ") continue 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 if np.random.random() < self.noise_prob: for d in data: w = d[0: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) #label1 = np.zeros([1, llen, 2]) label1 = np.zeros([1, self.length, 3]) label1[0, :, 0] = 1 distclass = 0 else: if dist < 300: distclass = 1 else: distclass = 2 rdatabtime = btime + timedelta(seconds=(cidx/100)) rdataendtime = rdatabtime + timedelta(seconds=(self.length/100)) data4label = session.query(Phase).filter( Phase.station_id == skey, Phase.pick_time >= rdatabtime, Phase.pick_time <= rdataendtime )\ .group_by(Phase.station_id, Phase.pick_time, Phase.phase_type)\ .order_by(Phase.pick_time)\ .all() phases2 = {} for phase in data4label: if phase.phase_type in self.phase_dict2: pdelta = (phase.pick_time - rdatabtime).total_seconds() if pdelta < 0:continue if phase.phase_type not in phases2: phases2[phase.phase_type] = [] phases2[phase.phase_type].append(int(pdelta*100)) #print(phases2) #print(pidx) #print("phase index:", pidx2) #print(phases) #print(f"{phase.pick_time} | {phase.station_id} | {phase.phase_type} ") #print(len(data4label),skey, phases) 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) label1 = np.zeros([1, self.length, 3]) 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 #flen = False for pkey in phases2: pid = self.phase_dict[pkey] idxs = phases2[pkey] #if len(idxs) != 2: # flen = True # break for idx in idxs: label1[0, :, pid+1] += norm(t, idx*0.01, self.std) label1[0, :, pid+1] /= np.max(label1[0, :, pid+1]) if flen:continue label1[0, :, 0] = np.clip(1-label1[0, :, 1]-label1[0, :, 2], 0, 1) #label2[0, phase_intv["P"]:phase_intv["S"], 3] = 1 dqueue.put([rdata, label1, distclass]) count += 1 def batch_data(self, batch_size=32): x1, x2, x3 = [], [], [] for _ in range(batch_size): data, label1, distclass = self.dqueue.get() x1.append(data) x2.append(label1) x3.append(distclass) x1 = np.concatenate(x1, axis=0) x2 = np.concatenate(x2, axis=0) return x1, x2, x3 class DataWithNoisyMaxDistTest(): def __init__(self, h5_file_name="scdata/sc.refineps.h5", key_file_name= "scdata/sc3.npz", n_length=10240, stride=16, padlen=256, noise_prob = 0.1): self.h5_file_name = h5_file_name self.key_file_name = key_file_name self.length = n_length self.stride = stride self.padlen = padlen self.n_thread = 2 self.noise_prob = noise_prob self.phase_dict = { "Pg":0, "Sg":1, "Pn":2, # "Sn":3 } self.phase_dict = { "Pg":0, "Sg":1, # "Pn":2, # "Sn":3 } self.engine = create_engine('sqlite:///ayrdata/phases/phaseCSNCD.db') self.Session = sessionmaker(bind=self.engine) 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] elon = event.attrs["lon"] elat = event.attrs["lat"] for skey in event: #print(skey) station = event[skey] data = [0, 0, 0] 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 #print(data) phases = {} phase_count = {"P":0, "S":0} dist = -1 for akey in station.attrs: if "dist" in akey: dist = float(station.attrs[akey]) #print(dist) if "Pg" in akey: pname = akey.split(".")[-1] if pname in self.phase_dict: phase_count["P"] += 1 phases[pname] = datetime.datetime.strptime(station.attrs[akey], "%Y/%m/%d %H:%M:%S.%f") #phases["Noisy"] = datetime.datetime.strptime(station.attrs[akey], "%Y/%m/%d %H:%M:%S.%f") - datetime.timedelta(seconds=5) else: if akey in self.phase_dict: #print(type(station.attrs[akey])) #print(station.attrs[akey]) phases[akey] = 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 dist == -1:continue #if dist == -1:continue #if phase_count["P"] == 0 or phase_count["S"] == 0:continue #if "Pn" not in phases:continue if len(phases)==0 or (len(phases) == 1 and 'Pn' in phases):continue fqueue.put([data, btime, phases, ekey, skey, dist]) def process(self, fqueue, dqueue): count = 0 llen = self.length//self.stride session = self.Session() while True: data, btime, phases, ekey, skey, dist = fqueue.get() pidx = {} plist = [] if "Pn" in phases: noisytime = phases["Pn"] - datetime.timedelta(seconds=5) noisydelta = (noisytime-btime).total_seconds() noisydelta_idx = int(noisydelta * 100) elif "Pn" not in phases and "Pg" in phases: noisytime = phases["Pg"] - datetime.timedelta(seconds=5) noisydelta = (noisytime-btime).total_seconds() noisydelta_idx = int(noisydelta * 100) else: noisydelta_idx = 0 if 'Pn' in phases: del phases['Pn'] if noisydelta_idx < self.length:continue start_time = btime end_time = btime + timedelta(seconds=(self.length/100)) data4noisy = session.query(Phase).filter( Phase.station_id == skey, Phase.pick_time >= start_time, Phase.pick_time <= end_time ).order_by(Phase.pick_time).all() if len(data4noisy) != 0:continue 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) data_btime = btime + timedelta(seconds=cidx/100) rdata = [] flen = False if np.random.random() < self.noise_prob: for d in data: w = d[0: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) label1 = np.zeros([1, llen, 2]) phase_time = {0:-1, 1:-1} snrs = [-10000, -10000] distclass = 0 else: if dist < 300: distclass = 1 else: distclass = 2 rdatabtime = btime + timedelta(seconds=(cidx/100)) rdataendtime = rdatabtime + timedelta(seconds=(self.length/100)) data4label = session.query(Phase).filter( Phase.station_id == skey, Phase.pick_time >= rdatabtime, Phase.pick_time <= rdataendtime )\ .group_by(Phase.station_id, Phase.pick_time, Phase.phase_type)\ .order_by(Phase.pick_time)\ .all() phases2 = {} 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) phase_time = {0:-1, 1:-1} snrs = [-10000, -10000] def squ(x): x -= np.mean(x) return np.sum(x**2) for pkey in pidx: pid = self.phase_dict[pkey] idx = (pidx[pkey] - cidx) if idx > 200 and idx < self.length-200: phase_time[pid] = idx w = rdata[0] epre = w[idx-150:idx, 0] eaft = w[idx:idx+150, 0] npre = w[idx-150:idx, 1] naft = w[idx:idx+150, 1] zpre = w[idx-50:idx, 2] zaft = w[idx:idx+50, 2] esnr = 10 * np.log10((squ(eaft))/(squ(epre)+1e-6)+1e-6) nsnr = 10 * np.log10((squ(naft))/(squ(npre)+1e-6)+1e-6) zsnr = 10 * np.log10((squ(zaft))/(squ(zpre)+1e-6)+1e-6) if pkey in ["Pg", "P"]: snrs[pid] = zsnr elif pkey in ["Sg", "S"]: snrs[pid] = (esnr + nsnr) / 2 dqueue.put([rdata, [phase_time[0], phase_time[1]], ekey, skey, snrs, dist, distclass, data_btime]) count += 1 def batch_data(self, batch_size=50): x1, x2, x3, x4, x5, x6, x7, x8 = [], [], [], [], [], [], [], [] for _ in range(batch_size): data, label1, ekey ,skey, snrs, dist, distclass, data_btime= self.dqueue.get() x1.append(data) x2.append(label1) x3.append(ekey) x4.append(skey) x5.append(snrs) x6.append(dist) x7.append(distclass) x8.append(data_btime) x1 = np.concatenate(x1, axis=0) return x1, x2, x3, x4, x5, x6, x7, x8 class DataPnSnTest(): def __init__(self, h5_file_name="scdata/sc.refineps.h5", key_file_name= "scdata/sc3.npz", n_length=10240, stride=16, padlen=256, noise_prob = 0.1): self.h5_file_name = h5_file_name self.key_file_name = key_file_name self.length = n_length self.stride = stride self.padlen = padlen self.n_thread = 2 self.phase_dict = { "Pg":0, "Sg":1, "Pn":2, "Sn":3 } self.engine = create_engine('sqlite:///ayrdata/phases/phaseCSNCD.db') self.Session = sessionmaker(bind=self.engine) fqueue = multiprocessing.Queue(100) self.dqueue = multiprocessing.Queue(100) for year in [2009+i for i in range(11)]: 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") 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[:2]) loc = [float(sline[3]), float(sline[4])] sloc[skey] = loc #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] elon = event.attrs["lon"] elat = event.attrs["lat"] for skey in event: sskey = ".".join(skey.split(".")[:2]) if sskey not in sloc: print("Station not found!") continue slon, slat = sloc[sskey]#台站位置 try: dist, abz, baz = gps2dist_azimuth(elat, elon, slat, slon) dist = dist/1000 except: continue #print(skey) station = event[skey] data = [0, 0, 0] 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 #print(data) phases = {} phase_count = {"P":0, "S":0} for akey in station.attrs: if "Pg" in akey: pname = akey.split(".")[-1] if pname in self.phase_dict: phase_count["P"] += 1 phases[pname] = datetime.datetime.strptime(station.attrs[akey], "%Y/%m/%d %H:%M:%S.%f") #phases["Noisy"] = datetime.datetime.strptime(station.attrs[akey], "%Y/%m/%d %H:%M:%S.%f") - datetime.timedelta(seconds=5) else: if akey in self.phase_dict: #print(type(station.attrs[akey])) #print(station.attrs[akey]) phases[akey] = 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 dist < 400:continue #if dist < 100:continue #if phase_count["P"] == 0 or phase_count["S"] == 0:continue #if "Pn" not in phases:continue if len(phases)==0 or (len(phases) == 1 and 'Pn' in phases):continue fqueue.put([data, btime, phases, ekey, skey, dist]) def process(self, fqueue, dqueue): count = 0 llen = self.length//self.stride session = self.Session() while True: data, btime, phases, ekey, skey, dist = 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.max(w)+1e-6) rdata.append(w[np.newaxis, :, np.newaxis]) if flen:continue rdata = np.concatenate(rdata, axis=2) dqueue.put([rdata, ekey, skey, dist]) count += 1 def batch_data(self, batch_size=50): x1, x2, x3, x4, x5,x6 = [], [], [], [], [],[] for _ in range(batch_size): data, ekey ,skey, dist= self.dqueue.get() x1.append(data) x2.append(ekey) x4.append(skey) x4.append(dist) x1 = np.concatenate(x1, axis=0) return x1, x2, x3, x4