# -*- coding: utf-8 -*-

"""
Tests the TextReader class in parsers.pyx, which
is integral to the C engine in parsers.py
"""

import pytest

from pandas.compat import StringIO, BytesIO, map
from pandas import compat

import os
import sys

from numpy import nan
import numpy as np

from pandas import DataFrame
from pandas.io.parsers import (read_csv, TextFileReader)
from pandas.util.testing import assert_frame_equal

import pandas.util.testing as tm

from pandas._libs.parsers import TextReader
import pandas._libs.parsers as parser


class TestTextReader(object):

    @pytest.fixture(autouse=True)
    def setup_method(self, datapath):
        self.dirpath = datapath('io', 'parser', 'data')
        self.csv1 = os.path.join(self.dirpath, 'test1.csv')
        self.csv2 = os.path.join(self.dirpath, 'test2.csv')
        self.xls1 = os.path.join(self.dirpath, 'test.xls')

    def test_file_handle(self):
        with open(self.csv1, 'rb') as f:
            reader = TextReader(f)
            reader.read()

    def test_string_filename(self):
        reader = TextReader(self.csv1, header=None)
        reader.read()

    def test_file_handle_mmap(self):
        with open(self.csv1, 'rb') as f:
            reader = TextReader(f, memory_map=True, header=None)
            reader.read()

    def test_StringIO(self):
        with open(self.csv1, 'rb') as f:
            text = f.read()
        src = BytesIO(text)
        reader = TextReader(src, header=None)
        reader.read()

    def test_string_factorize(self):
        # should this be optional?
        data = 'a\nb\na\nb\na'
        reader = TextReader(StringIO(data), header=None)
        result = reader.read()
        assert len(set(map(id, result[0]))) == 2

    def test_skipinitialspace(self):
        data = ('a,   b\n'
                'a,   b\n'
                'a,   b\n'
                'a,   b')

        reader = TextReader(StringIO(data), skipinitialspace=True,
                            header=None)
        result = reader.read()

        tm.assert_numpy_array_equal(result[0], np.array(['a', 'a', 'a', 'a'],
                                                        dtype=np.object_))
        tm.assert_numpy_array_equal(result[1], np.array(['b', 'b', 'b', 'b'],
                                                        dtype=np.object_))

    def test_parse_booleans(self):
        data = 'True\nFalse\nTrue\nTrue'

        reader = TextReader(StringIO(data), header=None)
        result = reader.read()

        assert result[0].dtype == np.bool_

    def test_delimit_whitespace(self):
        data = 'a  b\na\t\t "b"\n"a"\t \t b'

        reader = TextReader(StringIO(data), delim_whitespace=True,
                            header=None)
        result = reader.read()

        tm.assert_numpy_array_equal(result[0], np.array(['a', 'a', 'a'],
                                                        dtype=np.object_))
        tm.assert_numpy_array_equal(result[1], np.array(['b', 'b', 'b'],
                                                        dtype=np.object_))

    def test_embedded_newline(self):
        data = 'a\n"hello\nthere"\nthis'

        reader = TextReader(StringIO(data), header=None)
        result = reader.read()

        expected = np.array(['a', 'hello\nthere', 'this'], dtype=np.object_)
        tm.assert_numpy_array_equal(result[0], expected)

    def test_euro_decimal(self):
        data = '12345,67\n345,678'

        reader = TextReader(StringIO(data), delimiter=':',
                            decimal=',', header=None)
        result = reader.read()

        expected = np.array([12345.67, 345.678])
        tm.assert_almost_equal(result[0], expected)

    def test_integer_thousands(self):
        data = '123,456\n12,500'

        reader = TextReader(StringIO(data), delimiter=':',
                            thousands=',', header=None)
        result = reader.read()

        expected = np.array([123456, 12500], dtype=np.int64)
        tm.assert_almost_equal(result[0], expected)

    def test_integer_thousands_alt(self):
        data = '123.456\n12.500'

        reader = TextFileReader(StringIO(data), delimiter=':',
                                thousands='.', header=None)
        result = reader.read()

        expected = DataFrame([123456, 12500])
        tm.assert_frame_equal(result, expected)

    @tm.capture_stderr
    def test_skip_bad_lines(self):
        # too many lines, see #2430 for why
        data = ('a:b:c\n'
                'd:e:f\n'
                'g:h:i\n'
                'j:k:l:m\n'
                'l:m:n\n'
                'o:p:q:r')

        reader = TextReader(StringIO(data), delimiter=':',
                            header=None)
        pytest.raises(parser.ParserError, reader.read)

        reader = TextReader(StringIO(data), delimiter=':',
                            header=None,
                            error_bad_lines=False,
                            warn_bad_lines=False)
        result = reader.read()
        expected = {0: np.array(['a', 'd', 'g', 'l'], dtype=object),
                    1: np.array(['b', 'e', 'h', 'm'], dtype=object),
                    2: np.array(['c', 'f', 'i', 'n'], dtype=object)}
        assert_array_dicts_equal(result, expected)

        reader = TextReader(StringIO(data), delimiter=':',
                            header=None,
                            error_bad_lines=False,
                            warn_bad_lines=True)
        reader.read()
        val = sys.stderr.getvalue()

        assert 'Skipping line 4' in val
        assert 'Skipping line 6' in val

    def test_header_not_enough_lines(self):
        data = ('skip this\n'
                'skip this\n'
                'a,b,c\n'
                '1,2,3\n'
                '4,5,6')

        reader = TextReader(StringIO(data), delimiter=',', header=2)
        header = reader.header
        expected = [['a', 'b', 'c']]
        assert header == expected

        recs = reader.read()
        expected = {0: np.array([1, 4], dtype=np.int64),
                    1: np.array([2, 5], dtype=np.int64),
                    2: np.array([3, 6], dtype=np.int64)}
        assert_array_dicts_equal(recs, expected)

    def test_escapechar(self):
        data = ('\\"hello world\"\n'
                '\\"hello world\"\n'
                '\\"hello world\"')

        reader = TextReader(StringIO(data), delimiter=',', header=None,
                            escapechar='\\')
        result = reader.read()
        expected = {0: np.array(['"hello world"'] * 3, dtype=object)}
        assert_array_dicts_equal(result, expected)

    def test_eof_has_eol(self):
        # handling of new line at EOF
        pass

    def test_na_substitution(self):
        pass

    def test_numpy_string_dtype(self):
        data = """\
a,1
aa,2
aaa,3
aaaa,4
aaaaa,5"""

        def _make_reader(**kwds):
            return TextReader(StringIO(data), delimiter=',', header=None,
                              **kwds)

        reader = _make_reader(dtype='S5,i4')
        result = reader.read()

        assert result[0].dtype == 'S5'

        ex_values = np.array(['a', 'aa', 'aaa', 'aaaa', 'aaaaa'], dtype='S5')
        assert (result[0] == ex_values).all()
        assert result[1].dtype == 'i4'

        reader = _make_reader(dtype='S4')
        result = reader.read()
        assert result[0].dtype == 'S4'
        ex_values = np.array(['a', 'aa', 'aaa', 'aaaa', 'aaaa'], dtype='S4')
        assert (result[0] == ex_values).all()
        assert result[1].dtype == 'S4'

    def test_pass_dtype(self):
        data = """\
one,two
1,a
2,b
3,c
4,d"""

        def _make_reader(**kwds):
            return TextReader(StringIO(data), delimiter=',', **kwds)

        reader = _make_reader(dtype={'one': 'u1', 1: 'S1'})
        result = reader.read()
        assert result[0].dtype == 'u1'
        assert result[1].dtype == 'S1'

        reader = _make_reader(dtype={'one': np.uint8, 1: object})
        result = reader.read()
        assert result[0].dtype == 'u1'
        assert result[1].dtype == 'O'

        reader = _make_reader(dtype={'one': np.dtype('u1'),
                                     1: np.dtype('O')})
        result = reader.read()
        assert result[0].dtype == 'u1'
        assert result[1].dtype == 'O'

    def test_usecols(self):
        data = """\
a,b,c
1,2,3
4,5,6
7,8,9
10,11,12"""

        def _make_reader(**kwds):
            return TextReader(StringIO(data), delimiter=',', **kwds)

        reader = _make_reader(usecols=(1, 2))
        result = reader.read()

        exp = _make_reader().read()
        assert len(result) == 2
        assert (result[1] == exp[1]).all()
        assert (result[2] == exp[2]).all()

    def test_cr_delimited(self):
        def _test(text, **kwargs):
            nice_text = text.replace('\r', '\r\n')
            result = TextReader(StringIO(text), **kwargs).read()
            expected = TextReader(StringIO(nice_text), **kwargs).read()
            assert_array_dicts_equal(result, expected)

        data = 'a,b,c\r1,2,3\r4,5,6\r7,8,9\r10,11,12'
        _test(data, delimiter=',')

        data = 'a  b  c\r1  2  3\r4  5  6\r7  8  9\r10  11  12'
        _test(data, delim_whitespace=True)

        data = 'a,b,c\r1,2,3\r4,5,6\r,88,9\r10,11,12'
        _test(data, delimiter=',')

        sample = ('A,B,C,D,E,F,G,H,I,J,K,L,M,N,O\r'
                  'AAAAA,BBBBB,0,0,0,0,0,0,0,0,0,0,0,0,0\r'
                  ',BBBBB,0,0,0,0,0,0,0,0,0,0,0,0,0')
        _test(sample, delimiter=',')

        data = 'A  B  C\r  2  3\r4  5  6'
        _test(data, delim_whitespace=True)

        data = 'A B C\r2 3\r4 5 6'
        _test(data, delim_whitespace=True)

    def test_empty_field_eof(self):
        data = 'a,b,c\n1,2,3\n4,,'

        result = TextReader(StringIO(data), delimiter=',').read()

        expected = {0: np.array([1, 4], dtype=np.int64),
                    1: np.array(['2', ''], dtype=object),
                    2: np.array(['3', ''], dtype=object)}
        assert_array_dicts_equal(result, expected)

        # GH5664
        a = DataFrame([['b'], [nan]], columns=['a'], index=['a', 'c'])
        b = DataFrame([[1, 1, 1, 0], [1, 1, 1, 0]],
                      columns=list('abcd'),
                      index=[1, 1])
        c = DataFrame([[1, 2, 3, 4], [6, nan, nan, nan],
                       [8, 9, 10, 11], [13, 14, nan, nan]],
                      columns=list('abcd'),
                      index=[0, 5, 7, 12])

        for _ in range(100):
            df = read_csv(StringIO('a,b\nc\n'), skiprows=0,
                          names=['a'], engine='c')
            assert_frame_equal(df, a)

            df = read_csv(StringIO('1,1,1,1,0\n' * 2 + '\n' * 2),
                          names=list("abcd"), engine='c')
            assert_frame_equal(df, b)

            df = read_csv(StringIO('0,1,2,3,4\n5,6\n7,8,9,10,11\n12,13,14'),
                          names=list('abcd'), engine='c')
            assert_frame_equal(df, c)

    def test_empty_csv_input(self):
        # GH14867
        df = read_csv(StringIO(), chunksize=20, header=None,
                      names=['a', 'b', 'c'])
        assert isinstance(df, TextFileReader)


def assert_array_dicts_equal(left, right):
    for k, v in compat.iteritems(left):
        assert tm.assert_numpy_array_equal(np.asarray(v),
                                           np.asarray(right[k]))