test_nordic.py

#!/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, Event, Origin, Magnitude,
    OriginQuality, EventDescription, CreationInfo, OriginUncertainty,
    ConfidenceEllipsoid, QuantityError, FocalMechanism, MomentTensor,
    NodalPlane, NodalPlanes, ResourceIdentifier, Tensor)
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, readwavename, blanksfile,
    _write_nordic, nordpick, readheader, _int_conv, _readheader, _evmagtonor,
    write_select, NordicParsingError, _float_conv, _nortoevmag, _str_conv,
    _get_line_tags, xyz_to_confidence_ellipsoid, confidence_ellipsoid_to_xyz)


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()
        # Sort the magnitudes - they are sorted on writing and we need to check
        # like-for-like
        test_event.magnitudes.sort(key=lambda obj: obj['mag'], reverse=True)
        # 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)
        # 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 Dominic Evanzia.
        """
        # 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)
                tagged_lines = _get_line_tags(f=f)
                head_2 = _readheader(head_lines=tagged_lines['1'])
        _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'), ('s', 'Ms')]
        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)

    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)
            if not _is_sfile(tf.name):
                raise(NordicParsingError("format is wrong somehow"))
            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_more_than_three_mags(self):
        cat = Catalog()
        cat += full_test_event()
        cat[0].magnitudes.append(Magnitude(
            mag=0.9, magnitude_type='MS', creation_info=CreationInfo('TES'),
            origin_id=cat[0].origins[0].resource_id))
        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):
                self.assertTrue(
                    len(event_1.magnitudes) == len(event_2.magnitudes))
                self.assertTrue(test_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_convert_xyz_error(self):
        """
        Test that we are able to map between xyz errors and ellipsoid
        """
        errors = {'x_err': 640, 'y_err': 1300, 'z_err': 1700,
                  'xy_cov': -990000, 'xz_cov': -405200, 'yz_cov': 239200}
        confidence_ellipsoid = xyz_to_confidence_ellipsoid(errors)
        errors_back = confidence_ellipsoid_to_xyz(confidence_ellipsoid)
        self.assertEqual(errors, errors_back)
        # TODO: Test mapping for known result.


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"]:
                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
            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
    return True


def full_test_event():
    """
    Function to generate a basic, full test event
    """
    test_event = Event()
    test_event.origins.append(Origin(
        time=UTCDateTime("2012-03-26") + 1.2, latitude=45.0, longitude=25.0,
        depth=15000))
    test_event.event_descriptions.append(EventDescription())
    test_event.event_descriptions[0].text = 'LE'
    test_event.creation_info = CreationInfo(agency_id='TES')
    test_event.magnitudes.append(Magnitude(
        mag=0.1, magnitude_type='ML', creation_info=CreationInfo('TES'),
        origin_id=test_event.origins[0].resource_id))
    test_event.magnitudes.append(Magnitude(
        mag=0.5, magnitude_type='Mc', creation_info=CreationInfo('TES'),
        origin_id=test_event.origins[0].resource_id))
    test_event.magnitudes.append(Magnitude(
        mag=1.3, magnitude_type='Ms', creation_info=CreationInfo('TES'),
        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
    test_event.picks.append(
        Pick(waveform_id=_waveform_id_1, phase_hint='IAML',
             polarity='undecidable', time=UTCDateTime("2012-03-26") + 1.68,
             evaluation_mode="manual"))
    # Need a second pick for coda
    test_event.picks.append(
        Pick(waveform_id=_waveform_id_1, onset='impulsive', phase_hint='PN',
             polarity='positive', time=UTCDateTime("2012-03-26") + 1.68,
             evaluation_mode="manual"))
    # Unassociated pick
    test_event.picks.append(
        Pick(waveform_id=_waveform_id_2, onset='impulsive', phase_hint='SG',
             polarity='undecidable', time=UTCDateTime("2012-03-26") + 1.72,
             evaluation_mode="manual"))
    # Unassociated pick
    test_event.picks.append(
        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.append(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'))
    # Test a coda magnitude pick
    test_event.amplitudes.append(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))
    test_event.origins[0].arrivals.append(Arrival(
        time_weight=0, phase=test_event.picks[1].phase_hint,
        pick_id=test_event.picks[1].resource_id))
    test_event.origins[0].arrivals.append(Arrival(
        time_weight=2, phase=test_event.picks[2].phase_hint,
        pick_id=test_event.picks[2].resource_id, backazimuth_residual=5,
        time_residual=0.2, distance=15, azimuth=25))
    test_event.origins[0].arrivals.append(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))
    # Add in error info (line E)
    test_event.origins[0].quality = OriginQuality(
        standard_error=0.01, azimuthal_gap=36)
    # Origin uncertainty in Seisan is output as long-lat-depth, quakeML has
    # semi-major and semi-minor
    test_event.origins[0].origin_uncertainty = OriginUncertainty(
        confidence_ellipsoid=ConfidenceEllipsoid(
            semi_major_axis_length=3000, semi_minor_axis_length=1000,
            semi_intermediate_axis_length=2000, major_axis_plunge=20,
            major_axis_azimuth=100, major_axis_rotation=4))
    test_event.origins[0].time_errors = QuantityError(uncertainty=0.5)
    # Add in fault-plane solution info (line F) - Note have to check program
    # used to determine which fields are filled....
    test_event.focal_mechanisms.append(FocalMechanism(
        nodal_planes=NodalPlanes(nodal_plane_1=NodalPlane(
            strike=180, dip=20, rake=30, strike_errors=QuantityError(10),
            dip_errors=QuantityError(10), rake_errors=QuantityError(20))),
        method_id=ResourceIdentifier("smi:nc.anss.org/focalMechanism/FPFIT"),
        creation_info=CreationInfo(agency_id="NC"), misfit=0.5,
        station_distribution_ratio=0.8))
    # Need to test high-precision origin and that it is preferred origin.
    # Moment tensor includes another origin
    test_event.origins.append(Origin(
        time=UTCDateTime("2012-03-26") + 1.2, latitude=45.1, longitude=25.2,
        depth=14500))
    test_event.magnitudes.append(Magnitude(
        mag=0.1, magnitude_type='MW', creation_info=CreationInfo('TES'),
        origin_id=test_event.origins[-1].resource_id))
    # Moment tensors go with focal-mechanisms
    test_event.focal_mechanisms.append(FocalMechanism(
        moment_tensor=MomentTensor(
            derived_origin_id=test_event.origins[-1].resource_id,
            moment_magnitude_id=test_event.magnitudes[-1].resource_id,
            scalar_moment=100, tensor=Tensor(
                m_rr=100, m_tt=100, m_pp=10, m_rt=1, m_rp=20, m_tp=15),
            method_id=ResourceIdentifier('smi:nc.anss.org/momentTensor/BLAH'))))
    return test_event


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


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