Get Started with CLPM

'''
Created on Aug 30, 2018

@author: 
'''
from pyspark.sql import SparkSession
from pyspark.sql import *
from pyspark.sql import functions
from pyspark.sql.types import *
from pyspark.sql.dataframe import *
from pyspark.sql.functions import *
import sys
import time
from datetime import datetime

class ControlModeScript():
    # ################ DO NOT EDIT ########################
    MANUAL = "1.0"
    AUTO = "2.0"
    CASCADE = "3.0"
    SHUTDOWN = "4.0"
    QUALITY_GOOD = "3"
    QUALITY_BAD = "0"
    # ################ END ########################
    
    # ################ MODIFY THIS FUNCTION ########################
    # When modifying this function, TQVs for the custom tags (Tag_VV, Tag_XX, Tag_YY, Tag_ZZ) can be obtained as follows:
    #     * timestamp is common amongst all tags   
    #     * quality_XX is quality of Tag_XX, quality_YY is quality of Tag_YY and so on
    #     * value_XX is value of Tag_XX, value_YY is value of Tag_YY and so on
    #
    def control_mode_query(self):
        
        # The example SQL query below does the following:
        # select timestamp as the timestamp of the tag
        # always select quality as good
        # if quality is bad
        #    select value as shutdown
        # else if value of tag_XX is 1
        #    select value as MANUAL
        # else if value of tag_YY is 1
        #    select value as AUTO
        # else if value of tag_ZZ is 1
        #    select value as CASCADE
        # else
        #    select value as SHUTDOWN
        string = "timestamp as timestamp, CASE WHEN quality_XX != 3 OR quality_YY != 3 OR quality_ZZ != 3 OR quality_VV != 3 THEN " + self.SHUTDOWN
        string += " WHEN value_XX = 1 THEN "+self.MANUAL+" WHEN value_YY = 1 THEN "+self.AUTO+" WHEN value_ZZ = 1 THEN "+self.CASCADE+" WHEN value_VV = 1 THEN "+self.SHUTDOWN+" ELSE "+self.SHUTDOWN+" END as value," + self.QUALITY_GOOD +" as quality"         
        return string
    # ################ END ########################
    
    # ################ DO NOT EDIT ########################
    def run_job(self, spark_session, runtime_config, job_json, context_dict, logger):
            try:
                spark = spark_session
                logger.info("Starting analytic...")               
                configContext = context_dict["configDS"]
                tsContext = context_dict["timeseriesReadDS"]
                configDF = configContext.sql("select * from " + context_dict["configDS"].table_name)
                configDF.createOrReplaceTempView("configDF")
                tsDF = tsContext.sql("select * from " + context_dict["timeseriesReadDS"].table_name)
                tsDF.createOrReplaceTempView("timeseriesReadDF")
                
                Tag_XXDDF = tsContext.sql("SELECT c.AssetSourceKey as asset, t.timestamp as timestamp, t.value as value_XX, t.quality as quality_XX FROM `timeseriesReadDF`"
                    + " t JOIN `configDF` c on t.tag = c.MappingValue WHERE c.MappingKey = 'Tag_XX' DISTRIBUTE BY asset")
                Tag_XXDDF.createOrReplaceTempView("tag_XXDDF")

                 
                Tag_YYDDF = tsContext.sql("SELECT c.AssetSourceKey as asset, t.timestamp as timestamp, t.value as value_YY, t.quality as quality_YY FROM `timeseriesReadDF" 
                                             + "` t JOIN `configDF` c on t.tag = c.MappingValue WHERE c.MappingKey = 'Tag_YY' DISTRIBUTE BY asset")
                Tag_YYDDF.createOrReplaceTempView("tag_YYDDF")
             
                Tag_ZZDDF = tsContext.sql("SELECT c.AssetSourceKey as asset, t.timestamp as timestamp, t.value as value_ZZ, t.quality as quality_ZZ FROM `timeseriesReadDF" 
                                             + "` t join `configDF` c on t.tag = c.MappingValue WHERE c.MappingKey = 'Tag_ZZ' DISTRIBUTE BY asset")
                Tag_ZZDDF.createOrReplaceTempView("tag_ZZDDF")
 
                Tag_VVDDF = tsContext.sql("SELECT c.AssetSourceKey as asset, t.timestamp as timestamp, t.value as value_VV, t.quality as quality_VV FROM `timeseriesReadDF" 
                                             + "` t JOIN `configDF` c on t.tag = c.MappingValue WHERE c.MappingKey = 'Tag_VV' DISTRIBUTE BY asset")
                Tag_VVDDF.createOrReplaceTempView("tag_VVDDF")
                
                Tag_XXYYDF =  tsContext.sql("SELECT COALESCE(tag_XXDDF.asset, tag_YYDDF.asset) as asset,COALESCE(tag_XXDDF.timestamp,tag_YYDDF.timestamp) as timestamp, tag_XXDDF.value_XX, tag_XXDDF.quality_XX, tag_YYDDF.quality_YY,tag_YYDDF.value_YY FROM tag_XXDDF FULL OUTER JOIN tag_YYDDF ON tag_XXDDF.asset = tag_YYDDF.asset and tag_XXDDF.timestamp = tag_YYDDF.timestamp DISTRIBUTE BY asset")
                Tag_ZZVVDF =  tsContext.sql("SELECT COALESCE(tag_ZZDDF.asset, tag_VVDDF.asset) as asset,COALESCE(tag_ZZDDF.timestamp,tag_VVDDF.timestamp) as timestamp,tag_ZZDDF.quality_ZZ, tag_ZZDDF.value_ZZ, tag_VVDDF.quality_VV,tag_VVDDF.value_VV FROM tag_ZZDDF FULL OUTER JOIN tag_VVDDF ON tag_ZZDDF.asset = tag_VVDDF.asset and tag_ZZDDF.timestamp = tag_VVDDF.timestamp DISTRIBUTE BY asset")
                Tag_XXYYDF.createOrReplaceTempView("tag_XXYYDF")
                Tag_ZZVVDF.createOrReplaceTempView("tag_ZZVVDF")
                timeseriesDF =  tsContext.sql("SELECT COALESCE(tag_XXYYDF.asset,tag_ZZVVDF.asset) as asset, COALESCE(tag_XXYYDF.timestamp,tag_ZZVVDF.timestamp) as timestamp, tag_XXYYDF.quality_XX, tag_XXYYDF.value_XX, tag_XXYYDF.quality_YY,tag_XXYYDF.value_YY, tag_ZZVVDF.quality_ZZ,tag_ZZVVDF.value_ZZ, tag_ZZVVDF.quality_VV,tag_ZZVVDF.value_VV FROM tag_XXYYDF FULL OUTER JOIN tag_ZZVVDF ON tag_XXYYDF.asset = tag_ZZVVDF.asset and tag_XXYYDF.timestamp = tag_ZZVVDF.timestamp")
                timeseriesDF.createOrReplaceTempView("collectedDS")
                
                queryString = "SELECT configDF.MappingValue as tag, "
                queryString +=  self.control_mode_query()
                queryString += " FROM `collectedDS` JOIN `configDF` on collectedDS.asset = configDF.assetSourceKey WHERE configDF.mappingType = 'OutputMappings'"
                resultDF = tsContext.sql(queryString)
                resultDF.createOrReplaceTempView("resultDF")
                timeseriesWriteDF = tsContext.sql("SELECT tag, timestamp, CAST((value) as double) as value, quality FROM `resultDF`")
                logger.info("Returning result...")
                result = {"timeseriesWriteDS" : timeseriesWriteDF}
                return result
 
            except Exception as e:
                print("ERROR RETURNED")
                logger.info("Error: " + str(e))
                exc_tb = sys.exc_info()[2]
                logger.info("Line number: " + str(exc_tb.tb_lineno))
    # ################ END ########################                 
                

• Tag_VV
• Tag_XX
• Tag_YY
• Tag_ZZ

def control_mode_query(self):
        string = "timestamp as timestamp, CASE WHEN quality_XX != 3 OR quality_YY != 3 OR quality_ZZ != 3 OR quality_VV != 3 THEN " + self.SHUTDOWN
        string += " WHEN value_XX = 1 THEN "+self.MANUAL+" WHEN value_YY = 1 THEN "+self.AUTO+" WHEN value_ZZ = 1 THEN "+self.CASCADE+" WHEN value_VV = 1 THEN "+self.SHUTDOWN+" ELSE "+self.SHUTDOWN+" END as value," + self.QUALITY_GOOD +" as quality"         
        return string