Coverage for tests/test_analysis.py: 100%
166 statements
« prev ^ index » next coverage.py v7.17.0, created at 2026-10-30 18:01 +0000
1"""Tests of combining, analyzing, and plotting crime data.
3Contents:
4 agencies: a small table of agencies.
5 incidents: a small table of incidents.
6 test_analyses_accept_datasets: checks analyses of frames and datasets.
7 test_clearance_rates: checks clearance rates.
8 test_cli_exports_tables: checks the command line interface.
9 test_combining_keeps_label_and_groups: checks combined datasets.
10 test_count: checks counts.
11 test_count_crimes_and_describers: checks the techniques directly.
12 test_coverage_and_percent_change_of_datasets: checks tables of datasets.
13 test_crime_rates: checks crime rates.
14 test_crime_rates_load_agencies: checks crime rates of loaded agencies.
15 test_crosstab_and_coverage: checks crosstabs and coverage.
16 test_decode_nibrs: checks the decoding technique.
17 test_load_nibrs_in_a_project: checks loading NIBRS data in a project.
18 test_load_nibrs_needs_years: checks that loading needs years.
19 test_merge_keeps_categories: checks merging.
20 test_plots_return_figures: checks the plots.
21 test_rates_and_percent_change: checks rates and percent changes.
22 test_stack_keeps_categories: checks stacking.
23 test_techniques_are_in_the_amos_library: checks the techniques' kinds.
24 test_trend: checks trends.
25 test_utilities: checks the shared tools.
27"""
29from __future__ import annotations
31import pathlib
33import amos
34import pandas as pd
35import pytest
37import crimepy
38from crimepy import analyze, cli, combine, plot, utilities
40""" Fixtures """
43@pytest.fixture
44def incidents() -> pd.DataFrame:
45 """Returns a small table of incidents.
47 Returns:
48 Six incidents of two agencies in two states and years, with dates,
49 hours, categories, clearances, and victims.
51 """
52 return pd.DataFrame({
53 'data_year': [2022, 2022, 2022, 2023, 2023, 2023],
54 'agency_ori': ['A1', 'A1', 'B1', 'A1', 'B1', 'B1'],
55 'state': ['DE', 'DE', 'MD', 'DE', 'MD', 'MD'],
56 'incident_date': pd.to_datetime([
57 '2022-01-03', '2022-01-20', '2022-02-01', '2023-01-05',
58 '2023-03-04', '2023-03-09']),
59 'incident_hour': pd.array([1, 13, None, 23, 0, 12], dtype = 'Int64'),
60 'offense_category': pd.Categorical([
61 'Assault Offenses', 'Robbery', 'Assault Offenses',
62 'Assault Offenses', 'Robbery', 'Assault Offenses']),
63 'cleared': pd.array(
64 [True, False, True, None, False, True], dtype = 'boolean'),
65 'victim_age': pd.array([25, 40, None, 8, 70, 33], dtype = 'Int64'),
66 'victim_sex': ['Male', 'Female', 'Male', 'Female', 'Male', 'Female']})
69@pytest.fixture
70def agencies() -> pd.DataFrame:
71 """Returns a small table of agencies.
73 Returns:
74 The two agencies of `incidents` in each of two years, with their
75 populations and months active.
77 """
78 return pd.DataFrame({
79 'data_year': [2022, 2022, 2023, 2023],
80 'ori': ['A1', 'B1', 'A1', 'B1'],
81 'state': ['DE', 'MD', 'DE', 'MD'],
82 'population': pd.array([1000, 2000, 1000, 2000], dtype = 'Int64'),
83 'months_active': pd.array([12, 6, 12, 12], dtype = 'Int64'),
84 'covered_by_ori': [None, None, None, None]})
87""" Tests """
90def test_count(incidents: pd.DataFrame) -> None:
91 """Checks counts of all rows, groups, and periods, and of datasets.
93 Args:
94 incidents: a small table of incidents.
96 """
97 counted = analyze.count(incidents)
98 assert isinstance(counted, amos.Dataset)
99 assert counted.label == 'count'
100 assert counted.data['count'].item() == 6
101 by_state = analyze.count(incidents, 'state').data
102 counts = dict(zip(by_state['state'], by_state['count'], strict = True))
103 assert counts == {'DE': 3, 'MD': 3}
104 monthly = analyze.count(incidents, freq = 'MS').data
105 assert list(monthly.columns) == ['period', 'count']
106 assert monthly['count'].sum() == 6
107 dataset = amos.Dataset.create(incidents, label = 'cleared')
108 assert len(analyze.count(dataset, 'state').data) == 2
109 assert len(dataset.data) == 6
112def test_trend(incidents: pd.DataFrame) -> None:
113 """Checks counts over time, by group and for the top groups.
115 Args:
116 incidents: a small table of incidents.
118 """
119 table = analyze.trend(incidents, 'offense_category', freq = 'YS')
120 assert list(table.columns) == ['Assault Offenses', 'Robbery']
121 assert table.loc['2023-01-01', 'Assault Offenses'] == 2
122 top = analyze.trend(incidents, 'offense_category', freq = 'YS', top = 1)
123 assert list(top.columns) == ['Assault Offenses']
124 assert analyze.trend(incidents, freq = 'YS')['count'].tolist() == [3, 3]
127def test_rates_and_percent_change(incidents: pd.DataFrame) -> None:
128 """Checks rates per people and percent changes.
130 Args:
131 incidents: a small table of incidents.
133 """
134 counts = analyze.count(incidents, ['data_year', 'state'])
135 people = pd.DataFrame({
136 'data_year': [2022, 2022, 2023, 2023],
137 'state': ['DE', 'MD', 'DE', 'MD'],
138 'population': [1000, 0, 1000, 2000]})
139 table = analyze.rates(counts, people, per = 1000)
140 rates = table.set_index(['data_year', 'state'])['rate']
141 assert rates[(2022, 'DE')] == 2
142 assert pd.isna(rates[(2022, 'MD')])
143 change = analyze.percent_change(pd.Series([10, 15, 12]))
144 assert change.tolist()[1:] == pytest.approx([50.0, -20.0])
147def test_crime_rates(incidents: pd.DataFrame, agencies: pd.DataFrame) -> None:
148 """Checks that crime rates count only agencies that reported all year.
150 Args:
151 incidents: a small table of incidents.
152 agencies: a small table of agencies.
154 """
155 table = analyze.crime_rates(incidents, agencies, per = 1000)
156 rows = table.set_index(['data_year', 'state'])
157 assert (2022, 'MD') not in rows.index
158 assert rows.loc[(2022, 'DE'), 'count'] == 2
159 assert rows.loc[(2023, 'MD'), 'rate'] == 1
162def test_clearance_rates(incidents: pd.DataFrame) -> None:
163 """Checks the share of incidents cleared, overall and by state.
165 Args:
166 incidents: a small table of incidents.
168 """
169 overall = analyze.clearance_rates(incidents)
170 assert overall['clearance_rate'].item() == pytest.approx(0.5)
171 by_state = analyze.clearance_rates(incidents, 'state').set_index('state')
172 assert by_state.loc['MD', 'cleared'] == 2
175def test_crosstab_and_coverage(
176 incidents: pd.DataFrame,
177 agencies: pd.DataFrame) -> None:
178 """Checks crosstabs and the coverage of agencies.
180 Args:
181 incidents: a small table of incidents.
182 agencies: a small table of agencies.
184 """
185 table = analyze.crosstab(incidents, 'state', 'victim_sex')
186 assert table.loc['DE', 'Female'] == 2
187 covered = analyze.coverage(agencies).set_index(['data_year', 'state'])
188 assert covered.loc[(2022, 'MD'), 'reporting'] == 0
189 assert covered.loc[(2023, 'MD'), 'share_reporting'] == 1
192@pytest.mark.parametrize(
193 ('describe', 'kwargs', 'name'),
194 [
195 (analyze.trend, {'by': 'state', 'freq': 'YS'}, 'crime_trend'),
196 (analyze.clearance_rates, {'by': 'state'}, 'clearance_rates'),
197 (
198 analyze.crosstab, {'rows': 'state', 'columns': 'victim_sex'},
199 'crosstab')])
200def test_analyses_accept_datasets(
201 incidents: pd.DataFrame,
202 describe: object,
203 kwargs: dict[str, object],
204 name: str) -> None:
205 """Checks that analyses of a dataset match those of its data frame.
207 Args:
208 incidents: a small table of incidents.
209 describe: the analysis.
210 kwargs: arguments for the analysis.
211 name: the name of the technique, under which the dataset keeps the
212 table.
214 """
215 from_frame = describe(incidents, **kwargs) # type: ignore[operator]
216 dataset = amos.Dataset.create(incidents)
217 from_dataset = describe(dataset, **kwargs) # type: ignore[operator]
218 pd.testing.assert_frame_equal(from_frame, from_dataset)
219 assert dataset.tables[name] is from_dataset
222def test_coverage_and_percent_change_of_datasets(
223 incidents: pd.DataFrame,
224 agencies: pd.DataFrame) -> None:
225 """Checks that datasets keep the tables of coverage and percent change.
227 Args:
228 incidents: a small table of incidents.
229 agencies: a small table of agencies.
231 """
232 dataset = amos.Dataset.create(agencies)
233 analyze.coverage(dataset)
234 assert 'coverage' in dataset.tables
235 events = amos.Dataset.create(incidents)
236 change = crimepy.techniques.PercentChange().apply(
237 events, freq = 'YS').tables['percent_change']
238 assert change['count'].tolist()[1:] == [0.0]
239 yearly = analyze.trend(incidents, freq = 'YS')
240 assert analyze.percent_change(yearly)['count'].tolist()[1:] == [0.0]
243def test_crime_rates_load_agencies(data_folder: pathlib.Path) -> None:
244 """Checks that crime rates load the agencies of the incidents' years.
246 Args:
247 data_folder: a data folder with the master files of 2004 and 2025.
249 """
250 incidents = crimepy.nibrs.tables.build(
251 'incidents', 2004, folder = data_folder)
252 table = analyze.crime_rates(incidents, folder = data_folder)
253 assert table.set_index('state').loc['AZ', 'count'] == 2
254 assert 'crime_rates' in incidents.tables
257def test_stack_keeps_categories() -> None:
258 """Checks that stacking keeps categories and checks its labels."""
259 first = pd.DataFrame({'x': pd.Categorical(['a', 'b']), 'y': [1, 2]})
260 second = pd.DataFrame({'x': pd.Categorical(['c']), 'z': [3]})
261 stacked = combine.stack(
262 [first, second], source = 'year', labels = [2022, 2023]).data
263 assert isinstance(stacked['x'].dtype, pd.CategoricalDtype)
264 assert list(stacked['x']) == ['a', 'b', 'c']
265 assert list(stacked['year']) == [2022, 2022, 2023]
266 assert stacked['z'].isna().sum() == 2
267 with pytest.raises(ValueError, match = 'no tables'):
268 combine.stack([])
269 with pytest.raises(ValueError, match = 'label'):
270 combine.stack([first], labels = [1, 2])
273def test_merge_keeps_categories() -> None:
274 """Checks that merging keeps the categories of its keys."""
275 left = pd.DataFrame({'k': pd.Categorical(['a', 'b']), 'x': [1, 2]})
276 right = pd.DataFrame({'k': pd.Categorical(['b', 'c']), 'x': [3, 4]})
277 merged = combine.merge(left, right, 'k').data
278 assert isinstance(merged['k'].dtype, pd.CategoricalDtype)
279 assert merged['x_right'].tolist()[1] == 3
282def test_combining_keeps_label_and_groups() -> None:
283 """Checks that combined datasets keep the first one's label and groups."""
284 first = amos.Dataset.create(
285 pd.DataFrame({'k': ['a', 'b'], 'y': [1.5, 2.5], 'g': [1, 2]}),
286 label = 'y', groups = 'g')
287 second = pd.DataFrame({'k': ['c'], 'y': [3.5], 'g': [3]})
288 stacked = combine.stack([second, first])
289 assert (stacked.label, stacked.groups) == ('y', ['g'])
290 assert len(stacked.data) == 3
291 merged = combine.merge(
292 first, pd.DataFrame({'k': ['a'], 'z': [9]}), 'k')
293 assert (merged.label, merged.groups) == ('y', ['g'])
294 assert combine.stack([second]).label is None
297@pytest.mark.parametrize(
298 ('draw', 'kwargs'),
299 [
300 (plot.trend, {'by': 'offense_category', 'freq': 'MS'}),
301 (plot.top_values, {'column': 'offense_category', 'share': True}),
302 (plot.time_heatmap, {}),
303 (plot.age_pyramid, {'person': 'victim'})])
304def test_plots_return_figures(
305 incidents: pd.DataFrame,
306 draw: object,
307 kwargs: dict[str, object]) -> None:
308 """Checks that each plot returns a figure and is kept by a dataset.
310 Args:
311 incidents: a small table of incidents.
312 draw: the function that draws the plot.
313 kwargs: arguments for the plot.
315 """
316 figure = draw( # type: ignore[operator]
317 incidents, title = 'A figure', **kwargs)
318 assert figure.axes
319 assert figure.get_suptitle() == 'A figure'
320 dataset = amos.Dataset.create(incidents)
321 drawn = draw(dataset, **kwargs) # type: ignore[operator]
322 assert list(dataset.figures.values()) == [drawn]
325def test_techniques_are_in_the_amos_library() -> None:
326 """Checks that each technique is in the amos library as its kind."""
327 kinds = {
328 'load_nibrs': 'loader',
329 'decode_nibrs': 'munger',
330 'count_crimes': 'shaper',
331 'crime_trend': 'describer',
332 'population_rates': 'describer',
333 'crime_rates': 'describer',
334 'clearance_rates': 'describer',
335 'percent_change': 'describer',
336 'crosstab': 'describer',
337 'coverage': 'describer',
338 'trend_plot': 'plot',
339 'top_values': 'plot',
340 'time_heatmap': 'plot',
341 'age_pyramid': 'plot'}
342 for name, kind in kinds.items():
343 assert amos.library.classify(name) == kind
346def test_count_crimes_and_describers(incidents: pd.DataFrame) -> None:
347 """Checks the counting and describing techniques applied directly.
349 Args:
350 incidents: a small table of incidents.
352 """
353 counted = crimepy.techniques.CountCrimes().apply(
354 incidents, by = 'state, offense_category', label = 'count')
355 assert counted.label == 'count'
356 assert counted.data['count'].sum() == 6
357 trend = crimepy.techniques.CrimeTrend().apply(
358 incidents, by = 'state', freq = 'YS', top = '1')
359 assert trend.tables['crime_trend'].shape == (2, 1)
360 rates = crimepy.techniques.ClearanceRates().apply(incidents, by = 'state')
361 assert len(rates.tables['clearance_rates']) == 2
364def test_load_nibrs_in_a_project(data_folder: pathlib.Path) -> None:
365 """Checks that an amos project loads and counts NIBRS data.
367 Args:
368 data_folder: a data folder with the master files of 2004 and 2025.
370 """
371 settings = {
372 'general': {'seed': 43, 'label': 'count'},
373 'crime_project': {'crime_workers': ['wrangler']},
374 'wrangler': {'techniques': ['load_nibrs', 'count_crimes']},
375 'load_nibrs_parameters': {
376 'table': 'incidents', 'years': '2004', 'states': 'AZ, AL',
377 'folder': str(data_folder)},
378 'count_crimes_parameters': {'by': 'state, cleared'}}
379 project = amos.Project.create(settings, id = 'nibrs')
380 assert project.result.data['count'].sum() == 2
381 assert project.result.label == 'count'
384def test_load_nibrs_needs_years() -> None:
385 """Checks that loading NIBRS data without years fails."""
386 with pytest.raises(ValueError, match = 'years'):
387 crimepy.nibrs.techniques.LoadNibrs().apply(pd.DataFrame({'x': [1]}))
390def test_decode_nibrs() -> None:
391 """Checks that the decoding technique replaces codes with labels."""
392 data = pd.DataFrame({'location': pd.Categorical(['20']), 'x': [1]})
393 decoded = crimepy.nibrs.techniques.DecodeNibrs().apply(data)
394 assert decoded.data['location'].tolist() == ['Residence/Home']
397def test_utilities() -> None:
398 """Checks the conversions of years, names, and booleans."""
399 assert utilities.to_years('2019-2021, 2023') == [2019, 2020, 2021, 2023]
400 assert utilities.to_years([2022, range(2019, 2021), '2018']) == [
401 2018, 2019, 2020, 2022]
402 assert utilities.to_years(2020.0) == [2020]
403 with pytest.raises(ValueError, match = 'year'):
404 utilities.to_years('twenty')
405 assert utilities.to_names('DE, MD') == ['DE', 'MD']
406 assert utilities.to_names(None) == []
407 assert utilities.to_bool('yes') and not utilities.to_bool('false')
408 with pytest.raises(ValueError, match = 'true'):
409 utilities.to_bool('maybe')
412def test_cli_exports_tables(
413 data_folder: pathlib.Path,
414 tmp_path: pathlib.Path,
415 capsys: pytest.CaptureFixture[str]) -> None:
416 """Checks that the command line interface exports and converts tables.
418 Args:
419 data_folder: a data folder with the master files of 2004 and 2025.
420 tmp_path: the test's temporary folder.
421 capsys: captures what the commands print.
423 """
424 output = tmp_path / 'out' / 'incidents.csv'
425 status = cli.main([
426 'nibrs', 'export', 'incidents', '2004', '--folder', str(data_folder),
427 '--output', str(output)])
428 assert status == 0
429 assert len(pd.read_csv(output)) == 2
430 output = tmp_path / 'agencies.parquet'
431 cli.main([
432 'nibrs', 'export', 'agencies', '2004', '--folder', str(data_folder),
433 '--states', 'AZ', '--codes', '--output', str(output)])
434 assert pd.read_parquet(output)['ori'].tolist() == ['AZ0111300']
435 assert cli.main([
436 'nibrs', 'export', 'agencies', '2004', '--output', 'x.txt']) == 2
437 status = cli.main([
438 'nibrs', 'convert', '2025', '--folder', str(data_folder),
439 '--tables', 'offenders'])
440 assert status == 0
441 assert '2025: 1 tables' in capsys.readouterr().out
442 assert cli.main([]) == 1