Collisions Study - Pipeline¶

import urban_mapper as um
from urban_mapper.pipeline import UrbanPipeline

mapper = um.UrbanMapper()

# Define the pipeline
pipeline = UrbanPipeline([
    ("urban_layer", (
        mapper
        .urban_layer
        .with_type("streets_intersections")
        .from_place("Downtown Brooklyn, New York City, USA", network_type="drive")
        .with_mapping(
            longitude_column="LONGITUDE",
            latitude_column="LATITUDE",
            output_column="nearest_intersection"
        )
        .build()
    )),
    # Note: For the documentation interactive mode, we only query 5000 records from the dataset.  Feel free to remove for a more realistic analysis.
    ("loader", (
        mapper
        .loader
        .from_huggingface("oscur/NYC_vehicle_collisions", number_of_rows=5000, streaming=True)
        .with_columns(longitude_column="LONGITUDE", latitude_column="LATITUDE")
        .build()
    )),
    ("imputer", (
        mapper
        .imputer
        .with_type("SimpleGeoImputer")
        .on_columns("LONGITUDE", "LATITUDE")
        .build()
    )),
    ("filter", um.UrbanMapper().filter.with_type("BoundingBoxFilter").build()),
    ("enricher", (
        mapper
        .enricher
        .with_data(group_by="nearest_intersection")
        .count_by(output_column="collision_count")
        .build()
    )),
    ("visualiser", (
        mapper
        .visual
        .with_type("Interactive")
        .with_style({"tiles": "CartoDB Positron", "colorbar_text_color": "gray"})        
        .build()
    ))
])

# Execute the pipeline
mapped_data, enriched_layer = pipeline.compose_transform()

# Visualize results
pipeline.visualise(["collision_count"])

# Save the pipeline
pipeline.save("./collisions_pipeline.dill")