<reponame>earthinversion/Fnet_IRIS_data_automated_download
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
Functions for testing the obspy.io.nordic functions
"""
from __future__ import (absolute_import, division, print_function,
                        unicode_literals)
from future.builtins import *  # NOQA @UnusedWildImport

import inspect
import io
import os
import unittest
import warnings

from obspy import read_events, Catalog, UTCDateTime, read
from obspy.core.event import Pick, WaveformStreamID, Arrival, Amplitude
from obspy.core.event import Event, Origin, Magnitude, OriginQuality
from obspy.core.event import EventDescription, CreationInfo
from obspy.core.util.base import NamedTemporaryFile
from obspy.core.util.misc import TemporaryWorkingDirectory
from obspy.io.nordic.core import _is_sfile, read_spectral_info, read_nordic
from obspy.io.nordic.core import readwavename, blanksfile, _write_nordic
from obspy.io.nordic.core import nordpick, readheader
from obspy.io.nordic.core import _int_conv, _readheader, _evmagtonor
from obspy.io.nordic.core import write_select, NordicParsingError
from obspy.io.nordic.core import _float_conv, _nortoevmag, _str_conv


class TestNordicMethods(unittest.TestCase):
    """
    Test suite for nordic io operations.
    """
    def setUp(self):
        self.path = os.path.dirname(os.path.abspath(inspect.getfile(
            inspect.currentframe())))
        self.testing_path = os.path.join(self.path, "data")

    def test_read_write(self):
        """
        Function to test the read and write capabilities of sfile_util.
        """
        # Set-up a test event
        test_event = full_test_event()
        # Add the event to a catalogue which can be used for QuakeML testing
        test_cat = Catalog()
        test_cat += test_event
        # Check the read-write s-file functionality
        with TemporaryWorkingDirectory():
            sfile = _write_nordic(test_cat[0], filename=None, userid='TEST',
                                  evtype='L', outdir='.', wavefiles='test',
                                  explosion=True, overwrite=True)
            self.assertEqual(readwavename(sfile), ['test'])
            read_cat = Catalog()
            # raises "UserWarning: AIN in header, currently unsupported"
            with warnings.catch_warnings():
                warnings.simplefilter('ignore', UserWarning)
                read_cat += read_nordic(sfile)
        read_ev = read_cat[0]
        test_ev = test_cat[0]
        for read_pick, test_pick in zip(read_ev.picks, test_ev.picks):
            self.assertEqual(read_pick.time, test_pick.time)
            self.assertEqual(read_pick.backazimuth, test_pick.backazimuth)
            self.assertEqual(read_pick.onset, test_pick.onset)
            self.assertEqual(read_pick.phase_hint, test_pick.phase_hint)
            self.assertEqual(read_pick.polarity, test_pick.polarity)
            self.assertEqual(read_pick.waveform_id.station_code,
                             test_pick.waveform_id.station_code)
            self.assertEqual(read_pick.waveform_id.channel_code[-1],
                             test_pick.waveform_id.channel_code[-1])
        # assert read_ev.origins[0].resource_id ==\
        #     test_ev.origins[0].resource_id
        self.assertEqual(read_ev.origins[0].time,
                         test_ev.origins[0].time)
        # Note that time_residual_RMS is not a quakeML format
        self.assertEqual(read_ev.origins[0].longitude,
                         test_ev.origins[0].longitude)
        self.assertEqual(read_ev.origins[0].latitude,
                         test_ev.origins[0].latitude)
        self.assertEqual(read_ev.origins[0].depth,
                         test_ev.origins[0].depth)
        self.assertEqual(read_ev.magnitudes[0].mag,
                         test_ev.magnitudes[0].mag)
        self.assertEqual(read_ev.magnitudes[1].mag,
                         test_ev.magnitudes[1].mag)
        self.assertEqual(read_ev.magnitudes[2].mag,
                         test_ev.magnitudes[2].mag)
        self.assertEqual(read_ev.magnitudes[0].creation_info,
                         test_ev.magnitudes[0].creation_info)
        self.assertEqual(read_ev.magnitudes[1].creation_info,
                         test_ev.magnitudes[1].creation_info)
        self.assertEqual(read_ev.magnitudes[2].creation_info,
                         test_ev.magnitudes[2].creation_info)
        self.assertEqual(read_ev.magnitudes[0].magnitude_type,
                         test_ev.magnitudes[0].magnitude_type)
        self.assertEqual(read_ev.magnitudes[1].magnitude_type,
                         test_ev.magnitudes[1].magnitude_type)
        self.assertEqual(read_ev.magnitudes[2].magnitude_type,
                         test_ev.magnitudes[2].magnitude_type)
        self.assertEqual(read_ev.event_descriptions,
                         test_ev.event_descriptions)
        # assert read_ev.amplitudes[0].resource_id ==\
        #     test_ev.amplitudes[0].resource_id
        self.assertEqual(read_ev.amplitudes[0].period,
                         test_ev.amplitudes[0].period)
        self.assertEqual(read_ev.amplitudes[0].snr,
                         test_ev.amplitudes[0].snr)
        self.assertEqual(read_ev.amplitudes[2].period,
                         test_ev.amplitudes[2].period)
        self.assertEqual(read_ev.amplitudes[2].snr,
                         test_ev.amplitudes[2].snr)
        # Check coda magnitude pick
        # Resource ids get overwritten because you can't have two the same in
        # memory
        # self.assertEqual(read_ev.amplitudes[1].resource_id,
        #                  test_ev.amplitudes[1].resource_id)
        self.assertEqual(read_ev.amplitudes[1].type,
                         test_ev.amplitudes[1].type)
        self.assertEqual(read_ev.amplitudes[1].unit,
                         test_ev.amplitudes[1].unit)
        self.assertEqual(read_ev.amplitudes[1].generic_amplitude,
                         test_ev.amplitudes[1].generic_amplitude)
        # Resource ids get overwritten because you can't have two the same in
        # memory
        # self.assertEqual(read_ev.amplitudes[1].pick_id,
        #                  test_ev.amplitudes[1].pick_id)
        self.assertEqual(read_ev.amplitudes[1].waveform_id.station_code,
                         test_ev.amplitudes[1].waveform_id.station_code)
        self.assertEqual(read_ev.amplitudes[1].waveform_id.channel_code,
                         test_ev.amplitudes[1].
                         waveform_id.channel_code[0] +
                         test_ev.amplitudes[1].
                         waveform_id.channel_code[-1])
        self.assertEqual(read_ev.amplitudes[1].magnitude_hint,
                         test_ev.amplitudes[1].magnitude_hint)
        # snr is not supported in s-file
        # self.assertEqual(read_ev.amplitudes[1].snr,
        #                  test_ev.amplitudes[1].snr)
        self.assertEqual(read_ev.amplitudes[1].category,
                         test_ev.amplitudes[1].category)

    def test_fail_writing(self):
        """
        Test a deliberate fail.
        """
        test_event = full_test_event()
        # Add the event to a catalogue which can be used for QuakeML testing
        test_cat = Catalog()
        test_cat += test_event
        test_ev = test_cat[0]
        test_cat.append(full_test_event())
        with self.assertRaises(NordicParsingError):
            # Raises error due to multiple events in catalog
            _write_nordic(test_cat, filename=None, userid='TEST',
                          evtype='L', outdir='.',
                          wavefiles='test', explosion=True,
                          overwrite=True)
        with self.assertRaises(NordicParsingError):
            # Raises error due to too long userid
            _write_nordic(test_ev, filename=None, userid='TESTICLE',
                          evtype='L', outdir='.',
                          wavefiles='test', explosion=True,
                          overwrite=True)
        with self.assertRaises(NordicParsingError):
            # Raises error due to unrecognised event type
            _write_nordic(test_ev, filename=None, userid='TEST',
                          evtype='U', outdir='.',
                          wavefiles='test', explosion=True,
                          overwrite=True)
        with self.assertRaises(NordicParsingError):
            # Raises error due to no output directory
            _write_nordic(test_ev, filename=None, userid='TEST',
                          evtype='L', outdir='albatross',
                          wavefiles='test', explosion=True,
                          overwrite=True)
        invalid_origin = test_ev.copy()

        invalid_origin.origins = []
        with self.assertRaises(NordicParsingError):
            _write_nordic(invalid_origin, filename=None, userid='TEST',
                          evtype='L', outdir='.',
                          wavefiles='test', explosion=True,
                          overwrite=True)
        invalid_origin = test_ev.copy()
        invalid_origin.origins[0].time = None
        with self.assertRaises(NordicParsingError):
            _write_nordic(invalid_origin, filename=None, userid='TEST',
                          evtype='L', outdir='.',
                          wavefiles='test', explosion=True,
                          overwrite=True)
        # Write a near empty origin
        valid_origin = test_ev.copy()
        valid_origin.origins[0].latitude = None
        valid_origin.origins[0].longitude = None
        valid_origin.origins[0].depth = None
        with NamedTemporaryFile() as tf:
            _write_nordic(valid_origin, filename=tf.name, userid='TEST',
                          evtype='L', outdir='.', wavefiles='test',
                          explosion=True, overwrite=True)
            self.assertTrue(os.path.isfile(tf.name))

    def test_blanksfile(self):
        st = read()
        with TemporaryWorkingDirectory():
            testing_path = 'Temporary_wavefile'
            st.write(testing_path, format='MSEED')
            sfile = blanksfile(testing_path, 'L', 'TEST', overwrite=True)
            self.assertTrue(os.path.isfile(sfile))
            os.remove(sfile)
            sfile = blanksfile(testing_path, 'L', 'TEST', overwrite=True,
                               evtime=UTCDateTime())
            self.assertTrue(os.path.isfile(sfile))
            os.remove(sfile)
            with self.assertRaises(NordicParsingError):
                # No wavefile
                blanksfile('albert', 'L', 'TEST', overwrite=True)
            with self.assertRaises(NordicParsingError):
                # USER ID too long
                blanksfile(testing_path, 'L', 'TESTICLE', overwrite=True)
            with self.assertRaises(NordicParsingError):
                # Unknown event type
                blanksfile(testing_path, 'U', 'TEST', overwrite=True)
            # Check that it breaks when writing multiple versions
            sfiles = []
            for _i in range(10):
                # raises UserWarning: Desired sfile exists, will not overwrite
                with warnings.catch_warnings():
                    warnings.simplefilter('ignore', UserWarning)
                    sfiles.append(blanksfile(testing_path, 'L', 'TEST'))
            with self.assertRaises(NordicParsingError):
                # raises UserWarning: Desired sfile exists, will not overwrite
                with warnings.catch_warnings():
                    warnings.simplefilter('ignore', UserWarning)
                    blanksfile(testing_path, 'L', 'TEST')
            for sfile in sfiles:
                self.assertTrue(os.path.isfile(sfile))

    def test_write_empty(self):
        """
        Function to check that writing a blank event works as it should.
        """
        test_event = Event()
        with self.assertRaises(NordicParsingError):
            _write_nordic(test_event, filename=None, userid='TEST', evtype='L',
                          outdir='.', wavefiles='test')
        test_event.origins.append(Origin())
        with self.assertRaises(NordicParsingError):
            _write_nordic(test_event, filename=None, userid='TEST', evtype='L',
                          outdir='.', wavefiles='test')
        test_event.origins[0].time = UTCDateTime()
        with TemporaryWorkingDirectory():
            test_sfile = _write_nordic(test_event, filename=None,
                                       userid='TEST', evtype='L', outdir='.',
                                       wavefiles='test')
            self.assertTrue(os.path.isfile(test_sfile))

    def test_read_empty_header(self):
        """
        Function to check a known issue, empty header info S-file: Bug found \
        by <NAME>.
        """
        # raises "UserWarning: AIN in header, currently unsupported"
        with warnings.catch_warnings():
            warnings.simplefilter('ignore', UserWarning)

            test_event = read_nordic(os.path.join(self.testing_path,
                                                  'Sfile_no_location'))[0]
        self.assertFalse(test_event.origins[0].latitude)
        self.assertFalse(test_event.origins[0].longitude)
        self.assertFalse(test_event.origins[0].depth)

    def test_read_extra_header(self):
        testing_path = os.path.join(self.testing_path, 'Sfile_extra_header')
        not_extra_header = os.path.join(self.testing_path,
                                        '01-0411-15L.S201309')
        # raises "UserWarning: AIN in header, currently unsupported"
        with warnings.catch_warnings():
            warnings.simplefilter('ignore', UserWarning)
            test_event = read_nordic(testing_path)[0]
            header_event = read_nordic(not_extra_header)[0]
        self.assertEqual(test_event.origins[0].time,
                         header_event.origins[0].time)
        self.assertEqual(test_event.origins[0].latitude,
                         header_event.origins[0].latitude)
        self.assertEqual(test_event.origins[0].longitude,
                         header_event.origins[0].longitude)
        self.assertEqual(test_event.origins[0].depth,
                         header_event.origins[0].depth)

    def test_header_mapping(self):
        head_1 = readheader(os.path.join(self.testing_path,
                                         '01-0411-15L.S201309'))
        with open(os.path.join(self.testing_path,
                               '01-0411-15L.S201309'), 'r') as f:
            # raises "UserWarning: AIN in header, currently unsupported"
            with warnings.catch_warnings():
                warnings.simplefilter('ignore', UserWarning)
                head_2 = _readheader(f=f)
        _assert_similarity(head_1, head_2)

    def test_missing_header(self):
        # Check that a suitable error is raised
        with self.assertRaises(NordicParsingError):
            readheader(os.path.join(self.testing_path, 'Sfile_no_header'))

    def test_reading_string_io(self):
        filename = os.path.join(self.testing_path, '01-0411-15L.S201309')
        with open(filename, "rt") as fh:
            file_object = io.StringIO(fh.read())

        # raises "UserWarning: AIN in header, currently unsupported"
        with warnings.catch_warnings():
            warnings.simplefilter('ignore', UserWarning)

            cat = read_events(file_object)
            file_object.close()

            ref_cat = read_events(filename)
            _assert_similarity(cat[0], ref_cat[0])

    def test_reading_bytes_io(self):
        filename = os.path.join(self.testing_path, '01-0411-15L.S201309')
        with open(filename, "rb") as fh:
            file_object = io.BytesIO(fh.read())

        # raises "UserWarning: AIN in header, currently unsupported"
        with warnings.catch_warnings():
            warnings.simplefilter('ignore', UserWarning)

            cat = read_events(file_object)
            file_object.close()

            ref_cat = read_events(filename)
            _assert_similarity(cat[0], ref_cat[0])

    def test_corrupt_header(self):
        filename = os.path.join(self.testing_path, '01-0411-15L.S201309')
        f = open(filename, 'r')
        with NamedTemporaryFile(suffix='.sfile') as tmp_file:
            fout = open(tmp_file.name, 'w')
            for line in f:
                fout.write(line[0:78])
            f.close()
            fout.close()
            with self.assertRaises(NordicParsingError):
                readheader(tmp_file.name)

    def test_multi_writing(self):
        event = full_test_event()
        # Try to write the same event multiple times, but not overwrite
        sfiles = []
        with TemporaryWorkingDirectory():
            for _i in range(59):
                sfiles.append(_write_nordic(event=event, filename=None,
                                            overwrite=False))
            with self.assertRaises(NordicParsingError):
                _write_nordic(event=event, filename=None, overwrite=False)

    def test_mag_conv(self):
        """
        Check that we convert magnitudes as we should!
        """
        magnitude_map = [('L', 'ML'),
                         ('B', 'mB'),
                         ('S', 'Ms'),
                         ('W', 'MW'),
                         ('G', 'MbLg'),
                         ('C', 'Mc'),
                         ]
        for magnitude in magnitude_map:
            self.assertEqual(magnitude[0], _evmagtonor(magnitude[1]))
            self.assertEqual(_nortoevmag(magnitude[0]), magnitude[1])

    def test_str_conv(self):
        """
        Test the simple string conversions.
        """
        self.assertEqual(_int_conv('albert'), None)
        self.assertEqual(_float_conv('albert'), None)
        self.assertEqual(_str_conv('albert'), 'albert')
        self.assertEqual(_int_conv('1'), 1)
        self.assertEqual(_float_conv('1'), 1.0)
        self.assertEqual(_str_conv(1), '1')
        self.assertEqual(_int_conv('1.0256'), None)
        self.assertEqual(_float_conv('1.0256'), 1.0256)
        self.assertEqual(_str_conv(1.0256), '1.0256')

    def test_read_wavename(self):
        testing_path = os.path.join(self.testing_path, '01-0411-15L.S201309')
        wavefiles = readwavename(testing_path)
        self.assertEqual(len(wavefiles), 1)
        # Test that full paths are handled
        test_event = full_test_event()
        # Add the event to a catalogue which can be used for QuakeML testing
        test_cat = Catalog()
        test_cat += test_event
        # Check the read-write s-file functionality
        with TemporaryWorkingDirectory():
            sfile = _write_nordic(
                test_cat[0], filename=None, userid='TEST', evtype='L',
                outdir='.', wavefiles=['walrus/test'], explosion=True,
                overwrite=True)
            self.assertEqual(readwavename(sfile), ['test'])
        # Check that multiple wavefiles are read properly
        with TemporaryWorkingDirectory():
            sfile = _write_nordic(
                test_cat[0], filename=None, userid='TEST', evtype='L',
                outdir='.', wavefiles=['walrus/test', 'albert'],
                explosion=True, overwrite=True)
            self.assertEqual(readwavename(sfile), ['test', 'albert'])

    def test_read_event(self):
        """
        Test the wrapper.
        """
        testing_path = os.path.join(self.testing_path, '01-0411-15L.S201309')
        # raises "UserWarning: AIN in header, currently unsupported"
        with warnings.catch_warnings():
            warnings.simplefilter('ignore', UserWarning)
            event = read_nordic(testing_path)[0]
        self.assertEqual(len(event.origins), 1)

    def test_read_many_events(self):
        testing_path = os.path.join(self.testing_path, 'select.out')
        # raises "UserWarning: AIN in header, currently unsupported"
        with warnings.catch_warnings():
            warnings.simplefilter('ignore', UserWarning)
            catalog = read_nordic(testing_path)
        self.assertEqual(len(catalog), 50)

    def test_write_select(self):
        cat = read_events()
        with NamedTemporaryFile(suffix='.out') as tf:
            # raises "UserWarning: mb is not convertible"
            with warnings.catch_warnings():
                warnings.simplefilter('ignore', UserWarning)
                write_select(cat, filename=tf.name)
            cat_back = read_events(tf.name)
            for event_1, event_2 in zip(cat, cat_back):
                _assert_similarity(event_1=event_1, event_2=event_2)

    def test_write_plugin(self):
        cat = read_events()
        cat.append(full_test_event())
        with NamedTemporaryFile(suffix='.out') as tf:
            # raises UserWarning: mb is not convertible
            with warnings.catch_warnings():
                warnings.simplefilter('ignore', UserWarning)
                cat.write(tf.name, format='nordic')
            # raises "UserWarning: AIN in header, currently unsupported"
            with warnings.catch_warnings():
                warnings.simplefilter('ignore', UserWarning)
                cat_back = read_events(tf.name)
            for event_1, event_2 in zip(cat, cat_back):
                _assert_similarity(event_1=event_1, event_2=event_2)

    def test_inaccurate_picks(self):
        testing_path = os.path.join(self.testing_path, 'bad_picks.sfile')
        # raises "UserWarning: AIN in header, currently unsupported"
        with warnings.catch_warnings():
            warnings.simplefilter('ignore', UserWarning)
            cat = read_nordic(testing_path)
        pick_string = nordpick(cat[0])
        for pick in pick_string:
            self.assertEqual(len(pick), 80)

    def test_round_len(self):
        testing_path = os.path.join(self.testing_path, 'round_len_undef.sfile')
        # raises "UserWarning: AIN in header, currently unsupported"
        with warnings.catch_warnings():
            warnings.simplefilter('ignore', UserWarning)
            event = read_nordic(testing_path)[0]
        pick_string = nordpick(event)
        for pick in pick_string:
            self.assertEqual(len(pick), 80)

    def test_read_moment(self):
        """
        Test the reading of seismic moment from the s-file.
        """
        testing_path = os.path.join(self.testing_path, 'automag.out')
        # raises "UserWarning: AIN in header, currently unsupported"
        with warnings.catch_warnings():
            warnings.simplefilter('ignore', UserWarning)
            event = read_nordic(testing_path)[0]
        mag = [m for m in event.magnitudes if m.magnitude_type == 'MW']
        self.assertEqual(len(mag), 1)
        self.assertEqual(mag[0].mag, 0.7)

    def test_read_moment_info(self):
        """
        Test reading the info from spectral analysis.
        """
        testing_path = os.path.join(self.testing_path, 'automag.out')
        spec_inf = read_spectral_info(testing_path)
        self.assertEqual(len(spec_inf), 5)
        # This should actually test that what we are reading in is correct.
        average = spec_inf[('AVERAGE', '')]
        check_av = {u'channel': '', u'corner_freq': 5.97, u'decay': 0.0,
                    u'moment': 12589254117.941662, u'moment_mag': 0.7,
                    u'source_radius': 0.231,
                    u'spectral_level': 0.3981071705534972,
                    u'station': 'AVERAGE', u'stress_drop': 0.006,
                    u'window_length': 1.6}
        for key in average.keys():
            if isinstance(average.get(key), str):
                self.assertEqual(average.get(key), check_av.get(key))
            else:
                self.assertEqual(round(average.get(key), 4),
                                 round(check_av.get(key), 4))

    def test_is_sfile(self):
        sfiles = ['01-0411-15L.S201309', 'automag.out', 'bad_picks.sfile',
                  'round_len_undef.sfile', 'Sfile_extra_header',
                  'Sfile_no_location']
        for sfile in sfiles:
            self.assertTrue(_is_sfile(os.path.join(self.testing_path, sfile)))
        self.assertFalse(_is_sfile(os.path.join(self.testing_path,
                                                'Sfile_no_header')))
        self.assertFalse(_is_sfile(os.path.join(self.path, '..', '..',
                                                'nlloc', 'tests', 'data',
                                                'nlloc.hyp')))

    def test_read_picks_across_day_end(self):
        testing_path = os.path.join(self.testing_path, 'sfile_over_day')
        # raises "UserWarning: AIN in header, currently unsupported"
        with warnings.catch_warnings():
            warnings.simplefilter('ignore', UserWarning)
            event = read_nordic(testing_path)[0]
        pick_times = [pick.time for pick in event.picks]
        for pick in event.picks:
            # Pick should come after origin time
            self.assertGreater(pick.time, event.origins[0].time)
            # All picks in this event are within 60s of origin time
            self.assertLessEqual((pick.time - event.origins[0].time), 60)
        # Make sure zero hours and 24 hour picks are handled the same.
        testing_path = os.path.join(self.testing_path, 'sfile_over_day_zeros')
        # raises "UserWarning: AIN in header, currently unsupported"
        with warnings.catch_warnings():
            warnings.simplefilter('ignore', UserWarning)
            event_2 = read_nordic(testing_path)[0]
        for pick in event_2.picks:
            # Pick should come after origin time
            self.assertGreater(pick.time, event_2.origins[0].time)
            # All picks in this event are within 60s of origin time
            self.assertLessEqual((pick.time - event_2.origins[0].time), 60)
            # Each pick should be the same as one pick in the previous event
            self.assertTrue(pick.time in pick_times)
        self.assertEqual(event_2.origins[0].time, event.origins[0].time)

    def test_distance_conversion(self):
        """
        Check that distances are converted properly.
        """
        testing_path = os.path.join(self.testing_path, '01-0411-15L.S201309')
        # raises "UserWarning: AIN in header, currently unsupported"
        with warnings.catch_warnings():
            warnings.simplefilter('ignore', UserWarning)
            cat = read_events(testing_path)
        event = cat[0]
        self.assertAlmostEqual(
            sorted(event.origins[0].arrivals,
                   key=lambda x: x.distance)[0].distance, 0.035972864236749225)
        pick_strings = nordpick(event)
        self.assertEqual(
            int([p for p in pick_strings if p.split()[0] == 'GCSZ' and
                 p.split()[1] == 'SZ'][0].split()[-1]), 304)
        self.assertEqual(
            int([p for p in pick_strings if p.split()[0] == 'WZ11' and
                 p.split()[1] == 'HZ'][0].split()[-1]), 30)

    def test_large_negative_longitude(self):
        event = full_test_event()
        event.origins[0].longitude = -120
        with NamedTemporaryFile(suffix=".out") as tf:
            event.write(tf.name, format="NORDIC")
            event_back = read_events(tf.name)
            _assert_similarity(event, event_back[0])


def _assert_similarity(event_1, event_2, verbose=False):
    """
    Raise AssertionError if testing similarity fails
    """
    if not _test_similarity(event_1, event_2, verbose=verbose):
        raise AssertionError('Events failed similarity check')


def _test_similarity(event_1, event_2, verbose=False):
    """
    Check the similarity of the components of obspy events, discounting
    resource IDs, which are not maintained in nordic files.

    Raise AssertionError if test fails

    :type event_1: obspy.core.event.Event
    :param event_1: First event
    :type event_2: obspy.core.event.Event
    :param event_2: Comparison event
    :type verbose: bool
    :param verbose: If true and fails will output why it fails.
    """
    # Check origins
    if len(event_1.origins) != len(event_2.origins):
        return False
    for ori_1, ori_2 in zip(event_1.origins, event_2.origins):
        for key in ori_1.keys():
            if key not in ["resource_id", "comments", "arrivals",
                           "method_id", "origin_uncertainty", "depth_type",
                           "quality", "creation_info", "evaluation_mode",
                           "depth_errors", "time_errors"]:
                if ori_1[key] != ori_2[key]:
                    if verbose:
                        print('%s is not the same as %s for key %s' %
                              (ori_1[key], ori_2[key], key))
                    return False
            elif key == "arrivals":
                if len(ori_1[key]) != len(ori_2[key]):
                    print('%i is not the same as %i for key %s' %
                          (len(ori_1[key]), len(ori_2[key]), key))
                    return False
                for arr_1, arr_2 in zip(ori_1[key], ori_2[key]):
                    for arr_key in arr_1.keys():
                        if arr_key not in ["resource_id", "pick_id",
                                           "distance"]:
                            if arr_1[arr_key] != arr_2[arr_key]:
                                if verbose:
                                    print('%s does not match %s for key %s' %
                                          (arr_1[arr_key], arr_2[arr_key],
                                           arr_key))
                                return False
                    if arr_1["distance"] and round(
                            arr_1["distance"]) != round(arr_2["distance"]):
                            if verbose:
                                print('%s does not match %s for key %s' %
                                      (arr_1[arr_key], arr_2[arr_key],
                                       arr_key))
                            return False
    # Check picks
    if len(event_1.picks) != len(event_2.picks):
        if verbose:
            print('Number of picks is not equal')
        return False
    for pick_1, pick_2 in zip(event_1.picks, event_2.picks):
        # Assuming same ordering of picks...
        for key in pick_1.keys():
            if key not in ["resource_id", "waveform_id"]:
                if pick_1[key] != pick_2[key]:
                    if verbose:
                        print('%s is not the same as %s for key %s' %
                              (pick_1[key], pick_2[key], key))
                    return False
            elif key == "waveform_id":
                if pick_1[key].station_code != pick_2[key].station_code:
                    if verbose:
                        print('Station codes do not match')
                    return False
                if pick_1[key].channel_code[0] != pick_2[key].channel_code[0]:
                    if verbose:
                        print('Channel codes do not match')
                    return False
                if pick_1[key].channel_code[-1] !=\
                   pick_2[key].channel_code[-1]:
                    if verbose:
                        print('Channel codes do not match')
                    return False
    # Check amplitudes
    if not len(event_1.amplitudes) == len(event_2.amplitudes):
        if verbose:
            print('Not the same number of amplitudes')
        return False
    for amp_1, amp_2 in zip(event_1.amplitudes, event_2.amplitudes):
        # Assuming same ordering of amplitudes
        for key in amp_1.keys():
            if key not in ["resource_id", "pick_id", "waveform_id", "snr",
                           "magnitude_hint", 'type']:
                if not amp_1[key] == amp_2[key]:
                    if verbose:
                        print('%s is not the same as %s for key %s' %
                              (amp_1[key], amp_2[key], key))
                        print(amp_1)
                        print(amp_2)
                    return False
            elif key == "waveform_id":
                if pick_1[key].station_code != pick_2[key].station_code:
                    if verbose:
                        print('Station codes do not match')
                    return False
                if pick_1[key].channel_code[0] != pick_2[key].channel_code[0]:
                    if verbose:
                        print('Channel codes do not match')
                    return False
                if pick_1[key].channel_code[-1] !=\
                   pick_2[key].channel_code[-1]:
                    if verbose:
                        print('Channel codes do not match')
                    return False
            elif key in ["magnitude_hint", "type"]:
                # Reading back in will define both, but input event might have
                # None
                if amp_1[key] is not None:
                    if not amp_1[key] == amp_2[key]:
                        if verbose:
                            print('%s is not the same as %s for key %s' %
                                  (amp_1[key], amp_2[key], key))
                        return False
    return True


def full_test_event():
    """
    Function to generate a basic, full test event
    """
    test_event = Event()
    test_event.origins.append(Origin())
    test_event.origins[0].time = UTCDateTime("2012-03-26") + 1.2
    test_event.event_descriptions.append(EventDescription())
    test_event.event_descriptions[0].text = 'LE'
    test_event.origins[0].latitude = 45.0
    test_event.origins[0].longitude = 25.0
    test_event.origins[0].depth = 15000
    test_event.creation_info = CreationInfo(agency_id='TES')
    test_event.origins[0].quality = OriginQuality(standard_error=0.01)
    test_event.magnitudes.append(Magnitude())
    test_event.magnitudes[0].mag = 0.1
    test_event.magnitudes[0].magnitude_type = 'ML'
    test_event.magnitudes[0].creation_info = CreationInfo('TES')
    test_event.magnitudes[0].origin_id = test_event.origins[0].resource_id
    test_event.magnitudes.append(Magnitude())
    test_event.magnitudes[1].mag = 0.5
    test_event.magnitudes[1].magnitude_type = 'Mc'
    test_event.magnitudes[1].creation_info = CreationInfo('TES')
    test_event.magnitudes[1].origin_id = test_event.origins[0].resource_id
    test_event.magnitudes.append(Magnitude())
    test_event.magnitudes[2].mag = 1.3
    test_event.magnitudes[2].magnitude_type = 'Ms'
    test_event.magnitudes[2].creation_info = CreationInfo('TES')
    test_event.magnitudes[2].origin_id = test_event.origins[0].resource_id

    # Define the test pick
    _waveform_id_1 = WaveformStreamID(station_code='FOZ', channel_code='SHZ',
                                      network_code='NZ')
    _waveform_id_2 = WaveformStreamID(station_code='WTSZ', channel_code='BH1',
                                      network_code=' ')
    # Pick to associate with amplitude - 0
    test_event.picks = [
        Pick(waveform_id=_waveform_id_1, phase_hint='IAML',
             polarity='undecidable', time=UTCDateTime("2012-03-26") + 1.68,
             evaluation_mode="manual"),
        Pick(waveform_id=_waveform_id_1, onset='impulsive', phase_hint='PN',
             polarity='positive', time=UTCDateTime("2012-03-26") + 1.68,
             evaluation_mode="manual"),
        Pick(waveform_id=_waveform_id_1, phase_hint='IAML',
             polarity='undecidable', time=UTCDateTime("2012-03-26") + 1.68,
             evaluation_mode="manual"),
        Pick(waveform_id=_waveform_id_2, onset='impulsive', phase_hint='SG',
             polarity='undecidable', time=UTCDateTime("2012-03-26") + 1.72,
             evaluation_mode="manual"),
        Pick(waveform_id=_waveform_id_2, onset='impulsive', phase_hint='PN',
             polarity='undecidable', time=UTCDateTime("2012-03-26") + 1.62,
             evaluation_mode="automatic")]
    # Test a generic local magnitude amplitude pick
    test_event.amplitudes = [
        Amplitude(generic_amplitude=2.0, period=0.4,
                  pick_id=test_event.picks[0].resource_id,
                  waveform_id=test_event.picks[0].waveform_id, unit='m',
                  magnitude_hint='ML', category='point', type='AML'),
        Amplitude(generic_amplitude=10,
                  pick_id=test_event.picks[1].resource_id,
                  waveform_id=test_event.picks[1].waveform_id, type='END',
                  category='duration', unit='s', magnitude_hint='Mc',
                  snr=2.3),
        Amplitude(generic_amplitude=5.0, period=0.6,
                  pick_id=test_event.picks[2].resource_id,
                  waveform_id=test_event.picks[0].waveform_id, unit='m',
                  category='point', type='AML')]
    test_event.origins[0].arrivals = [
        Arrival(time_weight=0, phase=test_event.picks[1].phase_hint,
                pick_id=test_event.picks[1].resource_id),
        Arrival(time_weight=2, phase=test_event.picks[3].phase_hint,
                pick_id=test_event.picks[3].resource_id,
                backazimuth_residual=5, time_residual=0.2, distance=15,
                azimuth=25),
        Arrival(time_weight=2, phase=test_event.picks[4].phase_hint,
                pick_id=test_event.picks[4].resource_id,
                backazimuth_residual=5, time_residual=0.2, distance=15,
                azimuth=25)]
    return test_event


def suite():
    return unittest.makeSuite(TestNordicMethods, 'test')


if __name__ == '__main__':
    unittest.main(defaultTest='suite')

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
Functions for testing the obspy.io.nordic functions
"""
from __future__ import (absolute_import, division, print_function,
                        unicode_literals)
