data_validation.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Created on Mon Jun 26 15:02:59 2023

@author: hhacot, modified by vslonosky
"""

import argparse
import db_connection as db
import glob2
import json
from matplotlib import pyplot as plt
import pandas as pd
import pymysql


plt.rcParams['backend'] = "MacOSX"

#sefField="p"
#sefCfg="config_Kingston.json"
sampleSize=100

# Load SEF File into a dataframe
def parseSef(cfg):
    # Find SEF file and load it into a dataframe
    sef_pattern=cfg["site"]["source"]+"_"+cfg["site"]["org"]+"_"+cfg["site"]["name"]+"_*-"+args.f[0]+".tsv"
    files = glob.glob(sef_pattern)
    if files:
        return pd.read_csv(files[0],sep='\t',header=12)
    return None

# Plotting outliers
def plotOutliers(outliers,ans,x,y,std_count,standard_deviation,outlierFile):
    ans_max=ans+std_count*standard_deviation
    ans_min=ans-std_count*standard_deviation
    fig, ax =plt.subplots(1, figsize=(20, 8))
    fig.autofmt_xdate()
    ax.plot( x, y, '.', color='black', label="data")
    ax.plot(x, ans, '--', color="blue", label="rolling mean")
    ax.plot(x,ans_min, '-', color='green', label=str(std_count)+" * sigma")
    ax.plot(x,ans_max, '-', color='green')
    ax.plot(outliers.observation_date, outliers.Value, 'o',color='red', label='Outliers')
    ax.set_title(" Field: "+args.f[0])
    ax.legend()
    #plt.show()
    figname=outlierFile.name[:-4]+"_"+x.array[0].strftime("%Y-%m-%d")+"_"+x.array[-1].strftime("%Y-%m-%d")+".jpg"
    fig.savefig(figname)

# Print link to outlier fixing
def saveOutliers(outliers, outlierFile, myCursor,cfg):
    url="https://eccc.opendatarescue.org/en/transcriptions/"
    for index,outlier in outliers.iterrows():
        orig_value=outlier.Meta.split()[0][5:]
        sql_query=" SELECT a.transcription_id from data_entries de join annotations a on a.id=de.annotation_id  where  de.value='"+ orig_value +\
                   "' and a.observation_date LIKE '"+outlier.observation_date.strftime("%Y-%m-%d %")+"' "+\
                   " and a.page_id in (select id from pages where title like '"+cfg["site"]["prefix"]+"%') "
        #print(sql_query)
        myCursor.execute(sql_query)
        results=myCursor.fetchall()
        for result in results:
            if result is not None:
                trans=result[0]
                outlierFile.write(url+str(trans)+"/edit,"+orig_value+","+str(outlier.observation_date)+"\n")


# Identify outliers
def findOutliers(df,outlierFile,myCursor,cfg):
    list_partial=[]
    obs_date=None
    days_gap_allowed=5
    std_count=5
    for index,row in df.iterrows():
        new_observation_date=row['observation_date']
        #print(new_observation_date)
        if row.Value == -999 or row.Value == -888:
            pass
        elif (obs_date==None or (new_observation_date-obs_date).days < days_gap_allowed) and len(list_partial)<1000:
            list_partial.append(row)
            obs_date=new_observation_date
            print(".",end='')
        else:
            print("\nEvaluating list_partial because gap is " + str((new_observation_date-obs_date).days)+" and size is "+str(len(list_partial)))
            df_partial=pd.DataFrame(list_partial)
            x=df_partial.observation_date
            y=df_partial.Value
            if y.size>6:
                ans=y.rolling(5,center=True).mean()
                delta=(y-ans).abs()
                standard_deviation=delta.std()
                outliers=df[df.index.isin(delta[delta.gt(std_count*standard_deviation)].index)]
                #print(outliers)
                if outliers.size >0:
                    plotOutliers(outliers,ans,x,y,std_count,standard_deviation,outlierFile)
                    saveOutliers(outliers, outlierFile, myCursor,cfg)

            list_partial=[]
            obs_date=new_observation_date
            list_partial.append(row)



parser = argparse.ArgumentParser(description='Validates SEF data.')
parser.add_argument('config', metavar='config', type=str, nargs=1,
                     help='json config file for each station')
parser.add_argument('-f', metavar='field', type=str, nargs=1,
                    help='SEF field')
args = parser.parse_args()

# open config file
with open(args.config[0]) as json_data_file:
#json_data_file = sefCfg
    cfg = json.load(json_data_file)	 # cfg is the json config file

mycursor=db.connect(cfg)

df=parseSef(cfg)
if df is None:
    print("No SEF file has been found.")
else:
    #Add an observation date column that concatenates the Year Month Day Hour Minute fields for easier manipulation
    df['observation_date']= pd.to_datetime(df[['Day','Month','Year', 'Hour', 'Minute']]
                       .astype(str).apply(' '.join, 1), format='%d %m %Y %H %M')

    outlierFile = open(cfg["site"]["source"]+"_"+cfg["site"]["org"]+"_"+cfg["site"]["name"]+"-"+args.f[0]+"-outliers.csv","w",buffering=1)


    findOutliers(df,outlierFile,mycursor,cfg)
    outlierFile.close()