from __future__ import print_function
import os
from hdfs import Config
from hdfs import InsecureClient
import sys
from operator import add
from pyspark import SparkConf
from datetime import datetime
from pyspark.sql import SparkSession
from pyspark import SparkContext
from Transportation import *

class Spark():
    def __init__(self):
        self.sc = None
        self.sorted_user_record_list = None
    def connect(self):
        conf=SparkConf()
        conf.setMaster("spark://namenode:7077")
        conf.setAppName("off-peak")
        conf.set("spark.executor.memory", "128g")
        conf.set("spark.executor.cores", "16")
        conf.set("spark.scheduler.mode", "FAIR")
        sc = SparkContext(conf=conf)
        sc.addPyFile("Module/Shapely-1.6b4.zip")
        sc.addFile("exploring/Spark.py")
        sc.addFile("exploring/Transportation.py")
        sc.addFile("exploring/Record.py")
        sc.addFile("exploring/TransRegions.py")
        sc.addFile("exploring/__init__.py")
        self.sc = sc

class SubwaySpark(Spark):
    def __init__(self):
        self.file_path = None
        self.input_data = None
        self.subway = Subway()
        self.sorted_record_list = None
        self.user_trip_pair = None
        self.trip_average_time  = None
        self.in_vehicle_time = None
        self.connect()
    def setLocalInputFile(self,inputFile):
        self.input_data = self.sc.parallelize(inputFile)
    def setInputData(self,input_data):
        self.input_data = input_data
    def setLocalFilePath(self,file_path):
        self.file_path = file_path
        self.setLocalInputFile(open(self.file_path))
    def setHDFSFilePath(self,file_path):
        self.file_path = file_path
        self.setInputData(self.sc.textFile(self.file_path).cache())

    def buildRecordList(self):
        '''
        build the record list for each user and sort it by the time
        :return:
        '''
        one_file = self.input_data
        subway_record_list = one_file.map(self.subway.subwayOneRow).filter(lambda x: x is not None)#.map(lambda record: print(record.station_id))
        user_record_list = subway_record_list.map(lambda x: (x.user_id,x)).groupByKey()
        self.sorted_user_record_list = user_record_list.mapValues(list).filter(lambda (k,v): len(v) > 1).map(lambda (k,v): (k, sorted(v, key=lambda record: record.time)))

    def buildTripList(self):
        self.user_trip_pair = self.sorted_user_record_list.map(self.subway.sortedRecordsToTrip).filter(lambda (k,v): len(v)>0)

    def buildInVehicleTime(self):
        trip_time_pair = self.user_trip_pair.flatMap(self.subway.mapToStationTimeMapping)
        self.in_vehicle_time = trip_time_pair.reduceByKey(self.subway.minimumByKey)
        #self.in_vehicle_time.saveAsTextFile('/zf72/transportation_data/result/in_vehicle_time/0601')

    def buildWaitingTime(self):
        #read in_vehicle_time mapping
        user_trips_in_vehicle_time = self.user_trip_pair.flatMap(self.subway.mapToODTrip).join(self.in_vehicle_time)
        timeslot_waiting_time = user_trips_in_vehicle_time.map(self.subway.mapToTripWaitingTime).map(lambda (k,v1,v2): (k,v2))
        self.timeslot_ave_waiting_time = timeslot_waiting_time.combineByKey(
            lambda v: (v,1),
            lambda x,v: (x[0]+v,x[1]+1),
            lambda x,y: (x[0]+y[0],x[1]+y[1])
        ).map(
            lambda (k,v): (k,float(v[0])/float(v[1]))
        )
        result_json = {}
        for record in self.timeslot_ave_waiting_time.collect():
            result_json[record[0]] = record[1]
        print(result_json)
        #json.dump(result_json,open('../../data/result/waiting_time.json'))
        # self.user_trip_pair.map(
        #     lambda (k,v):
        # )
        # self.user_trip_pair.join(self.in_vehicle_time)

    def buildTripAveTimeMatrix(self):
        trip_time_pair = self.user_trip_pair.flatMap(self.subway.mapToStationTimeMapping)
        trip_time_count = trip_time_pair.combineByKey(
            lambda v: (v,1),
            lambda x,v: (x[0]+v,x[1]+1),
            lambda x,y: (x[0]+y[0],x[1]+y[1])
        )
        self.trip_average_time = trip_time_count.map(
            lambda (k,v): (k,float(v[0])/float(v[1]))
        )


    def saveAverageTripTime(self,output_file_path,local=False):
        if not local:
            self.trip_average_time.saveAsTextFile(output_file_path)
        else:
            result = self.trip_average_time.collect()
            print(result)
    def waitingTimeOneDay(self):
        pass
class Bus(Spark):
    def __init__(self):
        pass