641 lines
24 KiB
Python
641 lines
24 KiB
Python
from __future__ import annotations
|
|
|
|
import io
|
|
import json
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from contextlib import redirect_stdout
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
import numpy as np
|
|
|
|
from evaluate_floor_state_changes import (
|
|
NUM_CLASSES,
|
|
NUM_HORIZONS,
|
|
collect_floor_event_candidates,
|
|
evaluate_event_horizon,
|
|
evaluate_horizon,
|
|
load_arrays,
|
|
main,
|
|
parse_runs,
|
|
print_event_timeline_examples,
|
|
validate_recalls,
|
|
)
|
|
|
|
|
|
def activity(wearer: int, others: int) -> list[int]:
|
|
return [int(not wearer and not others), wearer, others]
|
|
|
|
|
|
class FloorStateMetricTests(unittest.TestCase):
|
|
def test_post_boundary_candidate_uses_first_post_transition_row(self) -> None:
|
|
target = np.asarray(
|
|
[
|
|
activity(0, 1),
|
|
activity(0, 0), # observed other offset
|
|
activity(1, 0), # wearer response
|
|
activity(1, 0),
|
|
],
|
|
dtype=np.float32,
|
|
)
|
|
session_ids = np.zeros(len(target), dtype=np.int32)
|
|
pre, _ = collect_floor_event_candidates(
|
|
target,
|
|
session_ids,
|
|
1,
|
|
event_type="response",
|
|
decision_timing="pre-boundary",
|
|
)
|
|
post, diagnostics = collect_floor_event_candidates(
|
|
target,
|
|
session_ids,
|
|
1,
|
|
event_type="response",
|
|
decision_timing="post-boundary",
|
|
)
|
|
self.assertEqual([(item.source_row, item.label) for item in pre], [(0, True)])
|
|
self.assertEqual([(item.source_row, item.label) for item in post], [(1, True)])
|
|
self.assertEqual(diagnostics["num_simultaneous_boundary_outcomes_excluded"], 0)
|
|
|
|
def test_post_boundary_excludes_outcome_that_already_happened_at_anchor(self) -> None:
|
|
target = np.asarray(
|
|
[
|
|
activity(0, 1),
|
|
activity(1, 0), # simultaneous other offset and wearer onset
|
|
activity(1, 0),
|
|
],
|
|
dtype=np.float32,
|
|
)
|
|
session_ids = np.zeros(len(target), dtype=np.int32)
|
|
pre, _ = collect_floor_event_candidates(
|
|
target,
|
|
session_ids,
|
|
0,
|
|
event_type="response",
|
|
decision_timing="pre-boundary",
|
|
)
|
|
post, diagnostics = collect_floor_event_candidates(
|
|
target,
|
|
session_ids,
|
|
0,
|
|
event_type="response",
|
|
decision_timing="post-boundary",
|
|
)
|
|
self.assertEqual(len(pre), 1)
|
|
self.assertTrue(pre[0].label)
|
|
self.assertEqual(post, [])
|
|
self.assertEqual(diagnostics["num_source_boundaries"], 1)
|
|
self.assertEqual(diagnostics["num_simultaneous_boundary_outcomes_excluded"], 1)
|
|
|
|
def test_post_boundary_reports_competitor_with_no_observable_bin(self) -> None:
|
|
target = np.asarray(
|
|
[
|
|
activity(0, 1),
|
|
activity(0, 0),
|
|
activity(0, 1), # immediate restart after reactive source
|
|
activity(0, 1),
|
|
],
|
|
dtype=np.float32,
|
|
)
|
|
candidates, diagnostics = collect_floor_event_candidates(
|
|
target,
|
|
np.zeros(len(target), dtype=np.int32),
|
|
1,
|
|
event_type="response",
|
|
decision_timing="post-boundary",
|
|
)
|
|
self.assertEqual(candidates, [])
|
|
self.assertEqual(diagnostics["num_zero_window_competing_risk"], 1)
|
|
|
|
def test_response_window_stops_before_next_other_onset(self) -> None:
|
|
target = np.asarray(
|
|
[
|
|
activity(0, 1), # session 0: other offset anchor
|
|
activity(0, 0),
|
|
activity(0, 1), # competing other onset; window ends before here
|
|
activity(1, 1), # must not count as response to row-1 offset
|
|
activity(0, 1), # session 1: other offset anchor
|
|
activity(0, 0),
|
|
activity(1, 0), # valid response before any competing onset
|
|
activity(1, 1), # later competitor; response has already won
|
|
],
|
|
dtype=np.float32,
|
|
)
|
|
prediction = np.full(
|
|
(len(target), NUM_CLASSES, NUM_HORIZONS), 0.5, dtype=np.float32
|
|
)
|
|
prediction[1, 1, 0] = 0.1
|
|
prediction[2, 1, 1] = 0.2
|
|
prediction[3, 1, 2] = 0.99 # excluded by competing-risk censoring
|
|
prediction[5, 1, 0] = 0.2
|
|
prediction[6, 1, 1] = 0.9
|
|
prediction[7, 1, 2] = 0.9
|
|
valid = np.ones((len(target), NUM_HORIZONS), dtype=bool)
|
|
session_ids = np.asarray([0, 0, 0, 0, 1, 1, 1, 1], dtype=np.int32)
|
|
|
|
result = evaluate_event_horizon(
|
|
target,
|
|
prediction,
|
|
valid,
|
|
session_ids,
|
|
2,
|
|
[0.8],
|
|
midturn_pause_max_steps=10,
|
|
)
|
|
response = result["other_offset_to_wearer_response_event"]
|
|
self.assertEqual(response["num_positive"], 1)
|
|
self.assertEqual(response["num_negative"], 1)
|
|
self.assertEqual(response["num_competing_risk_censored"], 1)
|
|
self.assertEqual(response["num_short_resume_hard_negative"], 1)
|
|
self.assertAlmostEqual(response["average_precision"], 1.0)
|
|
self.assertAlmostEqual(response["min_effective_window_seconds"], 0.2)
|
|
self.assertAlmostEqual(response["max_effective_window_seconds"], 0.4)
|
|
|
|
def test_yield_window_stops_before_next_other_offset(self) -> None:
|
|
target = np.asarray(
|
|
[
|
|
activity(1, 0), # session 0: other onset anchor
|
|
activity(1, 1),
|
|
activity(1, 0), # competing other offset; window ends before here
|
|
activity(0, 0), # must not count as yield to row-1 onset
|
|
activity(1, 0), # session 1: other onset anchor
|
|
activity(1, 1),
|
|
activity(0, 1), # valid yield while other remains active
|
|
activity(0, 0), # later competitor; yield has already won
|
|
],
|
|
dtype=np.float32,
|
|
)
|
|
prediction = np.full(
|
|
(len(target), NUM_CLASSES, NUM_HORIZONS), 0.5, dtype=np.float32
|
|
)
|
|
prediction[1, 1, 0] = 0.9
|
|
prediction[2, 1, 1] = 0.2
|
|
prediction[3, 1, 2] = 0.01 # excluded by competing-risk censoring
|
|
prediction[5, 1, 0] = 0.8
|
|
prediction[6, 1, 1] = 0.1
|
|
prediction[7, 1, 2] = 0.1
|
|
valid = np.ones((len(target), NUM_HORIZONS), dtype=bool)
|
|
session_ids = np.asarray([0, 0, 0, 0, 1, 1, 1, 1], dtype=np.int32)
|
|
|
|
result = evaluate_event_horizon(
|
|
target,
|
|
prediction,
|
|
valid,
|
|
session_ids,
|
|
2,
|
|
[0.8],
|
|
midturn_pause_max_steps=10,
|
|
)
|
|
yield_metric = result["other_onset_to_wearer_yield_event"]
|
|
self.assertEqual(yield_metric["num_positive"], 1)
|
|
self.assertEqual(yield_metric["num_negative"], 1)
|
|
self.assertEqual(yield_metric["num_competing_risk_censored"], 1)
|
|
self.assertAlmostEqual(yield_metric["average_precision"], 1.0)
|
|
point = yield_metric["operating_points"][0]
|
|
self.assertEqual(point["negative_false_positive_count"], 0)
|
|
|
|
def test_observed_response_before_recording_end_is_not_right_censored(self) -> None:
|
|
target = np.asarray(
|
|
[
|
|
activity(0, 1),
|
|
activity(0, 0),
|
|
activity(1, 0), # observed response, but nominal 0.6-s end is absent
|
|
],
|
|
dtype=np.float32,
|
|
)
|
|
prediction = np.full(
|
|
(len(target), NUM_CLASSES, NUM_HORIZONS), 0.5, dtype=np.float32
|
|
)
|
|
valid = np.ones((len(target), NUM_HORIZONS), dtype=bool)
|
|
session_ids = np.zeros(len(target), dtype=np.int32)
|
|
result = evaluate_event_horizon(
|
|
target,
|
|
prediction,
|
|
valid,
|
|
session_ids,
|
|
2,
|
|
[0.8],
|
|
midturn_pause_max_steps=10,
|
|
)
|
|
response = result["other_offset_to_wearer_response_event"]
|
|
self.assertEqual(response["num_positive"], 1)
|
|
self.assertEqual(response["num_right_censored"], 0)
|
|
|
|
def test_unresolved_event_at_recording_end_is_right_censored(self) -> None:
|
|
target = np.asarray(
|
|
[
|
|
activity(0, 1),
|
|
activity(0, 0),
|
|
activity(0, 0),
|
|
],
|
|
dtype=np.float32,
|
|
)
|
|
prediction = np.full(
|
|
(len(target), NUM_CLASSES, NUM_HORIZONS), 0.5, dtype=np.float32
|
|
)
|
|
valid = np.ones((len(target), NUM_HORIZONS), dtype=bool)
|
|
session_ids = np.zeros(len(target), dtype=np.int32)
|
|
result = evaluate_event_horizon(
|
|
target,
|
|
prediction,
|
|
valid,
|
|
session_ids,
|
|
2,
|
|
[0.8],
|
|
midturn_pause_max_steps=10,
|
|
)
|
|
response = result["other_offset_to_wearer_response_event"]
|
|
self.assertEqual(response["num_evaluated"], 0)
|
|
self.assertEqual(response["num_right_censored"], 1)
|
|
|
|
def test_simultaneous_wearer_entry_at_competing_onset_is_excluded(self) -> None:
|
|
target = np.asarray(
|
|
[
|
|
activity(0, 1),
|
|
activity(0, 0),
|
|
activity(1, 1), # wearer and others restart simultaneously
|
|
activity(1, 1),
|
|
],
|
|
dtype=np.float32,
|
|
)
|
|
prediction = np.full(
|
|
(len(target), NUM_CLASSES, NUM_HORIZONS), 0.5, dtype=np.float32
|
|
)
|
|
valid = np.ones((len(target), NUM_HORIZONS), dtype=bool)
|
|
session_ids = np.zeros(len(target), dtype=np.int32)
|
|
result = evaluate_event_horizon(
|
|
target,
|
|
prediction,
|
|
valid,
|
|
session_ids,
|
|
2,
|
|
[0.8],
|
|
midturn_pause_max_steps=10,
|
|
)
|
|
response = result["other_offset_to_wearer_response_event"]
|
|
self.assertEqual(response["num_positive"], 0)
|
|
self.assertEqual(response["num_negative"], 1)
|
|
self.assertEqual(response["num_competing_risk_censored"], 1)
|
|
self.assertEqual(response["num_short_resume_hard_negative"], 0)
|
|
|
|
def test_timeline_marks_competing_boundary_as_excluded(self) -> None:
|
|
target = np.asarray(
|
|
[
|
|
activity(0, 1),
|
|
activity(0, 0),
|
|
activity(0, 1),
|
|
activity(1, 1),
|
|
],
|
|
dtype=np.float32,
|
|
)
|
|
prediction = np.full(
|
|
(len(target), NUM_CLASSES, NUM_HORIZONS), 0.5, dtype=np.float32
|
|
)
|
|
valid = np.ones((len(target), NUM_HORIZONS), dtype=bool)
|
|
session_ids = np.zeros(len(target), dtype=np.int32)
|
|
stream = io.StringIO()
|
|
with redirect_stdout(stream):
|
|
print_event_timeline_examples(
|
|
target,
|
|
prediction,
|
|
valid,
|
|
session_ids,
|
|
["Session_1"],
|
|
examples_per_type=1,
|
|
horizon_steps=3,
|
|
)
|
|
output = stream.getvalue()
|
|
self.assertIn("status=competing-censored negative", output)
|
|
self.assertIn("C/excluded", output)
|
|
|
|
def test_fixed_calibration_threshold_is_not_reselected(self) -> None:
|
|
target = np.asarray(
|
|
[
|
|
activity(0, 1),
|
|
activity(0, 0),
|
|
activity(1, 0),
|
|
activity(1, 0),
|
|
],
|
|
dtype=np.float32,
|
|
)
|
|
prediction = np.full(
|
|
(len(target), NUM_CLASSES, NUM_HORIZONS), 0.5, dtype=np.float32
|
|
)
|
|
valid = np.ones((len(target), NUM_HORIZONS), dtype=bool)
|
|
session_ids = np.zeros(len(target), dtype=np.int32)
|
|
result = evaluate_event_horizon(
|
|
target,
|
|
prediction,
|
|
valid,
|
|
session_ids,
|
|
1,
|
|
[0.8],
|
|
midturn_pause_max_steps=10,
|
|
operating_thresholds={
|
|
"response": {0.8: 0.75},
|
|
"yield": {0.8: 0.25},
|
|
},
|
|
threshold_source="calibration_sessions:Session_12",
|
|
)
|
|
point = result["other_offset_to_wearer_response_event"][
|
|
"operating_points"
|
|
][0]
|
|
self.assertEqual(point["threshold"], 0.75)
|
|
self.assertEqual(
|
|
point["threshold_source"], "calibration_sessions:Session_12"
|
|
)
|
|
self.assertEqual(point["achieved_positive_recall"], 0.0)
|
|
|
|
def test_event_score_uses_full_observable_window(self) -> None:
|
|
target = np.asarray(
|
|
[
|
|
activity(0, 1),
|
|
activity(0, 0),
|
|
activity(1, 0), # response occurs here
|
|
activity(1, 0),
|
|
],
|
|
dtype=np.float32,
|
|
)
|
|
prediction = np.full(
|
|
(len(target), NUM_CLASSES, NUM_HORIZONS), 0.5, dtype=np.float32
|
|
)
|
|
prediction[1, 1, 0] = 0.1
|
|
prediction[2, 1, 1] = 0.2
|
|
prediction[3, 1, 2] = 0.9 # still part of the by-horizon event score
|
|
valid = np.ones((len(target), NUM_HORIZONS), dtype=bool)
|
|
session_ids = np.zeros(len(target), dtype=np.int32)
|
|
result = evaluate_event_horizon(
|
|
target,
|
|
prediction,
|
|
valid,
|
|
session_ids,
|
|
2,
|
|
[0.8],
|
|
midturn_pause_max_steps=10,
|
|
)
|
|
response = result["other_offset_to_wearer_response_event"]
|
|
self.assertAlmostEqual(response["max_effective_window_seconds"], 0.6)
|
|
self.assertAlmostEqual(response["operating_points"][0]["threshold"], 0.9)
|
|
|
|
def test_main_calibrates_on_disjoint_training_session(self) -> None:
|
|
target = np.asarray(
|
|
[
|
|
activity(0, 1),
|
|
activity(0, 0), # response anchor
|
|
activity(1, 0), # response
|
|
activity(1, 0),
|
|
activity(1, 1), # yield anchor
|
|
activity(0, 1), # yield
|
|
activity(0, 1),
|
|
activity(0, 0), # response anchor
|
|
activity(0, 0),
|
|
activity(0, 1), # competing onset, no response
|
|
activity(1, 0), # response anchor and response
|
|
activity(1, 0),
|
|
activity(1, 1), # yield anchor
|
|
activity(1, 1),
|
|
activity(1, 0), # competing offset, no yield
|
|
]
|
|
+ [activity(0, 1)] * 15,
|
|
dtype=np.float32,
|
|
)
|
|
prediction = np.full(
|
|
(len(target), NUM_CLASSES, NUM_HORIZONS), 0.5, dtype=np.float32
|
|
)
|
|
valid = np.ones((len(target), NUM_HORIZONS), dtype=bool)
|
|
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
root = Path(temporary)
|
|
target_root = root / "target"
|
|
run_root = root / "run"
|
|
(run_root / "pred").mkdir(parents=True)
|
|
(run_root / "valid").mkdir(parents=True)
|
|
target_root.mkdir()
|
|
for session in ("Session_1", "Session_12"):
|
|
np.save(target_root / f"{session}.npy", target)
|
|
np.save(run_root / "pred" / f"{session}.npy", prediction)
|
|
np.save(run_root / "valid" / f"{session}.npy", valid)
|
|
|
|
output = root / "report.json"
|
|
argv = [
|
|
"evaluate_floor_state_changes.py",
|
|
"--target-root",
|
|
str(target_root),
|
|
"--run",
|
|
f"model={run_root}",
|
|
"--sessions",
|
|
"Session_1",
|
|
"--calibration-sessions",
|
|
"Session_12",
|
|
"--matched-recalls",
|
|
"0.8",
|
|
"--output",
|
|
str(output),
|
|
]
|
|
with mock.patch.object(sys, "argv", argv), redirect_stdout(io.StringIO()):
|
|
main()
|
|
|
|
payload = json.loads(output.read_text())
|
|
self.assertEqual(
|
|
payload["protocol"], "wearer_floor_event_competing_risk_v2"
|
|
)
|
|
self.assertEqual(payload["calibration_sessions"], ["Session_12"])
|
|
point = payload["runs"]["model"]["event_horizons"][1][
|
|
"other_offset_to_wearer_response_event"
|
|
]["operating_points"][0]
|
|
self.assertEqual(
|
|
point["threshold_source"],
|
|
"calibration_sessions:Session_12",
|
|
)
|
|
framewise_point = payload["runs"]["model"]["horizons"][1][
|
|
"eot_response_vs_other_hold_proxy"
|
|
]["operating_points"][0]
|
|
self.assertEqual(
|
|
framewise_point["threshold_source"],
|
|
"calibration_sessions:Session_12",
|
|
)
|
|
self.assertTrue(output.with_name("report_events.csv").is_file())
|
|
|
|
def test_main_refuses_official_test_without_calibration(self) -> None:
|
|
argv = [
|
|
"evaluate_floor_state_changes.py",
|
|
"--target-root",
|
|
"unused",
|
|
"--run",
|
|
"model=unused",
|
|
"--sessions",
|
|
"Session_1",
|
|
]
|
|
with mock.patch.object(sys, "argv", argv):
|
|
with self.assertRaisesRegex(ValueError, "requires --calibration-sessions"):
|
|
main()
|
|
|
|
def test_main_refuses_test_session_as_official_calibration(self) -> None:
|
|
argv = [
|
|
"evaluate_floor_state_changes.py",
|
|
"--target-root",
|
|
"unused",
|
|
"--run",
|
|
"model=unused",
|
|
"--sessions",
|
|
"Session_1",
|
|
"--calibration-sessions",
|
|
"Session_2",
|
|
]
|
|
with mock.patch.object(sys, "argv", argv):
|
|
with self.assertRaisesRegex(ValueError, "Sessions 4--12"):
|
|
main()
|
|
|
|
def test_event_level_counts_each_boundary_once(self) -> None:
|
|
target = np.asarray(
|
|
[
|
|
activity(0, 1), # other offset follows
|
|
activity(0, 0),
|
|
activity(1, 0), # wearer responds within 0.4s
|
|
activity(1, 0), # other entry follows
|
|
activity(1, 1),
|
|
activity(0, 1), # wearer yields within 0.4s
|
|
activity(0, 1),
|
|
activity(0, 1),
|
|
activity(0, 0), # short pause, no wearer response
|
|
activity(0, 1),
|
|
activity(1, 0), # another other entry follows
|
|
activity(1, 1),
|
|
activity(1, 1), # wearer continues
|
|
],
|
|
dtype=np.float32,
|
|
)
|
|
prediction = np.full(
|
|
(len(target), NUM_CLASSES, NUM_HORIZONS), 0.5, dtype=np.float32
|
|
)
|
|
# Fixed source row 0: response score max(0.2, 0.9).
|
|
prediction[1, 1, 0] = 0.2
|
|
prediction[2, 1, 1] = 0.9
|
|
# Fixed source row 7: no-response score max(0.1, 0.2).
|
|
prediction[8, 1, 0] = 0.1
|
|
prediction[9, 1, 1] = 0.2
|
|
# Fixed source row 3: yield score max(0.2, 0.9).
|
|
prediction[4, 1, 0] = 0.8
|
|
prediction[5, 1, 1] = 0.1
|
|
# Fixed source row 10: continue score max(0.1, 0.2).
|
|
prediction[11, 1, 0] = 0.9
|
|
prediction[12, 1, 1] = 0.8
|
|
valid = np.ones((len(target), NUM_HORIZONS), dtype=bool)
|
|
session_ids = np.zeros(len(target), dtype=np.int32)
|
|
session_ids[10:] = 1
|
|
|
|
result = evaluate_event_horizon(
|
|
target,
|
|
prediction,
|
|
valid,
|
|
session_ids,
|
|
1,
|
|
[0.8],
|
|
midturn_pause_max_steps=10,
|
|
)
|
|
response = result["other_offset_to_wearer_response_event"]
|
|
self.assertEqual(response["num_positive"], 1)
|
|
self.assertEqual(response["num_negative"], 1)
|
|
self.assertEqual(response["num_short_resume_hard_negative"], 1)
|
|
self.assertAlmostEqual(response["average_precision"], 1.0)
|
|
self.assertAlmostEqual(
|
|
response["operating_points"][0]["negative_false_positive_rate"],
|
|
0.0,
|
|
)
|
|
yield_metric = result["other_onset_to_wearer_yield_event"]
|
|
self.assertEqual(yield_metric["num_positive"], 1)
|
|
self.assertEqual(yield_metric["num_negative"], 1)
|
|
self.assertAlmostEqual(yield_metric["average_precision"], 1.0)
|
|
|
|
def test_horizon_one_proxies_and_session_boundary(self) -> None:
|
|
# t0->t1: EOT-response; t3->t4: HOLD; t4->t5: early entry;
|
|
# t6->t7: other-entry yield; t8->t9: other-entry continue.
|
|
target = np.asarray(
|
|
[
|
|
activity(0, 1),
|
|
activity(1, 0),
|
|
activity(0, 0),
|
|
activity(0, 1),
|
|
activity(0, 1),
|
|
activity(1, 1),
|
|
activity(1, 0),
|
|
activity(0, 1),
|
|
activity(1, 0),
|
|
activity(1, 1),
|
|
],
|
|
dtype=np.float32,
|
|
)
|
|
prediction = np.zeros((len(target), NUM_CLASSES, NUM_HORIZONS), dtype=np.float32)
|
|
prediction[:, 1, :] = 0.5
|
|
prediction[1, 1, 0] = 0.9
|
|
prediction[4, 1, 0] = 0.1
|
|
prediction[5, 1, 0] = 0.8
|
|
prediction[7, 1, 0] = 0.1
|
|
prediction[9, 1, 0] = 0.9
|
|
valid = np.ones((len(target), NUM_HORIZONS), dtype=bool)
|
|
session_ids = np.zeros(len(target), dtype=np.int32)
|
|
|
|
result = evaluate_horizon(
|
|
target, prediction, valid, session_ids, 0, [0.8]
|
|
)
|
|
floor = result["eot_response_vs_other_hold_proxy"]
|
|
self.assertEqual(floor["num_positive"], 2)
|
|
self.assertEqual(floor["num_negative"], 1)
|
|
self.assertEqual(floor["num_early_entry_proxy"], 1)
|
|
self.assertAlmostEqual(floor["average_precision"], 1.0)
|
|
self.assertAlmostEqual(
|
|
floor["operating_points"][0]["negative_false_positive_rate"], 0.0
|
|
)
|
|
yield_metric = result["other_entry_yield_vs_continue_proxy"]
|
|
self.assertEqual(yield_metric["num_positive"], 1)
|
|
self.assertEqual(yield_metric["num_negative"], 1)
|
|
self.assertAlmostEqual(yield_metric["average_precision"], 1.0)
|
|
|
|
# The apparent t4->t5 early entry is removed if t5 begins a new session.
|
|
session_ids[5:] = 1
|
|
bounded = evaluate_horizon(
|
|
target, prediction, valid, session_ids, 0, [0.8]
|
|
)
|
|
self.assertEqual(
|
|
bounded["eot_response_vs_other_hold_proxy"]["num_early_entry_proxy"],
|
|
0,
|
|
)
|
|
|
|
def test_load_arrays_validates_layout(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
root = Path(temporary)
|
|
target_root = root / "target"
|
|
run_root = root / "run"
|
|
(run_root / "pred").mkdir(parents=True)
|
|
(run_root / "valid").mkdir(parents=True)
|
|
target_root.mkdir()
|
|
target = np.asarray([activity(0, 0), activity(1, 0)], dtype=np.float32)
|
|
np.save(target_root / "Session_1.npy", target)
|
|
np.save(
|
|
run_root / "pred/Session_1.npy",
|
|
np.zeros((2, NUM_CLASSES, NUM_HORIZONS), dtype=np.float32),
|
|
)
|
|
np.save(
|
|
run_root / "valid/Session_1.npy",
|
|
np.ones((2, NUM_HORIZONS), dtype=bool),
|
|
)
|
|
loaded = load_arrays(target_root, run_root, ["Session_1"])
|
|
self.assertEqual(loaded[0].shape, (2, NUM_CLASSES))
|
|
self.assertEqual(loaded[1].shape, (2, NUM_CLASSES, NUM_HORIZONS))
|
|
|
|
def test_argument_validation(self) -> None:
|
|
self.assertEqual(parse_runs(["G=results/g"]), [("G", Path("results/g"))])
|
|
with self.assertRaises(ValueError):
|
|
parse_runs(["results/g"])
|
|
with self.assertRaises(ValueError):
|
|
parse_runs(["G=a", "G=b"])
|
|
self.assertEqual(validate_recalls([0.9, 0.7]), [0.7, 0.9])
|
|
with self.assertRaises(ValueError):
|
|
validate_recalls([0.0])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|