from future.builtins import *  # NOQA @UnusedWildImport

import inspect
import io
import os
import unittest
import warnings

from obspy import read_events, Catalog, UTCDateTime, read
from obspy.core.event import Pick, WaveformStreamID, Arrival, Amplitude
from obspy.core.event import Event, Origin, Magnitude, OriginQuality
from obspy.core.event import EventDescription, CreationInfo
from obspy.core.util.base import NamedTemporaryFile
from obspy.core.util.misc import TemporaryWorkingDirectory
from obspy.io.nordic.core import _is_sfile, read_spectral_info, read_nordic
from obspy.io.nordic.core import readwavename, blanksfile, _write_nordic
from obspy.io.nordic.core import nordpick, readheader
from obspy.io.nordic.core import _int_conv, _readheader, _evmagtonor
from obspy.io.nordic.core import write_select, NordicParsingError
from obspy.io.nordic.core import _float_conv, _nortoevmag, _str_conv


class TestNordicMethods(unittest.TestCase):
    """
    Test suite for nordic io operations.
    """
    def setUp(self):
        self.path = os.path.dirname(os.path.abspath(inspect.getfile(
            inspect.currentframe())))
        self.testing_path = os.path.join(self.path, "data")

    def test_read_write(self):
        """
        Function to test the read and write capabilities of sfile_util.
        """
        # Set-up a test event
        test_event = full_test_event()
        # Add the event to a catalogue which can be used for QuakeML testing
        test_cat = Catalog()
        test_cat += test_event
        # Check the read-write s-file functionality
        with TemporaryWorkingDirectory():
            sfile = _write_nordic(test_cat[0], filename=None, userid='TEST',
                                  evtype='L', outdir='.', wavefiles='test',
                                  explosion=True, overwrite=True)
            self.assertEqual(readwavename(sfile), ['test'])
            read_cat = Catalog()
            # raises "UserWarning: AIN in header, currently unsupported"
            with warnings.catch_warnings():
                warnings.simplefilter('ignore', UserWarning)
                read_cat += read_nordic(sfile)
        read_ev = read_cat[0]
        test_ev = test_cat[0]
        for read_pick, test_pick in zip(read_ev.picks, test_ev.picks):
            self.assertEqual(read_pick.time, test_pick.time)
            self.assertEqual(read_pick.backazimuth, test_pick.backazimuth)
            self.assertEqual(read_pick.onset, test_pick.onset)
            self.assertEqual(read_pick.phase_hint, test_pick.phase_hint)
            self.assertEqual(read_pick.polarity, test_pick.polarity)
            self.assertEqual(read_pick.waveform_id.station_code,
                             test_pick.waveform_id.station_code)
            self.assertEqual(read_pick.waveform_id.channel_code[-1],
                             test_pick.waveform_id.channel_code[-1])
        # assert read_ev.origins[0].resource_id ==\
        #     test_ev.origins[0].resource_id
        self.assertEqual(read_ev.origins[0].time,
                         test_ev.origins[0].time)
        # Note that time_residual_RMS is not a quakeML format
        self.assertEqual(read_ev.origins[0].longitude,
                         test_ev.origins[0].longitude)
        self.assertEqual(read_ev.origins[0].latitude,
                         test_ev.origins[0].latitude)
        self.assertEqual(read_ev.origins[0].depth,
                         test_ev.origins[0].depth)
        self.assertEqual(read_ev.magnitudes[0].mag,
                         test_ev.magnitudes[0].mag)
        self.assertEqual(read_ev.magnitudes[1].mag,
                         test_ev.magnitudes[1].mag)
        self.assertEqual(read_ev.magnitudes[2].mag,
                         test_ev.magnitudes[2].mag)
        self.assertEqual(read_ev.magnitudes[0].creation_info,
                         test_ev.magnitudes[0].creation_info)
        self.assertEqual(read_ev.magnitudes[1].creation_info,
                         test_ev.magnitudes[1].creation_info)
        self.assertEqual(read_ev.magnitudes[2].creation_info,
                         test_ev.magnitudes[2].creation_info)
        self.assertEqual(read_ev.magnitudes[0].magnitude_type,
                         test_ev.magnitudes[0].magnitude_type)
        self.assertEqual(read_ev.magnitudes[1].magnitude_type,
                         test_ev.magnitudes[1].magnitude_type)
        self.assertEqual(read_ev.magnitudes[2].magnitude_type,
                         test_ev.magnitudes[2].magnitude_type)
        self.assertEqual(read_ev.event_descriptions,
                         test_ev.event_descriptions)
        # assert read_ev.amplitudes[0].resource_id ==\
        #     test_ev.amplitudes[0].resource_id
        self.assertEqual(read_ev.amplitudes[0].period,
                         test_ev.amplitudes[0].period)
        self.assertEqual(read_ev.amplitudes[0].snr,
                         test_ev.amplitudes[0].snr)
        self.assertEqual(read_ev.amplitudes[2].period,
                         test_ev.amplitudes[2].period)
        self.assertEqual(read_ev.amplitudes[2].snr,
                         test_ev.amplitudes[2].snr)
        # Check coda magnitude pick
        # Resource ids get overwritten because you can't have two the same in
        # memory
        # self.assertEqual(read_ev.amplitudes[1].resource_id,
        #                  test_ev.amplitudes[1].resource_id)
        self.assertEqual(read_ev.amplitudes[1].type,
                         test_ev.amplitudes[1].type)
        self.assertEqual(read_ev.amplitudes[1].unit,
                         test_ev.amplitudes[1].unit)
        self.assertEqual(read_ev.amplitudes[1].generic_amplitude,
                         test_ev.amplitudes[1].generic_amplitude)
        # Resource ids get overwritten because you can't have two the same in
        # memory
        # self.assertEqual(read_ev.amplitudes[1].pick_id,
        #                  test_ev.amplitudes[1].pick_id)
        self.assertEqual(read_ev.amplitudes[1].waveform_id.station_code,
                         test_ev.amplitudes[1].waveform_id.station_code)
        self.assertEqual(read_ev.amplitudes[1].waveform_id.channel_code,
                         test_ev.amplitudes[1].
                         waveform_id.channel_code[0] +
                         test_ev.amplitudes[1].
                         waveform_id.channel_code[-1])
        self.assertEqual(read_ev.amplitudes[1].magnitude_hint,
                         test_ev.amplitudes[1].magnitude_hint)
        # snr is not supported in s-file
        # self.assertEqual(read_ev.amplitudes[1].snr,
        #                  test_ev.amplitudes[1].snr)
        self.assertEqual(read_ev.amplitudes[1].category,
                         test_ev.amplitudes[1].category)

    def test_fail_writing(self):
        """
        Test a deliberate fail.
        """
        test_event = full_test_event()
        # Add the event to a catalogue which can be used for QuakeML testing
        test_cat = Catalog()
        test_cat += test_event
        test_ev = test_cat[0]
        test_cat.append(full_test_event())
        with self.assertRaises(NordicParsingError):
            # Raises error due to multiple events in catalog
            _write_nordic(test_cat, filename=None, userid='TEST',
                          evtype='L', outdir='.',
                          wavefiles='test', explosion=True,
                          overwrite=True)
        with self.assertRaises(NordicParsingError):
            # Raises error due to too long userid
            _write_nordic(test_ev, filename=None, userid='TESTICLE',
                          evtype='L', outdir='.',
                          wavefiles='test', explosion=True,
                          overwrite=True)
        with self.assertRaises(NordicParsingError):
            # Raises error due to unrecognised event type
            _write_nordic(test_ev, filename=None, userid='TEST',
                          evtype='U', outdir='.',
                          wavefiles='test', explosion=True,
                          overwrite=True)
        with self.assertRaises(NordicParsingError):
            # Raises error due to no output directory
            _write_nordic(test_ev, filename=None, userid='TEST',
                          evtype='L', outdir='albatross',
                          wavefiles='test', explosion=True,
                          overwrite=True)
        invalid_origin = test_ev.copy()

        invalid_origin.origins = []
        with self.assertRaises(NordicParsingError):
            _write_nordic(invalid_origin, filename=None, userid='TEST',
                          evtype='L', outdir='.',
                          wavefiles='test', explosion=True,
                          overwrite=True)
        invalid_origin = test_ev.copy()
        invalid_origin.origins[0].time = None
        with self.assertRaises(NordicParsingError):
            _write_nordic(invalid_origin, filename=None, userid='TEST',
                          evtype='L', outdir='.',
                          wavefiles='test', explosion=True,
                          overwrite=True)
        # Write a near empty origin
        valid_origin = test_ev.copy()
        valid_origin.origins[0].latitude = None
        valid_origin.origins[0].longitude = None
        valid_origin.origins[0].depth = None
        with NamedTemporaryFile() as tf:
            _write_nordic(valid_origin, filename=tf.name, userid='TEST',
                          evtype='L', outdir='.', wavefiles='test',
                          explosion=True, overwrite=True)
            self.assertTrue(os.path.isfile(tf.name))

    def test_blanksfile(self):
        st = read()
        with TemporaryWorkingDirectory():
            testing_path = 'Temporary_wavefile'
            st.write(testing_path, format='MSEED')
            sfile = blanksfile(testing_path, 'L', 'TEST', overwrite=True)
            self.assertTrue(os.path.isfile(sfile))
            os.remove(sfile)
            sfile = blanksfile(testing_path, 'L', 'TEST', overwrite=True,
                               evtime=UTCDateTime())
            self.assertTrue(os.path.isfile(sfile))
            os.remove(sfile)
            with self.assertRaises(NordicParsingError):
                # No wavefile
                blanksfile('albert', 'L', 'TEST', overwrite=True)
            with self.assertRaises(NordicParsingError):
                # USER ID too long
                blanksfile(testing_path, 'L', 'TESTICLE', overwrite=True)
            with self.assertRaises(NordicParsingError):
                # Unknown event type
                blanksfile(testing_path, 'U', 'TEST', overwrite=True)
            # Check that it breaks when writing multiple versions
            sfiles = []
            for _i in range(10):
                # raises UserWarning: Desired sfile exists, will not overwrite
                with warnings.catch_warnings():
                    warnings.simplefilter('ignore', UserWarning)
                    sfiles.append(blanksfile(testing_path, 'L', 'TEST'))
            with self.assertRaises(NordicParsingError):
                # raises UserWarning: Desired sfile exists, will not overwrite
                with warnings.catch_warnings():
                    warnings.simplefilter('ignore', UserWarning)
                    blanksfile(testing_path, 'L', 'TEST')
            for sfile in sfiles:
                self.assertTrue(os.path.isfile(sfile))

    def test_write_empty(self):
        """
        Function to check that writing a blank event works as it should.
        """
        test_event = Event()
        with self.assertRaises(NordicParsingError):
            _write_nordic(test_event, filename=None, userid='TEST', evtype='L',
                          outdir='.', wavefiles='test')
        test_event.origins.append(Origin())
        with self.assertRaises(NordicParsingError):
            _write_nordic(test_event, filename=None, userid='TEST', evtype='L',
                          outdir='.', wavefiles='test')
        test_event.origins[0].time = UTCDateTime()
        with TemporaryWorkingDirectory():
            test_sfile = _write_nordic(test_event, filename=None,
                                       userid='TEST', evtype='L', outdir='.',
                                       wavefiles='test')
            self.assertTrue(os.path.isfile(test_sfile))

    def test_read_empty_header(self):
        """
        Function to check a known issue, empty header info S-file: Bug found \
        by <NAME>.
        """
        # raises "UserWarning: AIN in header, currently unsupported"
        with warnings.catch_warnings():
            warnings.simplefilter('ignore', UserWarning)

            test_event = read_nordic(os.path.join(self.testing_path,
                                                  'Sfile_no_location'))[0]
        self.assertFalse(test_event.origins[0].latitude)
        self.assertFalse(test_event.origins[0].longitude)
        self.assertFalse(test_event.origins[0].depth)

    def test_read_extra_header(self):
        testing_path = os.path.join(self.testing_path, 'Sfile_extra_header')
        not_extra_header = os.path.join(self.testing_path,
                                        '01-0411-15L.S201309')
        # raises "UserWarning: AIN in header, currently unsupported"
        with warnings.catch_warnings():
            warnings.simplefilter('ignore', UserWarning)
            test_event = read_nordic(testing_path)[0]
            header_event = read_nordic(not_extra_header)[0]
        self.assertEqual(test_event.origins[0].time,
                         header_event.origins[0].time)
        self.assertEqual(test_event.origins[0].latitude,
                         header_event.origins[0].latitude)
        self.assertEqual(test_event.origins[0].longitude,
                         header_event.origins[0].longitude)
        self.assertEqual(test_event.origins[0].depth,
                         header_event.origins[0].depth)

    def test_header_mapping(self):
        head_1 = readheader(os.path.join(self.testing_path,
                                         '01-0411-15L.S201309'))
        with open(os.path.join(self.testing_path,
                               '01-0411-15L.S201309'), 'r') as f:
            # raises "UserWarning: AIN in header, currently unsupported"
            with warnings.catch_warnings():
                warnings.simplefilter('ignore', UserWarning)
                head_2 = _readheader(f=f)
        _assert_similarity(head_1, head_2)

    def test_missing_header(self):
        # Check that a suitable error is raised
        with self.assertRaises(NordicParsingError):
            readheader(os.path.join(self.testing_path, 'Sfile_no_header'))

    def test_reading_string_io(self):
        filename = os.path.join(self.testing_path, '01-0411-15L.S201309')
        with open(filename, "rt") as fh:
            file_object = io.StringIO(fh.read())

        # raises "UserWarning: AIN in header, currently unsupported"
        with warnings.catch_warnings():
            warnings.simplefilter('ignore', UserWarning)

            cat = read_events(file_object)
            file_object.close()

            ref_cat = read_events(filename)
            _assert_similarity(cat[0], ref_cat[0])

    def test_reading_bytes_io(self):
        filename = os.path.join(self.testing_path, '01-0411-15L.S201309')
        with open(filename, "rb") as fh:
            file_object = io.BytesIO(fh.read())

        # raises "UserWarning: AIN in header, currently unsupported"
        with warnings.catch_warnings():
            warnings.simplefilter('ignore', UserWarning)

            cat = read_events(file_object)
            file_object.close()

            ref_cat = read_events(filename)
            _assert_similarity(cat[0], ref_cat[0])

    def test_corrupt_header(self):
        filename = os.path.join(self.testing_path, '01-0411-15L.S201309')
        f = open(filename, 'r')
        with NamedTemporaryFile(suffix='.sfile') as tmp_file:
            fout = open(tmp_file.name, 'w')
            for line in f:
                fout.write(line[0:78])
            f.close()
            fout.close()
            with self.assertRaises(NordicParsingError):
                readheader(tmp_file.name)

    def test_multi_writing(self):
        event = full_test_event()
        # Try to write the same event multiple times, but not overwrite
        sfiles = []
        with TemporaryWorkingDirectory():
            for _i in range(59):
                sfiles.append(_write_nordic(event=event, filename=None,
                                            overwrite=False))
            with self.assertRaises(NordicParsingError):
                _write_nordic(event=event, filename=None, overwrite=False)

    def test_mag_conv(self):
        """
        Check that we convert magnitudes as we should!
        """
        magnitude_map = [('L', 'ML'),
                         ('B', 'mB'),
                         ('S', 'Ms'),
                         ('W', 'MW'),
                         ('G', 'MbLg'),
                         ('C', 'Mc'),
                         ]
        for magnitude in magnitude_map:
            self.assertEqual(magnitude[0], _evmagtonor(magnitude[1]))
            self.assertEqual(_nortoevmag(magnitude[0]), magnitude[1])

    def test_str_conv(self):
        """
        Test the simple string conversions.
        """
        self.assertEqual(_int_conv('albert'), None)
        self.assertEqual(_float_conv('albert'), None)
        self.assertEqual(_str_conv('albert'), 'albert')
        self.assertEqual(_int_conv('1'), 1)
        self.assertEqual(_float_conv('1'), 1.0)
        self.assertEqual(_str_conv(1), '1')
        self.assertEqual(_int_conv('1.0256'), None)
        self.assertEqual(_float_conv('1.0256'), 1.0256)
        self.assertEqual(_str_conv(1.0256), '1.0256')

    def test_read_wavename(self):
        testing_path = os.path.join(self.testing_path, '01-0411-15L.S201309')
        wavefiles = readwavename(testing_path)
        self.assertEqual(len(wavefiles), 1)
        # Test that full paths are handled
        test_event = full_test_event()
        # Add the event to a catalogue which can be used for QuakeML testing
        test_cat = Catalog()
        test_cat += test_event
        # Check the read-write s-file functionality
        with TemporaryWorkingDirectory():
            sfile = _write_nordic(
                test_cat[0], filename=None, userid='TEST', evtype='L',
                outdir='.', wavefiles=['walrus/test'], explosion=True,
                overwrite=True)
            self.assertEqual(readwavename(sfile), ['test'])
        # Check that multiple wavefiles are read properly
        with TemporaryWorkingDirectory():
            sfile = _write_nordic(
                test_cat[0], filename=None, userid='TEST', evtype='L',
                outdir='.', wavefiles=['walrus/test', 'albert'],
                explosion=True, overwrite=True)
            self.assertEqual(readwavename(sfile), ['test', 'albert'])

    def test_read_event(self):
        """
        Test the wrapper.
        """
        testing_path = os.path.join(self.testing_path, '01-0411-15L.S201309')
        # raises "UserWarning: AIN in header, currently unsupported"
        with warnings.catch_warnings():
            warnings.simplefilter('ignore', UserWarning)
            event = read_nordic(testing_path)[0]
        self.assertEqual(len(event.origins), 1)

    def test_read_many_events(self):
        testing_path = os.path.join(self.testing_path, 'select.out')
        # raises "UserWarning: AIN in header, currently unsupported"
        with warnings.catch_warnings():
            warnings.simplefilter('ignore', UserWarning)
            catalog = read_nordic(testing_path)
        self.assertEqual(len(catalog), 50)

    def test_write_select(self):
        cat = read_events()
        with NamedTemporaryFile(suffix='.out') as tf:
            # raises "UserWarning: mb is not convertible"
            with warnings.catch_warnings():
                warnings.simplefilter('ignore', UserWarning)
                write_select(cat, filename=tf.name)
            cat_back = read_events(tf.name)
            for event_1, event_2 in zip(cat, cat_back):
                _assert_similarity(event_1=event_1, event_2=event_2)

    def test_write_plugin(self):
        cat = read_events()
        cat.append(full_test_event())
        with NamedTemporaryFile(suffix='.out') as tf:
            # raises UserWarning: mb is not convertible
            with warnings.catch_warnings():
                warnings.simplefilter('ignore', UserWarning)
                cat.write(tf.name, format='nordic')
            # raises "UserWarning: AIN in header, currently unsupported"
            with warnings.catch_warnings():
                warnings.simplefilter('ignore', UserWarning)
                cat_back = read_events(tf.name)
            for event_1, event_2 in zip(cat, cat_back):
                _assert_similarity(event_1=event_1, event_2=event_2)

    def test_inaccurate_picks(self):
        testing_path = os.path.join(self.testing_path, 'bad_picks.sfile')
        # raises "UserWarning: AIN in header, currently unsupported"
        with warnings.catch_warnings():
            warnings.simplefilter('ignore', UserWarning)
            cat = read_nordic(testing_path)
        pick_string = nordpick(cat[0])
        for pick in pick_string:
            self.assertEqual(len(pick), 80)

    def test_round_len(self):
        testing_path = os.path.join(self.testing_path, 'round_len_undef.sfile')
        # raises "UserWarning: AIN in header, currently unsupported"
        with warnings.catch_warnings():
            warnings.simplefilter('ignore', UserWarning)
            event = read_nordic(testing_path)[0]
        pick_string = nordpick(event)
        for pick in pick_string:
            self.assertEqual(len(pick), 80)

    def test_read_moment(self):
        """
        Test the reading of seismic moment from the s-file.
        """
        testing_path = os.path.join(self.testing_path, 'automag.out')
        # raises "UserWarning: AIN in header, currently unsupported"
        with warnings.catch_warnings():
            warnings.simplefilter('ignore', UserWarning)
            event = read_nordic(testing_path)[0]
        mag = [m for m in event.magnitudes if m.magnitude_type == 'MW']
        self.assertEqual(len(mag), 1)
        self.assertEqual(mag[0].mag, 0.7)

    def test_read_moment_info(self):
        """
        Test reading the info from spectral analysis.
        """
        testing_path = os.path.join(self.testing_path, 'automag.out')
        spec_inf = read_spectral_info(testing_path)
        self.assertEqual(len(spec_inf), 5)
        # This should actually test that what we are reading in is correct.
        average = spec_inf[('AVERAGE', '')]
        check_av = {u'channel': '', u'corner_freq': 5.97, u'decay': 0.0,
                    u'moment': 12589254117.941662, u'moment_mag': 0.7,
                    u'source_radius': 0.231,
                    u'spectral_level': 0.3981071705534972,
                    u'station': 'AVERAGE', u'stress_drop': 0.006,
                    u'window_length': 1.6}
        for key in average.keys():
            if isinstance(average.get(key), str):
                self.assertEqual(average.get(key), check_av.get(key))
            else:
                self.assertEqual(round(average.get(key), 4),
                                 round(check_av.get(key), 4))

    def test_is_sfile(self):
        sfiles = ['01-0411-15L.S201309', 'automag.out', 'bad_picks.sfile',
                  'round_len_undef.sfile', 'Sfile_extra_header',
                  'Sfile_no_location']
        for sfile in sfiles:
            self.assertTrue(_is_sfile(os.path.join(self.testing_path, sfile)))
        self.assertFalse(_is_sfile(os.path.join(self.testing_path,
                                                'Sfile_no_header')))
        self.assertFalse(_is_sfile(os.path.join(self.path, '..', '..',
                                                'nlloc', 'tests', 'data',
                                                'nlloc.hyp')))

    def test_read_picks_across_day_end(self):
        testing_path = os.path.join(self.testing_path, 'sfile_over_day')
        # raises "UserWarning: AIN in header, currently unsupported"
        with warnings.catch_warnings():
            warnings.simplefilter('ignore', UserWarning)
            event = read_nordic(testing_path)[0]
        pick_times = [pick.time for pick in event.picks]
        for pick in event.picks:
            # Pick should come after origin time
            self.assertGreater(pick.time, event.origins[0].time)
            # All picks in this event are within 60s of origin time
            self.assertLessEqual((pick.time - event.origins[0].time), 60)
        # Make sure zero hours and 24 hour picks are handled the same.
        testing_path = os.path.join(self.testing_path, 'sfile_over_day_zeros')
        # raises "UserWarning: AIN in header, currently unsupported"
        with warnings.catch_warnings():
            warnings.simplefilter('ignore', UserWarning)
            event_2 = read_nordic(testing_path)[0]
        for pick in event_2.picks:
            # Pick should come after origin time
            self.assertGreater(pick.time, event_2.origins[0].time)
            # All picks in this event are within 60s of origin time
            self.assertLessEqual((pick.time - event_2.origins[0].time), 60)
            # Each pick should be the same as one pick in the previous event
            self.assertTrue(pick.time in pick_times)
        self.assertEqual(event_2.origins[0].time, event.origins[0].time)

    def test_distance_conversion(self):
        """
        Check that distances are converted properly.
        """
        testing_path = os.path.join(self.testing_path, '01-0411-15L.S201309')
        # raises "UserWarning: AIN in header, currently unsupported"
        with warnings.catch_warnings():
            warnings.simplefilter('ignore', UserWarning)
            cat = read_events(testing_path)
        event = cat[0]
        self.assertAlmostEqual(
            sorted(event.origins[0].arrivals,
                   key=lambda x: x.distance)[0].distance, 0.035972864236749225)
        pick_strings = nordpick(event)
        self.assertEqual(
            int([p for p in pick_strings if p.split()[0] == 'GCSZ' and
                 p.split()[1] == 'SZ'][0].split()[-1]), 304)
        self.assertEqual(
            int([p for p in pick_strings if p.split()[0] == 'WZ11' and
                 p.split()[1] == 'HZ'][0].split()[-1]), 30)

    def test_large_negative_longitude(self):
        event = full_test_event()
        event.origins[0].longitude = -120
        with NamedTemporaryFile(suffix=".out") as tf:
            event.write(tf.name, format="NORDIC")
            event_back = read_events(tf.name)
            _assert_similarity(event, event_back[0])


