Skip to content

Commit

Permalink
chore: Bring up to speed with changes in dev
Browse files Browse the repository at this point in the history
  • Loading branch information
r-leyshon committed Sep 20, 2023
2 parents e847951 + 93bd867 commit 003a99a
Show file tree
Hide file tree
Showing 6 changed files with 297 additions and 38 deletions.
18 changes: 15 additions & 3 deletions src/transport_performance/gtfs/validation.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@
import plotly.io as plotly_io
from pretty_html_table import build_table
import zipfile
import warnings
import pathlib
from typing import Union
from plotly.graph_objects import Figure as PlotlyFigure
Expand Down Expand Up @@ -224,7 +225,9 @@ def print_alerts(self, alert_type="error"):
elif isinstance(msgs, str):
print(msgs)
except KeyError:
print(f"No alerts of type {alert_type} were found.")
warnings.warn(
f"No alerts of type {alert_type} were found.", UserWarning
)

return None

Expand All @@ -236,6 +239,7 @@ def clean_feed(self):
# shows that shapes.txt is optional file.
self.feed = self.feed.clean()
except KeyError:
# TODO: Issue 74 - Improve this to clean feed when KeyError raised
print("KeyError. Feed was not cleaned.")

def viz_stops(
Expand Down Expand Up @@ -273,7 +277,9 @@ def viz_stops(

pre, ext = os.path.splitext(out_pth)
if ext != ".html":
print(f"{ext} format not implemented. Writing to .html")
warnings.warn(
f"{ext} format not implemented. Saving as .html", UserWarning
)
out_pth = os.path.normpath(pre + ".html")

# geoms defence
Expand Down Expand Up @@ -320,7 +326,13 @@ def viz_stops(
m.fit_bounds(m.get_bounds())
m.save(out_pth)
except KeyError:
print("Key Error. Map was not written.")
# KeyError inside of an except KeyError here. This is to provide
# a more detaailed error message on why a KeyError is being raised.
raise KeyError(
"The stops table has no 'stop_code' column. While "
"this is an optional field in a GTFS file, it "
"raises an error through the gtfs-kit package."
)

def _order_dataframe_by_day(
self, df: pd.DataFrame, day_column_name: str = "day"
Expand Down
24 changes: 14 additions & 10 deletions src/transport_performance/utils/raster.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,10 @@

from rioxarray.merge import merge_arrays
from rasterio.warp import Resampling
from transport_performance.utils.defence import (
_check_parent_dir_exists,
_is_expected_filetype,
)


def merge_raster_files(
Expand Down Expand Up @@ -73,8 +77,9 @@ def merge_raster_files(
"""
# defend against case where the provided input dir does not exist
if not os.path.exists(input_dir):
raise FileNotFoundError(f"{input_dir} can not be found")
# add a dummy file to purely check if provided parent directory exists
dummy_path = os.path.join(input_dir, "dummy.txt")
_check_parent_dir_exists(dummy_path, "input_dir", create=False)

# get tif files in directory, ensure some exist and select subset via regex
tif_filepaths = glob.glob(f"{input_dir}/*.tif")
Expand All @@ -101,12 +106,13 @@ def merge_raster_files(
# merge the datasets together
xds_merged = merge_arrays(arrays)

# make output_dir if it does not exist
if not os.path.exists(output_dir):
os.mkdir(output_dir)

# create full filepath for merged tif file and write to disk
# check expected file type and parent dir exists
MERGED_DIR = os.path.join(output_dir, output_filename)
_is_expected_filetype(
MERGED_DIR, "MERGED_DIR", check_existing=False, exp_ext=".tif"
)
_check_parent_dir_exists(MERGED_DIR, "MERGED_DIR", create=True)
xds_merged.rio.to_raster(MERGED_DIR)

# get boundaries of inputs and output raster
Expand Down Expand Up @@ -146,8 +152,7 @@ def sum_resample_file(
"""
# defend against case where the provided input dir does not exist
if not os.path.exists(input_filepath):
raise FileNotFoundError(f"{input_filepath} can not be found")
_is_expected_filetype(input_filepath, "input_filepath", exp_ext=".tif")

xds = rioxarray.open_rasterio(input_filepath, masked=True)

Expand All @@ -161,7 +166,6 @@ def sum_resample_file(
)

# make output_filepath's directory if it does not exist
if not os.path.exists(os.path.dirname(output_filepath)):
os.mkdir(output_filepath)
_check_parent_dir_exists(output_filepath, "output_filepath", create=True)

xds_resampled.rio.to_raster(output_filepath)
39 changes: 19 additions & 20 deletions tests/gtfs/test_validation.py
Original file line number Diff line number Diff line change
Expand Up @@ -259,8 +259,7 @@ def test_unmatched_service_id_behaviour(self, gtfs_fixture):
len(new_valid[new_valid.message == "Undefined service_id"]) == 1
), "gtfs-kit failed to identify missing service_id"

@patch("builtins.print")
def test_print_alerts_defence(self, mocked_print, gtfs_fixture):
def test_print_alerts_defence(self, gtfs_fixture):
"""Check defensive behaviour of print_alerts()."""
with pytest.raises(
AttributeError,
Expand All @@ -269,11 +268,10 @@ def test_print_alerts_defence(self, mocked_print, gtfs_fixture):
gtfs_fixture.print_alerts()

gtfs_fixture.is_valid()
gtfs_fixture.print_alerts(alert_type="doesnt_exist")
fun_out = mocked_print.mock_calls
assert fun_out == [
call("No alerts of type doesnt_exist were found.")
], f"Expected a print about alert_type but found: {fun_out}"
with pytest.warns(
UserWarning, match="No alerts of type doesnt_exist were found."
):
gtfs_fixture.print_alerts(alert_type="doesnt_exist")

@patch("builtins.print") # testing print statements
def test_print_alerts_single_case(self, mocked_print, gtfs_fixture):
Expand Down Expand Up @@ -302,8 +300,7 @@ def test_print_alerts_multi_case(self, mocked_print, gtfs_fixture):
call("Unrecognized column vehicle_journey_code"),
], f"Expected print statements about GTFS warnings. Found: {fun_out}"

@patch("builtins.print")
def test_viz_stops_defence(self, mocked_print, tmpdir, gtfs_fixture):
def test_viz_stops_defence(self, tmpdir, gtfs_fixture):
"""Check defensive behaviours of viz_stops()."""
tmp = os.path.join(tmpdir, "somefile.html")
with pytest.raises(
Expand All @@ -324,13 +321,15 @@ def test_viz_stops_defence(self, mocked_print, tmpdir, gtfs_fixture):
match="`geom_crs`.*string or integer. Found <class 'float'>",
):
gtfs_fixture.viz_stops(out_pth=tmp, geom_crs=1.1)
# check missing stop_id results in print instead of exception
# check missing stop_id results in an informative error message
gtfs_fixture.feed.stops.drop("stop_id", axis=1, inplace=True)
gtfs_fixture.viz_stops(out_pth=tmp)
fun_out = mocked_print.mock_calls
assert fun_out == [
call("Key Error. Map was not written.")
], f"Expected confirmation that map was not written. Found: {fun_out}"
with pytest.raises(
KeyError,
match="The stops table has no 'stop_code' column. While "
"this is an optional field in a GTFS file, it "
"raises an error through the gtfs-kit package.",
):
gtfs_fixture.viz_stops(out_pth=tmp)

@patch("builtins.print")
def test_viz_stops_point(self, mock_print, tmpdir, gtfs_fixture):
Expand All @@ -350,18 +349,18 @@ def test_viz_stops_point(self, mock_print, tmpdir, gtfs_fixture):
), f"{no_parent_pth} was expected to exist but it was not found."
# check behaviour when not implemented fileext used
tmp1 = os.path.join(tmpdir, "points2.svg")
gtfs_fixture.viz_stops(out_pth=pathlib.Path(tmp1))
with pytest.warns(
UserWarning,
match=".svg format not implemented. Saving as .html",
):
gtfs_fixture.viz_stops(out_pth=pathlib.Path(tmp1))
# need to use regex for the first print statement, as tmpdir will
# change.
start_pat = re.compile(r"Creating parent directory:.*")
out = mock_print.mock_calls[0].__str__()
assert bool(
start_pat.search(out)
), f"Print statement about directory creation expected. Found: {out}"
out_last = mock_print.mock_calls[-1]
assert out_last == call(
".svg format not implemented. Writing to .html"
), f"Expected print statement about .svg. Found: {out_last}"
write_pth = os.path.join(tmpdir, "points2.html")
assert os.path.exists(
write_pth
Expand Down
149 changes: 146 additions & 3 deletions tests/population/test_rasterpop.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,9 +4,6 @@
in Newport. The data vales do not represent any real or meaningfull measurments
, they are purely for use as a unit test and mimic input data used in this repo
.
Note: this suite of tests does not cover plotting methods. TODO add plotting
unit tests. See issue #43.
"""

import os
Expand All @@ -21,6 +18,7 @@
from numpy.dtypes import Float64DType
from pytest_lazyfixture import lazy_fixture
from _pytest.python_api import RaisesContext
from contextlib import nullcontext as does_not_raise

from transport_performance.population.rasterpop import RasterPop
from transport_performance.utils.test_utils import _np_to_rioxarray
Expand Down Expand Up @@ -508,3 +506,148 @@ def test_rasterpop_crs_conversion(
assert np.array_equal(
pop_gdf.within_urban_centre, xarr_1_uc[1]["within_uc"]
)

@pytest.mark.parametrize(
"which, save_folder, save_filename",
[
("folium", "outputs", "folium.html"),
("cartopy", "outputs", "cartopy.png"),
("matplotlib", "outputs", "matplotlib.png"),
],
)
def test_plot_on_pass(
self,
xarr_1_fpath: str,
xarr_1_aoi: tuple,
xarr_1_uc: tuple,
tmp_path: str,
which: str,
save_folder: str,
save_filename: str,
) -> None:
"""Test plotting methods.
Parameters
----------
xarr_1_fpath : str
filepath to dummy raster data
xarr_1_aoi : tuple
aoi polygon for dummy input
xarr_1_uc : tuple
urban centre polugon for dummy input
tmp_path : str
temporary path to save output within
which : str
plotting backend to use
save_folder: str
folder to save output within
save_filename : str
filename to use when saving the file within the temp directory
"""
# create the full output path
output_path = os.path.join(tmp_path, save_folder, save_filename)

# run raster pop and assert that the file is generated
rp = RasterPop(xarr_1_fpath)
rp.get_pop(xarr_1_aoi[0], urban_centre_bounds=xarr_1_uc[0])
rp.plot(which=which, save=output_path)
assert os.path.exists(output_path)

def test_plot_before_get_data(self, xarr_1_fpath: str) -> None:
"""Test case where plot is called before getting data.
Parameters
----------
xarr_1_fpath : str
file path to dummy data.
"""
rp = RasterPop(xarr_1_fpath)
with pytest.raises(
NotImplementedError,
match="Unable to call `plot` without calling `get_pop`.",
):
rp.plot()

def test_plot_unknown_which(
self,
xarr_1_fpath: str,
xarr_1_aoi: tuple,
xarr_1_uc: tuple,
) -> None:
"""Test a plot call with an unknown backend plot type.
Parameters
----------
xarr_1_fpath : str
file path to dummy data.
xarr_1_aoi : tuple
area of interest polygon for dummy data.
xarr_1_uc : tuple
urban centre polygon for dummy data.
"""
rp = RasterPop(xarr_1_fpath)
rp.get_pop(xarr_1_aoi[0], urban_centre_bounds=xarr_1_uc[0])
unknown_plot_backend = "unknown"
with pytest.raises(
ValueError,
match=(
f"Unrecognised value for `which` {unknown_plot_backend}. "
"Must be one of "
),
):
rp.plot(which=unknown_plot_backend)

def test_plot_foliumn_no_uc(
self,
xarr_1_fpath: str,
xarr_1_aoi: tuple,
) -> None:
"""Test folium plotting backend with no urban centre.
Unit test written to capture fix for issue 52.
Parameters
----------
xarr_1_fpath : str
file path to dummy data.
xarr_1_aoi : tuple
area of interest polygon for dummy data.
"""
rp = RasterPop(xarr_1_fpath)
rp.get_pop(xarr_1_aoi[0])
with does_not_raise():
rp.plot(which="folium")

def test_plot_cartopy_unknown_attr_location(
self,
xarr_1_fpath: str,
xarr_1_aoi: tuple,
) -> None:
"""Test cartopy plotting backend with an unknown attribute location.
Parameters
----------
xarr_1_fpath : str
Filepath to dummy data.
xarr_1_aoi : tuple
Area of interest for dummy data.
"""
rp = RasterPop(xarr_1_fpath)
rp.get_pop(xarr_1_aoi[0])
test_attr_location = "unknown"
with pytest.raises(
ValueError,
match=(
f"Unrecognised value for `attr_location` {test_attr_location}."
"Expecting one of "
),
):
rp.plot(
which="cartopy",
attr_location=test_attr_location,
)
23 changes: 23 additions & 0 deletions tests/population/test_vectorpop.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,23 @@
"""Unit tests for transport_performance/population/vectorpop.py.
At this stage, VectorPop has not been implemented, so the extent of this test
suite is to ensure the NotImplementedError is raised.
TODO: add unit test for methods once VectorPop has been implemented.
"""

import pytest

from transport_performance.population.vectorpop import VectorPop


class TestVectorPop:
"""A class to test VectorPop methods."""

def test_vectorpop_not_implemented(self) -> None:
"""Test VectorPop raises a not implemented error."""
with pytest.raises(
NotImplementedError,
match="This class has not yet been implemented",
):
VectorPop()
Loading

0 comments on commit 003a99a

Please sign in to comment.