code/utils/data.py

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