<commit_before># -*- coding: utf-8 -*-
# Copyright (c) 2016, German Neuroinformatics Node (G-Node)
#                     <NAME> <<EMAIL>>
#
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted under the terms of the BSD License. See
# LICENSE file in the root of the Project.
"""
Tests for NixIO
"""

import os
import shutil
try:
    from collections.abc import Iterable
except ImportError:
    from collections import Iterable
from datetime import datetime

from tempfile import mkdtemp

import unittest
import string
import numpy as np
import quantities as pq

from neo.core import (Block, Segment, ChannelIndex, AnalogSignal,
                      IrregularlySampledSignal, Unit, SpikeTrain, Event, Epoch)
from neo.test.iotest.common_io_test import BaseTestIO
from neo.io.nixio import NixIO, create_quantity, units_to_string, neover
from neo.io.nixio_fr import NixIO as NixIO_lazy
from neo.io.proxyobjects import AnalogSignalProxy, SpikeTrainProxy, EventProxy, EpochProxy

try:
    import nixio as nix

    HAVE_NIX = True
except ImportError:
    HAVE_NIX = False

try:
    from unittest import mock
    SKIPMOCK = False
except ImportError:
    SKIPMOCK = True


@unittest.skipUnless(HAVE_NIX, "Requires NIX")
class NixIOTest(unittest.TestCase):
    io = None
    tempdir = None
    filename = None

    def compare_blocks(self, neoblocks, nixblocks):
        for neoblock, nixblock in zip(neoblocks, nixblocks):
            self.compare_attr(neoblock, nixblock)
            self.assertEqual(len(neoblock.segments), len(nixblock.groups))
            for idx, neoseg in enumerate(neoblock.segments):
                nixgrp = nixblock.groups[neoseg.annotations["nix_name"]]
                self.compare_segment_group(neoseg, nixgrp)
            self.assertEqual(len(neoblock.channel_indexes),
                             len(nixblock.sources))
            for idx, neochx in enumerate(neoblock.channel_indexes):
                nixsrc = nixblock.sources[neochx.annotations["nix_name"]]
                self.compare_chx_source(neochx, nixsrc)
            self.check_refs(neoblock, nixblock)

    def compare_chx_source(self, neochx, nixsrc):
        self.compare_attr(neochx, nixsrc)
        nix_channels = list(src for src in nixsrc.sources
                            if src.type == "neo.channelindex")
        self.assertEqual(len(neochx.index), len(nix_channels))
        if len(neochx.channel_ids):
            nix_chanids = list(src.metadata["channel_id"] for src
                               in nixsrc.sources
                               if src.type == "neo.channelindex")
            self.assertEqual(len(neochx.channel_ids), len(nix_chanids))

        # coordinates can be 1D if there's only one channel
        if neochx.coordinates is not None:
            neocoordinates = neochx.coordinates
            if np.ndim(neocoordinates) == 1:
                neocoordinates = [neocoordinates]
        else:
            neocoordinates = []

        for nixchan in nix_channels:
            nixchanidx = nixchan.metadata["index"]
            try:
                neochanpos = list(neochx.index).index(nixchanidx)
            except ValueError:
                self.fail("Channel indexes do not match.")
            if len(neochx.channel_names):
                neochanname = neochx.channel_names[neochanpos]
                if ((not isinstance(neochanname, str)) and
                        isinstance(neochanname, bytes)):
                    neochanname = neochanname.decode()
                nixchanname = nixchan.metadata["neo_name"]
                self.assertEqual(neochanname, nixchanname)
            if len(neochx.channel_ids):
                neochanid = neochx.channel_ids[neochanpos]
                nixchanid = nixchan.metadata["channel_id"]
                self.assertEqual(neochanid, nixchanid)
            elif "channel_id" in nixchan.metadata:
                self.fail("Channel ID not loaded")

            if len(neocoordinates):
                neocoord = neocoordinates[neochanpos]
                nixcoord = nixchan.metadata.props["coordinates"]
                nixcoord = create_quantity(nixcoord.values, nixcoord.unit)
                self.assertTrue(all(neocoord == nixcoord),
                                msg="{} != {}".format(neocoord, nixcoord))
        nix_units = list(src for src in nixsrc.sources
                         if src.type == "neo.unit")
        self.assertEqual(len(neochx.units), len(nix_units))
        for neounit in neochx.units:
            nixunit = nixsrc.sources[neounit.annotations["nix_name"]]
            self.compare_attr(neounit, nixunit)

    def check_refs(self, neoblock, nixblock):
        """
        Checks whether the references between objects that are not nested are
        mapped correctly (e.g., SpikeTrains referenced by a Unit).

        :param neoblock: A Neo block<commit_msg>[nixio] Test reading channel names<commit_after># -*- coding: utf-8 -*-
# Copyright (c) 2016, German Neuroinformatics Node (G-Node)
#                     <NAME> <<EMAIL>>
#
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted under the terms of the BSD License. See
# LICENSE file in the root of the Project.
"""
Tests for NixIO
"""

import os
import shutil
try:
    from collections.abc import Iterable
except ImportError:
    from collections import Iterable
from datetime import datetime

from tempfile import mkdtemp

import unittest
import string
import numpy as np
import quantities as pq

from neo.core import (Block, Segment, ChannelIndex, AnalogSignal,
                      IrregularlySampledSignal, Unit, SpikeTrain, Event, Epoch)
from neo.test.iotest.common_io_test import BaseTestIO
from neo.io.nixio import NixIO, create_quantity, units_to_string, neover
from neo.io.nixio_fr import NixIO as NixIO_lazy
from neo.io.proxyobjects import AnalogSignalProxy, SpikeTrainProxy, EventProxy, EpochProxy

try:
    import nixio as nix

    HAVE_NIX = True
except ImportError:
    HAVE_NIX = False

try:
    from unittest import mock
    SKIPMOCK = False
except ImportError:
    SKIPMOCK = True


@unittest.skipUnless(HAVE_NIX, "Requires NIX")
class NixIOTest(unittest.TestCase):
    io = None
    tempdir = None
    filename = None

    def compare_blocks(self, neoblocks, nixblocks):
        for neoblock, nixblock in zip(neoblocks, nixblocks):
            self.compare_attr(neoblock, nixblock)
            self.assertEqual(len(neoblock.segments), len(nixblock.groups))
            for idx, neoseg in enumerate(neoblock.segments):
                nixgrp = nixblock.groups[neoseg.annotations["nix_name"]]
                self.compare_segment_group(neoseg, nixgrp)
            self.assertEqual(len(neoblock.channel_indexes),
                             len(nixblock.sources))
            for idx, neochx in enumerate(neoblock.channel_indexes):
                nixsrc = nixblock.sources[neochx.annotations["nix_name"]]
                self.compare_chx_source(neochx, nixsrc)
            self.check_refs(neoblock, nixblock)

    def compare_chx_source(self, neochx, nixsrc):
        self.compare_attr(neochx, nixsrc)
        nix_channels = list(src for src in nixsrc.sources
                            if src.type == "neo.channelindex")
        self.assertEqual(len(neochx.index), len(nix_channels))
        if len(neochx.channel_ids):
            nix_chanids = list(src.metadata["channel_id"] for src
                               in nixsrc.sources
                               if src.type == "neo.channelindex")
            self.assertEqual(len(neochx.channel_ids), len(nix_chanids))

        # coordinates can be 1D if there's only one channel
        if neochx.coordinates is not None:
            neocoordinates = neochx.coordinates
            if np.ndim(neocoordinates) == 1:
                neocoordinates = [neocoordinates]
        else:
            neocoordinates = []

        for nixchan in nix_channels:
            nixchanidx = nixchan.metadata["index"]
            try:
                neochanpos = list(neochx.index).index(nixchanidx)
            except ValueError:
                self.fail("Channel indexes do not match.")

            if len(neochx.channel_names):
                neochanname = neochx.channel_names[neochanpos]
                if ((not isinstance(neochanname, str)) and
                        isinstance(neochanname, bytes)):
                    neochanname = neochanname.decode()
                nixchanname = nixchan.metadata["neo_name"]
                self.assertEqual(neochanname, nixchanname)
            else:
                # Check if channel name exists but not loaded
                self.assertNotIn("neo_name", nixchan.metadata)
            if len(neochx.channel_ids):
                neochanid = neochx.channel_ids[neochanpos]
                nixchanid = nixchan.metadata["channel_id"]
                self.assertEqual(neochanid, nixchanid)
            elif "channel_id" in nixchan.metadata:
                self.fail("Channel ID not loaded")

            if len(neocoordinates):
                neocoord = neocoordinates[neochanpos]
                nixcoord = nixchan.metadata.props["coordinates"]
                nixcoord = create_quantity(nixcoord.values, nixcoord.unit)
                self.assertTrue(all(neocoord == nixcoord),
                                msg="{} != {}".format(neocoord, nixcoord))
        nix_units = list(src for src in nixsrc.sources
                         if src.type == "neo.unit")
        self.assertEqual(len(neochx.units), len(nix_units))
        for neounit in neochx.units:
            nixunit = nixsrc.sources[neounit.annotations["nix_name"]]
            self.compare_attr(neounit, nixunit)

    def check_refs(self, neoblock, nixblock):
        """
        Checks whether the references between objects that are not nested are
        mapped correctly (e.g., SpikeTrains referenced by a Unit).

        :param neoblock: A Neo block
{'fraction_non_alphanumeric': 0.08384009570331971, 'fraction_numerical': 0.0013956734124214933, 'mean_word_length': 2.293499671700591, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3601660', 'n_tokens_mistral': 3063, 'n_tokens_neox': 2881, 'n_words': 716}