def _assert_similarity(event_1, event_2, verbose=False):
    """
    Raise AssertionError if testing similarity fails
    """
    if not _test_similarity(event_1, event_2, verbose=verbose):
        raise AssertionError('Events failed similarity check')


def _test_similarity(event_1, event_2, verbose=False):
    """
    Check the similarity of the components of obspy events, discounting
    resource IDs, which are not maintained in nordic files.

    Raise AssertionError if test fails

    :type event_1: obspy.core.event.Event
    :param event_1: First event
    :type event_2: obspy.core.event.Event
    :param event_2: Comparison event
    :type verbose: bool
    :param verbose: If true and fails will output why it fails.
    """
    # Check origins
    if len(event_1.origins) != len(event_2.origins):
        return False
    for ori_1, ori_2 in zip(event_1.origins, event_2.origins):
        for key in ori_1.keys():
            if key not in ["resource_id", "comments", "arrivals",
                           "method_id", "origin_uncertainty", "depth_type",
                           "quality", "creation_info", "evaluation_mode",
                           "depth_errors", "time_errors"]:
                if ori_1[key] != ori_2[key]:
                    if verbose:
                        print('%s is not the same as %s for key %s' %
                              (ori_1[key], ori_2[key], key))
                    return False
            elif key == "arrivals":
                if len(ori_1[key]) != len(ori_2[key]):
                    print('%i is not the same as %i for key %s' %
                          (len(ori_1[key]), len(ori_2[key]), key))
                    return False
                for arr_1, arr_2 in zip(ori_1[key], ori_2[key]):
                    for arr_key in arr_1.keys():
                        if arr_key not in ["resource_id", "pick_id",
                                           "distance"]:
                            if arr_1[arr_key] != arr_2[arr_key]:
                                if verbose:
                                    print('%s does not match %s for key %s' %
                                          (arr_1[arr_key], arr_2[arr_key],
                                           arr_key))
                                return False
                    if arr_1["distance"] and round(
                            arr_1["distance"]) != round(arr_2["distance"]):
                            if verbose:
                                print('%s does not match %s for key %s' %
                                      (arr_1[arr_key], arr_2[arr_key],
                                       arr_key))
                            return False
    # Check picks
    if len(event_1.picks) != len(event_2.picks):
        if verbose:
            print('Number of picks is not equal')
        return False
    for pick_1, pick_2 in zip(event_1.picks, event_2.picks):
        # Assuming same ordering of picks...
        for key in pick_1.keys():
            if key not in ["resource_id", "waveform_id"]:
                if pick_1[key] != pick_2[key]:
                    if verbose:
                        print('%s is not the same as %s for key %s' %
                              (pick_1[key], pick_2[key], key))
                    return False
            elif key == "waveform_id":
                if pick_1[key].station_code != pick_2[key].station_code:
                    if verbose:
                        print('Station codes do not match')
                    return False
                if pick_1[key].channel_code[0] != pick_2[key].channel_code[0]:
                    if verbose:
                        print('Channel codes do not match')
                    return False
                if pick_1[key].channel_code[-1] !=\
                   pick_2[key].channel_code[-1]:
                    if verbose:
                        print('Channel codes do not match')
                    return False
    # Check amplitudes
    if not len(event_1.amplitudes) == len(event_2.amplitudes):
        if verbose:
            print('Not the same number of amplitudes')
        return False
    for amp_1, amp_2 in zip(event_1.amplitudes, event_2.amplitudes):
        # Assuming same ordering of amplitudes
        for key in amp_1.keys():
            if key not in ["resource_id", "pick_id", "waveform_id", "snr",
                           "magnitude_hint", 'type']:
                if not amp_1[key] == amp_2[key]:
                    if verbose:
                        print('%s is not the same as %s for key %s' %
                              (amp_1[key], amp_2[key], key))
                        print(amp_1)
                        print(amp_2)
                    return False
            elif key == "waveform_id":
                if pick_1[key].station_code != pick_2[key].station_code:
                    if verbose:
                        print('Station codes do not match')
                    return False
                if pick_1[key].channel_code[0] != pick_2[key].channel_code[0]:
                    if verbose:
                        print('Channel codes do not match')
                    return False
                if pick_1[key].channel_code[-1] !=\
                   pick_2[key].channel_code[-1]:
                    if verbose:
                        print('Channel codes do not match')
                    return False
            elif key in ["magnitude_hint", "type"]:
                # Reading back in will define both, but input event might have
                # None
                if amp_1[key] is not None:
                    if not amp_1[key] == amp_2[key]:
                        if verbose:
                            print('%s is not the same as %s for key %s' %
                                  (amp_1[key], amp_2[key], key))
                        return False
    return True


