Merge pull request #14 from olincollege/SV-25-traffic-lights

Sv 25 traffic lights

docs/source/explanations/datasets.rst

@@ -17,6 +17,8 @@ Since our project is focused on pedestrian safety at nighttime on crosswalks, we
 
 To estimate the danger of a road, we also need a dataset with information about speed limits per road. MassDOT provides a database (updated yearly) with speed limits for most roads in Massachusetts. This can be viewed and downloaded at `this link <https://geodot-massdot.hub.arcgis.com/search?groupIds=362562c527bb404884dd1608b4bfdb62>`
 
+To get a list of traffic lights, which have a correlation with safer crosswalks, we look into BostonMap's GIS. Although MassDOT does provide a database for the state of Massachusetts, this is pretty sparse and less useful. The dataset used can be found at `this link <https://bostonopendata-boston.opendata.arcgis.com/datasets/boston::traffic-signals/about>`
+
 Population Density Dataset
 --------------------------
 

tests/conftest.py

@@ -1,5 +1,4 @@
 from night_light.utils import query_geojson
-from night_light.road_characteristics import traffic_speed_data
 
 import pytest
 import asyncio
@@ -11,7 +10,11 @@
     "https://gis.massdot.state.ma.us/arcgis/rest/services/Roads/VMT/FeatureServer/10/query"
 )
 MA_MUNICIPALITIES_URL = "https://arcgisserver.digital.mass.gov/arcgisserver/rest/services/AGOL/Towns_survey_polym/FeatureServer/0/query"
-BOSTON_STREETLIGHTS_URL = "https://services.evari.io/evari/rest/services/Boston/Boston_Read_Only/MapServer/0/query"
+BOSTON_STREETLIGHTS_URL = (
+    "https://services.evari.io/evari/rest/services/Boston/Boston_Read_Only/MapServer/0/query"
+)
+
+BOSTON_TRAFFIC_LIGHTS_URL = "https://gisportal.boston.gov/arcgis/rest/services/Infrastructure/OpenData/MapServer/12/query?outFields=*&where=1%3D1&f=geojson"
 
 MUNICIPALITIES_MA_PARAMS = {
     "where": "1=1",
@@ -101,12 +104,14 @@ async def fetch_all_data():
                     features = result.get("features", [])
                     data.extend(features)
                     feature_count = len(features)
-                    if (
-                        not result.get("exceededTransferLimit")
-                        or feature_count < batch_size
-                    ):
+                    if not result.get("exceededTransferLimit") or feature_count < batch_size:
                         return data
 
     data = asyncio.run(fetch_all_data())
     gdf = gpd.GeoDataFrame.from_features(data, crs="EPSG:4326")
     return gdf
+
+
+@pytest.fixture
+def boston_traffic_lights():
+    return query_geojson.fetch_geojson_data(url=BOSTON_TRAFFIC_LIGHTS_URL, params={})

tests/test_boston_traffic_lights_query.py

@@ -0,0 +1,37 @@
+import os
+from night_light.utils import query_geojson
+
+
+def test_query_boston_traffic_light(boston_traffic_lights):
+    """
+    Test querying Boston traffic light data
+    """
+    # assert boston_traffic_lights.shape[0] > 0
+    assert boston_traffic_lights.geometry.name == "geometry"
+    assert set(boston_traffic_lights.geometry.geom_type.unique()) == {"Point"}
+    assert boston_traffic_lights.columns.to_list() == [
+        "geometry",
+        "OBJECTID",
+        "Count_",
+        "Int_Number",
+        "Location",
+        "Dist",
+    ]
+    assert boston_traffic_lights["Dist"].unique().size == 10
+
+
+def test_save_boston_traffic_light(boston_traffic_lights):
+    """
+    Test saving Boston traffic light data to GeoJSON
+    """
+    geojson_name = "test_boston_traffic_lights.geojson"
+    query_geojson.save_geojson(boston_traffic_lights, geojson_name)
+    saved_gdf = query_geojson.gpd.read_file(geojson_name)
+
+    assert boston_traffic_lights.crs == saved_gdf.crs
+    assert set(boston_traffic_lights.columns) == set(saved_gdf.columns)
+    assert boston_traffic_lights.index.equals(saved_gdf.index)
+    assert boston_traffic_lights.shape == saved_gdf.shape
+
+    os.remove(geojson_name)
+    assert not os.path.exists(geojson_name)