__author__ = 'chunk'

from ..mspark import SC

from ..common import *
from ..mdata import ILSVRC, ILSVRC_S

from pyspark.mllib.regression import LabeledPoint

import happybase

def test_whole():
    cols0 = [
        'cf_pic:data',
        'cf_info:width',
        'cf_info:height',
        'cf_info:size',
        'cf_info:capacity',
        'cf_info:quality',
        'cf_info:rate',
        'cf_tag:chosen',
        'cf_tag:class'
    ]
    cols1 = [
        'cf_pic:data',
        'cf_info:width',
        'cf_info:height',
        'cf_info:size',
        'cf_info:capacity',
        'cf_info:quality',
        'cf_info:rate',
        'cf_tag:chosen',
        'cf_tag:class',
        'cf_feat:bid',
    ]

    sparker = SC.Sparker(host='HPC-server', appname='ImageILSVRC-S', master='spark://HPC-server:7077')

    # rdd_data = sparker.read_hbase("ILSVRC2013_DET_val-Test_1", func=SC.rddparse_data_ILS, collect=False) \
    # .mapValues(lambda data: [data] + SC.rddinfo_ILS(data)) \

    # .flatMap(lambda x: SC.rddembed_ILS_EXT(x, rate=0.2)) \

    # .mapValues(lambda items: SC.rddfeat_ILS(items))

    rdd_data = sparker.read_hbase("ILSVRC2013_DET_val-Test_1", func=SC.rddparse_data_ILS, collect=False).mapValues(

        lambda data: [data] + SC.rddinfo_ILS(data))
    rdd_data_ext = rdd_data.map(lambda x: SC.rddembed_ILS(x, rate=0.2)).filter(lambda x: x != None)

    rdd_data = rdd_data.union(rdd_data_ext).mapValues(lambda items: SC.rddfeat_ILS(items))

    print len(rdd_data.collect())

    # sparker.write_hbase("ILSVRC2013_DET_val-Test_1", rdd_data, fromrdd=True, columns=cols1,

    # withdata=True)

def test_whole_ext(category='Train_100'):
    timer = Timer()

    print '[time]category:', category

    print '[time]formating table...'
    timer.mark()

    dil = ILSVRC.DataILSVRC(base_dir='/data/hadoop/ImageNet/ILSVRC/ILSVRC2013_DET_val', category=category)

    dil.delete_table()

    # dil.format()

    dil.store_img()
    timer.report()

    print '[time]reading table...'
    timer.mark()
    table_name = dil.table_name

    connection = happybase.Connection('HPC-server')
    tables = connection.tables()
    if table_name not in tables:

        families_compressed = {'cf_pic': dict(max_versions=2, compression='LZO'),
                               'cf_info': dict(max_versions=2, compression='LZO'),
                               'cf_tag': dict(max_versions=2, compression='LZO'),
                               'cf_feat': dict(max_versions=2, compression='LZO'),
                               }
        families = {'cf_pic': dict(max_versions=1),
                    'cf_info': dict(max_versions=1),
                    'cf_tag': dict(max_versions=1),
                    'cf_feat': dict(max_versions=1),

                    }
        connection.create_table(name=table_name, families=families)
    table = connection.table(name=table_name)

    cols = ['cf_pic:data']
    list_data = []
    for key, data in table.scan(columns=cols):

        data = data['cf_pic:data']
        list_data.append((key, data))

    timer.report()

    print '[time]processing...'
    timer.mark()

    sparker = SC.Sparker(host='HPC-server', appname='ImageILSVRC-S', master='spark://HPC-server:7077')

    rdd_data = sparker.sc.parallelize(list_data, 30) \

        .mapValues(lambda data: [data] + SC.rddinfo_ILS(data)) \

        .flatMap(lambda x: SC.rddembed_ILS_EXT(x, rate=0.2)) \

        .mapValues(lambda items: SC.rddfeat_ILS(items))

    items = rdd_data.collect()

    timer.report()

    print '[time]writing table...'
    timer.mark()

    try:
        with table.batch(batch_size=5000) as b:

            for item in items:

                imgname, imginfo = item[0], item[1]
                b.put(imgname,
                      {
                          'cf_pic:data': imginfo[0],
                          'cf_info:width': str(imginfo[1]),
                          'cf_info:height': str(imginfo[2]),
                          'cf_info:size': str(imginfo[3]),
                          'cf_info:capacity': str(imginfo[4]),
                          'cf_info:quality': str(imginfo[5]),
                          'cf_info:rate': str(imginfo[6]),
                          'cf_tag:chosen': str(imginfo[7]),
                          'cf_tag:class': str(imginfo[8]),

                          'cf_feat:ibd': imginfo[9],

                      })
    except ValueError:
        raise

    timer.report()

def test_whole_sp(category='Train_100'):
    timer = Timer()

    print '[time]category:', category

    print '[time]formating table...'
    timer.mark()
    dil = ILSVRC.DataILSVRC(base_dir='/data/hadoop/ImageNet/ILSVRC/ILSVRC2013_DET_val', category=category)
    dil.delete_table()
    # dil.format()
    dil.store_img()
    timer.report()

    table_name = dil.table_name
    feattype = 'ibd'
    cols = [
        'cf_pic:data',
        'cf_info:width',
        'cf_info:height',
        'cf_info:size',
        'cf_info:capacity',
        'cf_info:quality',
        'cf_info:rate',
        'cf_tag:chosen',
        'cf_tag:class',
        'cf_feat:' + feattype,
    ]

    print '[time]processing...'
    timer.mark()
    sparker = SC.Sparker(host='HPC-server', appname='ImageILSVRC-S', master='spark://HPC-server:7077')
    rdd_data = sparker.read_hbase(table_name, func=SC.rddparse_data_ILS,
                                  collect=False, parallelism=30) \
        .mapValues(lambda data: [data] + SC.rddinfo_ILS(data)) \
        .flatMap(lambda x: SC.rddembed_ILS_EXT(x, rate=0.2)) \
        .mapValues(lambda items: SC.rddfeat_ILS(items, feattype))

    print '[time]writing table...'
    sparker.write_hbase(table_name, rdd_data, fromrdd=True, columns=cols,
                        withdata=True)
    timer.report()