# ---
# jupyter:
#   jupytext:
#     text_representation:
#       extension: .py
#       format_name: light
#       format_version: '1.5'
#       jupytext_version: 1.14.4
#   kernelspec:
#     display_name: Python 2
#     language: python
#     name: python2
# ---

# + deletable=true editable=true
# # cd ~/datafiles
# # mkdir wind2014
# # cd wind2014
# wget http://freo.me/1LpKbGV -O wd2014.zip
# unzip wd2014.zip
# # rm wd2014.zip


# dt = parse("datestring")  # returns dateime.datetime
# def date_and_hour(s):
#     dt = parse(s.replace('?',''))
#     hour = dt.hour
#     return (dt.strftime("%Y-%m-%d"), hour)


# Incidents Schema
# |IncidntNum|Category|Descript|DayOfWeek|Date|Time|PdDistrict|Resolution|Address|X|Y|Location|PdId|
# Wind Schema
# |Station_ID|Station_Name|Location_Label|Interval_Minutes|Interval_End_Time|Wind_Velocity_Mtr_Sec|
#     Wind_Direction_Variance_Deg|Wind_Direction_Deg|Ambient_Temperature_Deg_C|Global_Horizontal_Irradiance|

# + deletable=true editable=true


# + deletable=true editable=true
from pyspark.sql import SQLContext
sqlc = SQLContext(sc)

df = sqlc.read.csv('file:///home/oxclo/datafiles/wind2014/*.csv', header='true', inferSchema='true')
# df.show(1)

# + deletable=true editable=true
from dateutil.parser import parse
from datetime import datetime
from pyspark.sql.functions import regexp_replace
from pyspark.sql.functions import to_date
from pyspark.sql.functions import hour

from pyspark.sql import SQLContext
sqlc = SQLContext(sc)

df = sqlc.read.csv('file:///home/oxclo/datafiles/wind2014/*.csv', header='true', inferSchema='true')

df = df.withColumn('Interval_End_Time', regexp_replace(df.Interval_End_Time, '(.*)\\? (.*)', '$1 $2'))
df = df.withColumn('zdate', to_date(df.Interval_End_Time)).withColumn('zhour', hour(df.Interval_End_Time))

df = df.filter("Wind_Velocity_Mtr_Sec > 0.0 and Ambient_Temperature_Deg_C > 0.0")\
.dropna(subset=['Wind_Velocity_Mtr_Sec', "Ambient_Temperature_Deg_C", "Interval_End_Time", "Interval_Minutes"])\
.withColumnRenamed('Station_ID', 'station')\
.withColumnRenamed('Wind_Velocity_Mtr_Sec', 'vel')\
.withColumnRenamed('Ambient_Temperature_Deg_C', 'deg')

df = df.select(df.station, df.zdate, df.zhour, df.vel, df.deg).orderBy(df.station, df.zdate, df.zhour)
df = df.groupBy(df.station, df.zdate, df.zhour).avg('vel', 'deg').orderBy(df.station, df.zdate, df.zhour)
df.show(10)


# + deletable=true editable=true
from scipy.spatial import KDTree
from pyspark.sql.functions import regexp_extract
from pyspark.sql.functions import to_date
from pyspark.sql.functions import year
from pyspark.sql.functions import from_unixtime
from pyspark.sql.functions import unix_timestamp
from pyspark.sql.functions import udf
from pyspark.sql.functions import lit
from pyspark.sql.functions import count
from pyspark.sql import SQLContext
sqlc = SQLContext(sc)

sfpds = sqlc.read.csv('file:///home/oxclo/datafiles/incidents/sfpd.csv', header='true', inferSchema='true')
sfpds = sfpds.withColumn('zdate', to_date(from_unixtime(unix_timestamp(sfpds.Date, 'MM/dd/yyy'))))\
.withColumn('zhour', regexp_extract(sfpds.Time, '(\d+):(\d+)', 1))

sfpds = sfpds.filter(year(sfpds.zdate) == 2014)

# NOTE: stations and their locations info from http://freo.me/oxclo-locate
def to_station(y,x):
    station_locations = [\
    [37.7816834,-122.3887657],[37.7469112,-122.4821759],[37.7411022,-120.8041510],[37.4834543,-122.3187302],\
    [37.7576436,-122.3916382],[37.7970013,-122.4140409],[37.7484960,-122.4567461],[37.7288155,-122.4210133],\
    [37.5839487,-121.9499339],[37.7157156,-122.4145311],[37.7329613,-122.5051491],[37.7575891,-122.3923824],\
    [37.7521169,-122.4497687]]
    stations = ["SF18", "SF04", "SF15", "SF17", "SF36", "SF37","SF07", "SF11", "SF12", "SF14", "SF16", "SF19", "SF34"]
    tree = KDTree(station_locations)
    return stations[tree.query([y, x])[1]]

kdtree_udf = udf(to_station)
sfpds = sfpds.withColumn('station', kdtree_udf(sfpds.Y, sfpds.X))

sfpds = sfpds.select(sfpds.station, sfpds.zdate, sfpds.zhour)\
.groupBy(sfpds.station, sfpds.zdate, sfpds.zhour)\
.agg(count('*').alias('incidents'))\
.orderBy('incidents', ascending=False)
sfpds.show(15)

# + deletable=true editable=true
from pyspark.mllib.linalg import Vectors
from pyspark.mllib.stat import Statistics

cond = [df.station == sfpds.station, df.zdate == sfpds.zdate, df.zhour == sfpds.zhour]
joined = df.join(sfpds, cond)\
.select(df.station, df.zdate, df.zhour, 'avg(deg)', 'avg(vel)', sfpds.incidents)\
.withColumnRenamed('avg(deg)', 't')\
.withColumnRenamed('avg(vel)', 'w')\
.withColumnRenamed('incidents', 'i')\
.orderBy(sfpds.incidents, ascending=False)

joined.show(15)

# remap the data into a vector of [t,w,i]
vecs = joined.rdd.map(lambda x: Vectors.dense(x.t, x.w, x.i))
print(Statistics.corr(vecs))

StarcoderdataJupyterScriptsDedupFiltered