def _subspace_process(streams, lowcut, highcut, filt_order, sampling_rate,<EOL>multiplex, align, shift_len, reject, no_missed=True,<EOL>stachans=None, parallel=False, plot=False, cores=<NUM_LIT:1>):
from multiprocessing import Pool, cpu_count<EOL>processed_streams = []<EOL>if not stachans:<EOL><INDENT>input_stachans = list(set([(tr.stats.station, tr.stats.channel)<EOL>for st in streams for tr in st.sort()]))<EOL><DEDENT>else:<EOL><INDENT>input_stachans = stachans<EOL><DEDENT>input_stachans.sort()  <EOL>first_length = len(streams[<NUM_LIT:0>][<NUM_LIT:0>].data) /streams[<NUM_LIT:0>][<NUM_LIT:0>].stats.sampling_rate<EOL>for st in streams:<EOL><INDENT>for tr in st:<EOL><INDENT>if not len(tr) / tr.stats.sampling_rate == first_length:<EOL><INDENT>msg = '<STR_LIT>'<EOL>raise IOError(msg)<EOL><DEDENT><DEDENT><DEDENT>for st in streams:<EOL><INDENT>if not parallel:<EOL><INDENT>processed_stream = Stream()<EOL>for stachan in input_stachans:<EOL><INDENT>dummy, tr = _internal_process(<EOL>st=st, lowcut=lowcut, highcut=highcut,<EOL>filt_order=filt_order, sampling_rate=sampling_rate,<EOL>first_length=first_length, stachan=stachan, debug=<NUM_LIT:0>)<EOL>processed_stream += tr<EOL><DEDENT>processed_streams.append(processed_stream)<EOL><DEDENT>else:<EOL><INDENT>pool = Pool(processes=min(cores, cpu_count()))<EOL>results = [pool.apply_async(<EOL>_internal_process, (st,),<EOL>{'<STR_LIT>': lowcut, '<STR_LIT>': highcut,<EOL>'<STR_LIT>': filt_order, '<STR_LIT>': sampling_rate,<EOL>'<STR_LIT>': first_length, '<STR_LIT>': stachan, '<STR_LIT>': <NUM_LIT:0>,<EOL>'<STR_LIT:i>': i}) for i, stachan in enumerate(input_stachans)]<EOL>pool.close()<EOL>try:<EOL><INDENT>processed_stream = [p.get() for p in results]<EOL><DEDENT>except KeyboardInterrupt as e:  <EOL><INDENT>pool.terminate()<EOL>raise e<EOL><DEDENT>pool.join()<EOL>processed_stream.sort(key=lambda tup: tup[<NUM_LIT:0>])<EOL>processed_stream = Stream([p[<NUM_LIT:1>] for p in processed_stream])<EOL>processed_streams.append(processed_stream)<EOL><DEDENT>if no_missed and multiplex:<EOL><INDENT>for tr in processed_stream:<EOL><INDENT>if np.count_nonzero(tr.data) == <NUM_LIT:0>:<EOL><INDENT>processed_streams.remove(processed_stream)<EOL>print('<STR_LIT>')<EOL>break<EOL><DEDENT><DEDENT><DEDENT><DEDENT>if align:<EOL><INDENT>processed_streams = align_design(<EOL>design_set=processed_streams, shift_len=shift_len, reject=reject,<EOL>multiplex=multiplex, plot=plot, no_missed=no_missed)<EOL><DEDENT>output_streams = []<EOL>for processed_stream in processed_streams:<EOL><INDENT>if len(processed_stream) == <NUM_LIT:0>:<EOL><INDENT>continue<EOL><DEDENT>_st = Stream()<EOL>for stachan in input_stachans:<EOL><INDENT>tr = processed_stream.select(<EOL>station=stachan[<NUM_LIT:0>], channel=stachan[<NUM_LIT:1>])<EOL>if len(tr) >= <NUM_LIT:1>:<EOL><INDENT>_st += tr[<NUM_LIT:0>]<EOL><DEDENT>elif multiplex and len(tr) == <NUM_LIT:0>:<EOL><INDENT>raise IndexError(<EOL>'<STR_LIT>' % (stachan[<NUM_LIT:0>], stachan[<NUM_LIT:1>]))<EOL><DEDENT><DEDENT>if multiplex:<EOL><INDENT>st = multi(stream=_st)<EOL>st = Stream(Trace(st))<EOL>st[<NUM_LIT:0>].stats.station = '<STR_LIT>'<EOL>st[<NUM_LIT:0>].stats.sampling_rate = sampling_rate<EOL><DEDENT>else:<EOL><INDENT>st = _st<EOL><DEDENT>for tr in st:<EOL><INDENT>norm = np.linalg.norm(tr.data)<EOL>if not norm == <NUM_LIT:0>:<EOL><INDENT>tr.data /= norm<EOL><DEDENT><DEDENT>output_streams.append(st)<EOL><DEDENT>return output_streams, input_stachans<EOL>
Process stream data, internal function.

:type streams: list
:param streams: List of obspy.core.stream.Stream to be used to \
    generate the subspace detector.  These should be pre-clustered \
    and aligned.
:type lowcut: float
:param lowcut: Lowcut in Hz, can be None to not apply filter
:type highcut: float
:param highcut: Highcut in Hz, can be None to not apply filter
:type filt_order: int
:param filt_order: Number of corners for filter.
:type sampling_rate: float
:param sampling_rate: Desired sampling rate in Hz
:type multiplex: bool
:param multiplex: Whether to multiplex the data or not.  Data are \
    multiplexed according to the method of Harris, see the multi \
    function for details.
:type stachans: list of tuple
:param stachans: list of tuples of (station, channel) to use.
:type align: bool
:param align: Whether to align the data or not - needs to be done \
    at some point
:type shift_len: float
:param shift_len: Maximum shift allowed for alignment in seconds.
:type reject: float
:param reject: Minimum correlation for traces, only used if align=True.
:type no_missed: bool
:param: no_missed: Reject streams with missed traces, defaults to True. \
    A missing trace from lots of events will reduce the quality of the \
    subspace detector if multiplexed.  Only used when multi is set to True.
:type plot: bool
:param plot: Passed down to align traces - used to check alignment process.

:return: Processed streams
:rtype: list
:return: Station, channel pairs in order
:rtype: list of tuple
:return: List of delays
:rtype: list