#!/usr/bin/env python
# test expression parsing

import os
import sys
import time
import unittest
import numpy
import ctypes
from contextlib import contextmanager
from epics import ca, dbr, caput

import pvnames

def _ca_connect(chid,timeout=5.0):
    n  = 0
    t0 = time.time()
    conn = 2==ca.state(chid)
    while (not conn) and (time.time()-t0 < timeout):
        ca.poll(1.e-6,1.e-4)
        conn = 2==ca.state(chid)
        n += 1
    return conn, time.time()-t0, n

def write(msg):
    sys.stdout.write('%s\n'% msg)
    sys.stdout.flush()

CONN_DAT ={}
CHANGE_DAT = {}

def onConnect(pvname=None, conn=None, chid=None,  **kws):
    write('  /// Connection status changed:  %s  %s' % (pvname, repr(kws)))
    global CONN_DAT
    CONN_DAT[pvname] = conn

def onChanges(pvname=None, value=None, **kws):
    write( '/// New Value: %s  value=%s, kw=%s' %( pvname, str(value), repr(kws)))
    global CHANGE_DAT
    CHANGE_DAT[pvname] = value

@contextmanager
def no_simulator_updates():
    '''Context manager which pauses and resumes simulator PV updating'''
    try:
        caput(pvnames.pause_pv, 1)
        yield
    finally:
        caput(pvnames.pause_pv, 0)