def full_test_event():
    """
    Function to generate a basic, full test event
    """
    test_event = Event()
    test_event.origins.append(Origin())
    test_event.origins[0].time = UTCDateTime("2012-03-26") + 1.2
    test_event.event_descriptions.append(EventDescription())
    test_event.event_descriptions[0].text = 'LE'
    test_event.origins[0].latitude = 45.0
    test_event.origins[0].longitude = 25.0
    test_event.origins[0].depth = 15000
    test_event.creation_info = CreationInfo(agency_id='TES')
    test_event.origins[0].quality = OriginQuality(standard_error=0.01)
    test_event.magnitudes.append(Magnitude())
    test_event.magnitudes[0].mag = 0.1
    test_event.magnitudes[0].magnitude_type = 'ML'
    test_event.magnitudes[0].creation_info = CreationInfo('TES')
    test_event.magnitudes[0].origin_id = test_event.origins[0].resource_id
    test_event.magnitudes.append(Magnitude())
    test_event.magnitudes[1].mag = 0.5
    test_event.magnitudes[1].magnitude_type = 'Mc'
    test_event.magnitudes[1].creation_info = CreationInfo('TES')
    test_event.magnitudes[1].origin_id = test_event.origins[0].resource_id
    test_event.magnitudes.append(Magnitude())
    test_event.magnitudes[2].mag = 1.3
    test_event.magnitudes[2].magnitude_type = 'Ms'
    test_event.magnitudes[2].creation_info = CreationInfo('TES')
    test_event.magnitudes[2].origin_id = test_event.origins[0].resource_id

    # Define the test pick
    _waveform_id_1 = WaveformStreamID(station_code='FOZ', channel_code='SHZ',
                                      network_code='NZ')
    _waveform_id_2 = WaveformStreamID(station_code='WTSZ', channel_code='BH1',
                                      network_code=' ')
    # Pick to associate with amplitude - 0
    test_event.picks = [
        Pick(waveform_id=_waveform_id_1, phase_hint='IAML',
             polarity='undecidable', time=UTCDateTime("2012-03-26") + 1.68,
             evaluation_mode="manual"),
        Pick(waveform_id=_waveform_id_1, onset='impulsive', phase_hint='PN',
             polarity='positive', time=UTCDateTime("2012-03-26") + 1.68,
             evaluation_mode="manual"),
        Pick(waveform_id=_waveform_id_1, phase_hint='IAML',
             polarity='undecidable', time=UTCDateTime("2012-03-26") + 1.68,
             evaluation_mode="manual"),
        Pick(waveform_id=_waveform_id_2, onset='impulsive', phase_hint='SG',
             polarity='undecidable', time=UTCDateTime("2012-03-26") + 1.72,
             evaluation_mode="manual"),
        Pick(waveform_id=_waveform_id_2, onset='impulsive', phase_hint='PN',
             polarity='undecidable', time=UTCDateTime("2012-03-26") + 1.62,
             evaluation_mode="automatic")]
    # Test a generic local magnitude amplitude pick
    test_event.amplitudes = [
        Amplitude(generic_amplitude=2.0, period=0.4,
                  pick_id=test_event.picks[0].resource_id,
                  waveform_id=test_event.picks[0].waveform_id, unit='m',
                  magnitude_hint='ML', category='point', type='AML'),
        Amplitude(generic_amplitude=10,
                  pick_id=test_event.picks[1].resource_id,
                  waveform_id=test_event.picks[1].waveform_id, type='END',
                  category='duration', unit='s', magnitude_hint='Mc',
                  snr=2.3),
        Amplitude(generic_amplitude=5.0, period=0.6,
                  pick_id=test_event.picks[2].resource_id,
                  waveform_id=test_event.picks[0].waveform_id, unit='m',
                  category='point', type='AML')]
    test_event.origins[0].arrivals = [
        Arrival(time_weight=0, phase=test_event.picks[1].phase_hint,
                pick_id=test_event.picks[1].resource_id),
        Arrival(time_weight=2, phase=test_event.picks[3].phase_hint,
                pick_id=test_event.picks[3].resource_id,
                backazimuth_residual=5, time_residual=0.2, distance=15,
                azimuth=25),
        Arrival(time_weight=2, phase=test_event.picks[4].phase_hint,
                pick_id=test_event.picks[4].resource_id,
                backazimuth_residual=5, time_residual=0.2, distance=15,
                azimuth=25)]
    return test_event


