NYCCrimesAndTaxis/crime_data_engineering_flow.py

import geopandas as gpd
from prefect import task, flow
from typing import Any, Dict, List

import pandas as pd

@task
def load_crimes(path: str) -> pd.DataFrame:
    crime_dat=pd.read_csv(path)
    return crime_dat

@task
def convert_date_crime(crime_dat: pd.DataFrame)-> pd.DataFrame:
    crime_dat['Date']=pd.to_datetime(crime_dat['ARREST_DATE'], format="%m/%d/%Y")
    crime_dat['DayOfWeekNum']=crime_dat['Date'].dt.dayofweek
    crime_dat['DayOfWeek'] = crime_dat['Date'].dt.day_name()
    crime_dat.drop('ARREST_DATE', axis=1)
    return crime_dat

@task
def add_boroughs_crime(crime_dat: pd.DataFrame)->pd.DataFrame:
    # Load the borough polygons
    gdf = gpd.read_file(gpd.datasets.get_path("nybb"))

    # Create a GeoDataFrame of points from the crime data
    point = gpd.points_from_xy(crime_dat['Longitude'], crime_dat['Latitude'], crs='EPSG:4326')
    gdf_point = gpd.GeoDataFrame(geometry=point)

    # Set the CRS of the GeoDataFrame to WGS84
    gdf_point = gdf_point.set_crs(epsg=4326)

    # Reproject the GeoDataFrame to EPSG:2263
    gdf_point = gdf_point.to_crs(epsg=2263)

    # spatial join crime coordinates to boroughs
    point_in_borough = gpd.tools.sjoin(gdf_point, gdf, predicate="within", how='left')
    crime_dat['Borough'] = point_in_borough['BoroName']

    return crime_dat

@task
def save_output_crime(crime_dat: pd.DataFrame):
    crime_dat.to_csv('data/crime.csv')

@flow
def crime_data_eng_pipeline():
    crime_dat=load_crimes('data/NYPD_Arrest_Data__Year_to_Date_.csv')
    crime_dat=convert_date_crime(crime_dat)
    crime_dat=add_boroughs_crime(crime_dat)
    save_output_crime(crime_dat)
    
if __name__=="__main__":   
    crime_data_eng_pipeline()