class CA_BasicTests(unittest.TestCase):

    def setUp(self):
        write('Starting %s...' % self.id().split(".")[-1])
    def tearDown(self):
        write('Completed %s...\n' % self.id().split(".")[-1])

    def testA_CreateChid(self):
        write('Simple Test: create chid')
        chid = ca.create_channel(pvnames.double_pv)
        self.assertIsNot(chid, None)

    def testA_GetNonExistentPV(self):
        write('Simple Test: get on a non-existent PV')
        chid = ca.create_channel('Definitely-Not-A-Real-PV')
        val = ca.get(chid)
        self.assertEqual(val, None)

    def testA_CreateChid_CheckTypeCount(self):
        write('Simple Test: create chid, check count, type, host, and access')
        chid = ca.create_channel(pvnames.double_pv)
        ret = ca.connect_channel(chid)
        ca.pend_event(1.e-3)

        ftype  = ca.field_type(chid)
        count  = ca.element_count(chid)
        host    = ca.host_name(chid)
        rwacc = ca.access(chid)

        self.assertIsNot(chid, None)
        self.assertIsNot(host, None)
        self.assertEqual(count, 1)
        self.assertEqual(ftype, 6)
        self.assertEqual(rwacc,'read/write')


    def testA_CreateChidWithConn(self):
        write('Simple Test: create chid with conn callback')
        chid = ca.create_channel(pvnames.int_pv,
                                 callback=onConnect)
        val = ca.get(chid)

        global CONN_DAT
        conn = CONN_DAT.get(pvnames.int_pv, None)
        self.assertEqual(conn, True)

    def test_dbrName(self):
        write( 'DBR Type Check')
        self.assertEqual(dbr.Name(dbr.STRING), 'STRING')
        self.assertEqual(dbr.Name(dbr.FLOAT),  'FLOAT')
        self.assertEqual(dbr.Name(dbr.ENUM), 'ENUM')
        self.assertEqual(dbr.Name(dbr.CTRL_CHAR), 'CTRL_CHAR')
        self.assertEqual(dbr.Name(dbr.TIME_DOUBLE), 'TIME_DOUBLE')
        self.assertEqual(dbr.Name(dbr.TIME_LONG), 'TIME_LONG')

        self.assertEqual(dbr.Name('STRING', reverse=True), dbr.STRING)
        self.assertEqual(dbr.Name('DOUBLE', reverse=True), dbr.DOUBLE)
        self.assertEqual(dbr.Name('CTRL_ENUM', reverse=True), dbr.CTRL_ENUM)
        self.assertEqual(dbr.Name('TIME_LONG', reverse=True), dbr.TIME_LONG)

    def test_Connect1(self):
        chid = ca.create_channel(pvnames.double_pv)
        conn,dt,n = _ca_connect(chid, timeout=2)
        write( 'CA Connection Test1: connect to existing PV')
        write( ' connected in %.4f sec' % (dt))
        self.assertEqual(conn,True)

    def test_Connected(self):
        pvn = pvnames.double_pv
        chid = ca.create_channel(pvn,connect=True)
        isconn = ca.isConnected(chid)
        write( 'CA test Connected (%s) = %s' % (pvn,isconn))
        self.assertEqual(isconn,True)
        state= ca.state(chid)
        self.assertEqual(state,ca.dbr.CS_CONN)
        acc = ca.access(chid)
        self.assertEqual(acc,'read/write')


    def test_DoubleVal(self):
        pvn = pvnames.double_pv
        chid = ca.create_channel(pvn,connect=True)
        cdict  = ca.get_ctrlvars(chid)
        write( 'CA testing CTRL Values for a Double (%s)'   % (pvn))
        self.failUnless('units' in cdict)
        self.failUnless('precision' in cdict)
        self.failUnless('severity' in cdict)

        hostname = ca.host_name(chid)
        self.failUnless(len(hostname) > 1)

        count = ca.element_count(chid)
        self.assertEqual(count,1)

        ftype= ca.field_type(chid)
        self.assertEqual(ftype,ca.dbr.DOUBLE)

        prec = ca.get_precision(chid)
        self.assertEqual(prec, pvnames.double_pv_prec)

        units= ca.BYTES2STR(ca.get_ctrlvars(chid)['units'])
        self.assertEqual(units, pvnames.double_pv_units)

        rwacc= ca.access(chid)
        self.failUnless(rwacc.startswith('read'))


    def test_UnConnected(self):
        write( 'CA Connection Test1: connect to non-existing PV (2sec timeout)')
        chid = ca.create_channel('impossible_pvname_certain_to_fail')
        conn,dt,n = _ca_connect(chid, timeout=2)
        self.assertEqual(conn,False)


    def test_putwait(self):
        'test put with wait'
        pvn = pvnames.non_updating_pv
        chid = ca.create_channel(pvn, connect=True)
        o  = ca.put(chid, -1, wait=True)
        val = ca.get(chid)
        self.assertEqual(val, -1)
        o  = ca.put(chid, 2, wait=True)
        val = ca.get(chid)
        self.assertEqual(val, 2)

    def test_promote_type(self):
        pvn = pvnames.double_pv
        chid = ca.create_channel(pvn,connect=True)
        write( 'CA promote type (%s)' % (pvn))
        f_t  = ca.promote_type(chid,use_time=True)
        f_c  = ca.promote_type(chid,use_ctrl=True)
        self.assertEqual(f_t, ca.dbr.TIME_DOUBLE)
        self.assertEqual(f_c, ca.dbr.CTRL_DOUBLE)

    def test_ProcPut(self):
        pvn  = pvnames.enum_pv
        chid = ca.create_channel(pvn, connect=True)
        write( 'CA test put to PROC Field (%s)' % (pvn))
        for input in (1, '1', 2, '2', 0, '0', 50, 1):
            ret = None
            try:
                ret = ca.put(chid, 1)
            except:
                pass
            self.assertIsNot(ret, None)

    def test_subscription_double(self):
        pvn = pvnames.updating_pv1
        chid = ca.create_channel(pvn,connect=True)
        cb, uarg, eventID = ca.create_subscription(chid, callback=onChanges)

        start_time = time.time()
        global CHANGE_DAT
        while time.time()-start_time < 5.0:
            time.sleep(0.01)
            if CHANGE_DAT.get(pvn, None) is not None:
                break
        val = CHANGE_DAT.get(pvn, None)
        ca.clear_subscription(eventID)
        self.assertIsNot(val, None)

    def test_subscription_custom(self):
        pvn = pvnames.updating_pv1
        chid = ca.create_channel(pvn, connect=True)

        global change_count
        change_count = 0

        def my_callback(pvname=None, value=None, **kws):
            write( ' Custom Callback  %s  value=%s' %(pvname, str(value)))
            global change_count
            change_count = change_count + 1

        cb, uarg, eventID = ca.create_subscription(chid, callback=my_callback)

        start_time = time.time()
        while time.time()-start_time < 2.0:
            time.sleep(0.01)

        ca.clear_subscription(eventID)
        time.sleep(0.2)
        self.assertTrue(change_count > 2)

    def test_subscription_str(self):

        pvn = pvnames.updating_str1
        write(" Subscription on string: %s " % pvn)
        chid = ca.create_channel(pvn,connect=True)
        cb, uarg, eventID = ca.create_subscription(chid, callback=onChanges)

        start_time = time.time()
        global CHANGE_DAT
        while time.time()-start_time < 3.0:
            time.sleep(0.01)
            ca.put(chid, "%.1f" % (time.time()-start_time) )
            if CHANGE_DAT.get(pvn, None) is not None:
                break
        val = CHANGE_DAT.get(pvn, None)
        # ca.clear_subscription(eventID)
        self.assertIsNot(val, None)
        time.sleep(0.2)


    def _test_array_callback(self, arrayname, array_type, length, element_type):
        """ Helper function to subscribe to a PV array and check it
        receives at least one subscription callback w/ specified type,
        length & uniform element type. Checks separately for normal,
        TIME & CTRL subscription variants. Returns the array or fails
        an assertion."""
        results = {}
        for form in [ 'normal', 'time', 'ctrl' ]:
            chid = ca.create_channel(arrayname,connect=True)
            cb, uarg, eventID = ca.create_subscription(chid, use_time=form=='time', use_ctrl=form=='ctrl', callback=onChanges)

            CHANGE_DAT.pop(arrayname, None)
            timeout=0
            # wait up to 6 seconds, if no callback probably due to simulator.py
            # not running...
            while timeout<120 and not arrayname in CHANGE_DAT:
                time.sleep(0.05)
                timeout = timeout+1
            val = CHANGE_DAT.get(arrayname, None)
            ca.clear_subscription(eventID)
            self.assertIsNot(val, None)
            self.assertEqual(type(val), array_type)
            self.assertEqual(len(val), length)
            self.assertEqual(type(val[0]), element_type)
            self.assertTrue(all( type(e)==element_type for e in val))
            results[form] = val
        return results

    def test_subscription_long_array(self):
        """ Check that numeric arrays callbacks successfully send correct data """
        self._test_array_callback(pvnames.long_arr_pv, numpy.ndarray, 2048, numpy.int32)

    def test_subscription_double_array(self):
        """ Check that double arrays callbacks successfully send correct data """
        self._test_array_callback(pvnames.double_arr_pv, numpy.ndarray, 2048, numpy.float64)

    def test_subscription_string_array(self):
        """ Check that string array callbacks successfully send correct data """
        results = self._test_array_callback(pvnames.string_arr_pv, list, 128, str)
        self.assertTrue(len(results["normal"][0]) > 0)
        self.assertTrue(len(results["time"][0]) > 0)
        self.assertTrue(len(results["ctrl"][0]) > 0)

    def test_subscription_char_array(self):
        """ Check that uchar array callbacks successfully send correct data as arrays """
        self._test_array_callback(pvnames.char_arr_pv, numpy.ndarray, 128, numpy.uint8)



    def test_Values(self):
        write( 'CA test Values (compare 5 values with caget)')
        os.system('rm ./caget.tst')
        vals = {}
        with no_simulator_updates():
            for pvn in (pvnames.str_pv,  pvnames.int_pv,
                        pvnames.float_pv, pvnames.enum_pv,
                        pvnames.long_pv):
                os.system('caget  -n -f5 %s >> ./caget.tst' % pvn)
                chid = ca.create_channel(pvn)
                ca.connect_channel(chid)
                vals[pvn] = ca.get(chid)
            rlines = open('./caget.tst', 'r').readlines()
            for line in rlines:
                pvn, sval = [i.strip() for i in line[:-1].split(' ', 1)]
                tval = str(vals[pvn])
                if pvn in (pvnames.float_pv,pvnames.double_pv):
                    # use float precision!
                    tval = "%.5f" % vals[pvn]
                self.assertEqual(tval, sval)

    def test_type_converions_1(self):
        write("CA type conversions scalars")
        pvlist = (pvnames.str_pv, pvnames.int_pv, pvnames.float_pv,
                  pvnames.enum_pv,  pvnames.long_pv,  pvnames.double_pv2)
        chids = []
        with no_simulator_updates():
            for name in pvlist:
                chid = ca.create_channel(name)
                ca.connect_channel(chid)
                chids.append((chid, name))
                ca.poll(evt=0.025, iot=5.0)
            ca.poll(evt=0.05, iot=10.0)

            values = {}
            for chid, name in chids:
                values[name] = ca.get(chid, as_string=True)

            for promotion in ('ctrl', 'time'):
                for chid, pvname in chids:
                    write('=== %s  chid=%s as %s' % (ca.name(chid), repr(chid),
                                                     promotion))
                    time.sleep(0.01)
                    if promotion == 'ctrl':
                        ntype = ca.promote_type(chid, use_ctrl=True)
                    else:
                        ntype = ca.promote_type(chid, use_time=True)

                    val  = ca.get(chid, ftype=ntype)
                    cval = ca.get(chid, as_string=True)
                    if ca.element_count(chid) > 1:
                        val = val[:12]
                    self.assertEqual(cval, values[pvname])

    def test_type_converions_2(self):
        write("CA type conversions arrays")
        pvlist = (pvnames.char_arr_pv,
                  pvnames.long_arr_pv,
                  pvnames.double_arr_pv)
        with no_simulator_updates():
            chids = []
            for name in pvlist:
                chid = ca.create_channel(name)
                ca.connect_channel(chid)
                chids.append((chid, name))
                ca.poll(evt=0.025, iot=5.0)
            ca.poll(evt=0.05, iot=10.0)

            values = {}
            for chid, name in chids:
                values[name] = ca.get(chid)
            for promotion in ('ctrl', 'time'):
                for chid, pvname in chids:
                    write('=== %s  chid=%s as %s' % (ca.name(chid), repr(chid),
                                                     promotion))
                    time.sleep(0.01)
                    if promotion == 'ctrl':
                        ntype = ca.promote_type(chid, use_ctrl=True)
                    else:
                        ntype = ca.promote_type(chid, use_time=True)

                    val  = ca.get(chid, ftype=ntype)
                    cval = ca.get(chid, as_string=True)
                    for a, b in zip(val, values[pvname]):
                        self.assertEqual(a, b)


    def test_Array0(self):
        write('Array Test: get double array as numpy array, ctypes Array, and list')
        chid = ca.create_channel(pvnames.double_arrays[0])
        aval = ca.get(chid)
        cval = ca.get(chid, as_numpy=False)

        self.assertTrue(isinstance(aval, numpy.ndarray))
        self.assertTrue(len(aval) > 2)

        self.assertTrue(isinstance(cval, ctypes.Array))
        self.assertTrue(len(cval) > 2)
        lval = list(cval)
        self.assertTrue(isinstance(lval, list))
        self.assertTrue(len(lval) > 2)
        self.assertTrue(lval == list(aval))

    def test_xArray1(self):
        write('Array Test: get(wait=False) / get_complete()')
        chid = ca.create_channel(pvnames.double_arrays[0])
        val0 = ca.get(chid)
        aval = ca.get(chid, wait=False)
        self.assertTrue(aval is  None)
        val1 = ca.get_complete(chid)
        self.assertTrue(all(val0 == val1))

    def test_xArray2(self):
        write('Array Test: get fewer than max vals using ca.get(count=0)')
        chid = ca.create_channel(pvnames.double_arrays[0])
        maxpts = ca.element_count(chid)
        npts = int(max(2, maxpts/2.3 - 1))
        write('max points is %s' % (maxpts, ))
        dat = numpy.random.normal(size=npts)
        write('setting array to a length of npts=%s' % (npts, ))
        ca.put(chid, dat)
        out1 = ca.get(chid)
        self.assertTrue(isinstance(out1, numpy.ndarray))
        self.assertEqual(len(out1), npts)
        out2 = ca.get(chid, count=0)
        self.assertTrue(isinstance(out2, numpy.ndarray))
        self.assertEqual(len(out2), npts)

    def test_xArray3(self):
        write('Array Test: get char array as string')
        chid = ca.create_channel(pvnames.char_arrays[0])
        val = ca.get(chid)
        self.assertTrue(isinstance(val, numpy.ndarray))
        char_val = ca.get(chid, as_string=True)
        self.assertTrue(isinstance(char_val, str))
        conv = ''.join([chr(i) for i in val])
        self.assertEqual(conv, char_val)


if __name__ == '__main__':
    suite = unittest.TestLoader().loadTestsFromTestCase( CA_BasicTests)
    unittest.TextTestRunner(verbosity=1).run(suite)


#     chid = ca.create_channel(pvnames.int_pv,
#                              callback=onConnect)
#
#     time.sleep(0.1)
# ;