def suite():
    return unittest.makeSuite(TestNordicMethods, 'test')


if __name__ == '__main__':
    unittest.main(defaultTest='suite')
#!/usr/bin/env python # -*- coding: utf-8 -*- Functions for testing the obspy.io.nordic functions # NOQA @UnusedWildImport Test suite for nordic io operations. Function to test the read and write capabilities of sfile_util. # Set-up a test event # Add the event to a catalogue which can be used for QuakeML testing # Check the read-write s-file functionality # raises "UserWarning: AIN in header, currently unsupported" # assert read_ev.origins[0].resource_id ==\ #     test_ev.origins[0].resource_id # Note that time_residual_RMS is not a quakeML format # assert read_ev.amplitudes[0].resource_id ==\ #     test_ev.amplitudes[0].resource_id # Check coda magnitude pick # Resource ids get overwritten because you can't have two the same in # memory # self.assertEqual(read_ev.amplitudes[1].resource_id, #                  test_ev.amplitudes[1].resource_id) # Resource ids get overwritten because you can't have two the same in # memory # self.assertEqual(read_ev.amplitudes[1].pick_id, #                  test_ev.amplitudes[1].pick_id) # snr is not supported in s-file # self.assertEqual(read_ev.amplitudes[1].snr, #                  test_ev.amplitudes[1].snr) Test a deliberate fail. # Add the event to a catalogue which can be used for QuakeML testing # Raises error due to multiple events in catalog # Raises error due to too long userid # Raises error due to unrecognised event type # Raises error due to no output directory # Write a near empty origin # No wavefile # USER ID too long # Unknown event type # Check that it breaks when writing multiple versions # raises UserWarning: Desired sfile exists, will not overwrite # raises UserWarning: Desired sfile exists, will not overwrite Function to check that writing a blank event works as it should. Function to check a known issue, empty header info S-file: Bug found \         by <NAME>. # raises "UserWarning: AIN in header, currently unsupported" # raises "UserWarning: AIN in header, currently unsupported" # raises "UserWarning: AIN in header, currently unsupported" # Check that a suitable error is raised # raises "UserWarning: AIN in header, currently unsupported" # raises "UserWarning: AIN in header, currently unsupported" # Try to write the same event multiple times, but not overwrite Check that we convert magnitudes as we should! Test the simple string conversions. # Test that full paths are handled # Add the event to a catalogue which can be used for QuakeML testing # Check the read-write s-file functionality # Check that multiple wavefiles are read properly Test the wrapper. # raises "UserWarning: AIN in header, currently unsupported" # raises "UserWarning: AIN in header, currently unsupported" # raises "UserWarning: mb is not convertible" # raises UserWarning: mb is not convertible # raises "UserWarning: AIN in header, currently unsupported" # raises "UserWarning: AIN in header, currently unsupported" # raises "UserWarning: AIN in header, currently unsupported" Test the reading of seismic moment from the s-file. # raises "UserWarning: AIN in header, currently unsupported" Test reading the info from spectral analysis. # This should actually test that what we are reading in is correct. # raises "UserWarning: AIN in header, currently unsupported" # Pick should come after origin time # All picks in this event are within 60s of origin time # Make sure zero hours and 24 hour picks are handled the same. # raises "UserWarning: AIN in header, currently unsupported" # Pick should come after origin time # All picks in this event are within 60s of origin time # Each pick should be the same as one pick in the previous event Check that distances are converted properly. # raises "UserWarning: AIN in header, currently unsupported" Raise AssertionError if testing similarity fails Check the similarity of the components of obspy events, discounting     resource IDs, which are not maintained in nordic files.      Raise AssertionError if test fails      :type event_1: obspy.core.event.Event     :param event_1: First event     :type event_2: obspy.core.event.Event     :param event_2: Comparison event     :type verbose: bool     :param verbose: If true and fails will output why it fails. # Check origins # Check picks # Assuming same ordering of picks... # Check amplitudes # Assuming same ordering of amplitudes # Reading back in will define both, but input event might have # None Function to generate a basic, full test event # Define the test pick # Pick to associate with amplitude - 0 # Test a generic local magnitude amplitude pick