add graceful error handling with return of status codes, closes #12

This commit is contained in:
Florian Förster 2025-11-10 15:27:30 +01:00
parent 08f70ee672
commit 4a94815bfe
4 changed files with 2610 additions and 2853 deletions

4
cli.py
View File

@ -12,7 +12,7 @@ class CliArgs:
calib_value_y: float calib_value_y: float
def main() -> None: def main() -> int:
parser = argparse.ArgumentParser( parser = argparse.ArgumentParser(
description=("simple CLI tool to analyse single sensor images for anomalies") description=("simple CLI tool to analyse single sensor images for anomalies")
) )
@ -33,7 +33,7 @@ def main() -> None:
) )
args = cast(CliArgs, parser.parse_args()) args = cast(CliArgs, parser.parse_args())
_interface.sensor_anomalies_detection( return _interface.sensor_anomalies_detection(
args.img_path, args.img_path,
args.calib_value_x, args.calib_value_x,
args.calib_value_y, args.calib_value_y,

View File

@ -1,16 +1,44 @@
"""main pipeline interface for external calls""" """main pipeline interface for external calls"""
from __future__ import annotations
import sys
from typing import TYPE_CHECKING, TextIO
from dopt_sensor_anomalies import detection from dopt_sensor_anomalies import detection
if TYPE_CHECKING:
from dopt_basics.result_pattern import Status
def _print_error_state(
state: Status,
out_stream: TextIO,
) -> None:
if state.ExceptionType is None:
raise RuntimeError("Tried to treat state as error, but no exception is registered")
msg = (
f"During the procedure the following exception occurred:"
f"\nType: {state.ExceptionType.__name__}\nDescription: {state.description}\n"
f"Message: {state.message}"
)
print(msg, flush=True, file=out_stream)
def sensor_anomalies_detection( def sensor_anomalies_detection(
user_img_path: str, user_img_path: str,
pixels_per_metric_X: float, pixels_per_metric_X: float,
pixels_per_metric_Y: float, pixels_per_metric_Y: float,
) -> None: ) -> int:
res = detection.pipeline( res = detection.pipeline(
user_img_path=user_img_path, user_img_path=user_img_path,
pixels_per_metric_X=pixels_per_metric_X, pixels_per_metric_X=pixels_per_metric_X,
pixels_per_metric_Y=pixels_per_metric_Y, pixels_per_metric_Y=pixels_per_metric_Y,
) )
if res.status.code != 0:
_print_error_state(res.status, out_stream=sys.stderr)
return 1
res.unwrap() res.unwrap()
return 0

File diff suppressed because one or more lines are too long

View File

@ -1,23 +1,61 @@
import shutil import shutil
from io import StringIO
from unittest.mock import patch from unittest.mock import patch
import pytest import pytest
from dopt_basics.result_pattern import wrap_result
from dopt_sensor_anomalies import _interface, constants, errors from dopt_sensor_anomalies import _interface, constants
def test_print_error_state_WrongState():
from dopt_basics.result_pattern import STATUS_HANDLER
MSG = "to treat state as error"
with pytest.raises(RuntimeError, match=MSG):
_interface._print_error_state(STATUS_HANDLER.SUCCESS, out_stream=StringIO())
def test_print_error_state(tmp_path):
@wrap_result(100)
def error_func() -> None:
# do something
raise RuntimeError("Oops, error occurred")
err_state = error_func().status
output_file = tmp_path / "t_output.txt"
output_file.touch()
with open(output_file, "w") as stream:
_interface._print_error_state(err_state, stream)
lines: list[str]
with open(output_file, "r") as file:
lines = file.readlines()
assert "following exception" in lines[0]
assert "Type: RuntimeError" in lines[1]
assert "Description:" in lines[2]
assert "Message: Oops, error occurred" in lines[3]
@patch("dopt_sensor_anomalies._find_paths.STOP_FOLDER_NAME", "lib") @patch("dopt_sensor_anomalies._find_paths.STOP_FOLDER_NAME", "lib")
def test_sensor_anomalies_detection_FailImagePath(setup_temp_dir): def test_sensor_anomalies_detection_FailImagePath(setup_temp_dir):
img_path = str(setup_temp_dir / "not-existing.bmp") img_path = str(setup_temp_dir / "not-existing.bmp")
pixels_per_metric_X: float = 0.251 pixels_per_metric_X: float = 0.251
pixels_per_metric_Y: float = 0.251 pixels_per_metric_Y: float = 0.251
MESSAGE = "The provided path seems not to exist" MESSAGE = "The provided path seems not to exist"
with pytest.raises(FileNotFoundError, match=MESSAGE): with patch("sys.stderr", new_callable=StringIO) as mock_err:
_interface.sensor_anomalies_detection( ret = _interface.sensor_anomalies_detection(
img_path, pixels_per_metric_X, pixels_per_metric_Y img_path, pixels_per_metric_X, pixels_per_metric_Y
) )
captured = mock_err.getvalue()
assert ret != 0
assert "FileNotFoundError" in captured
assert MESSAGE in captured
@patch("dopt_sensor_anomalies._find_paths.STOP_FOLDER_NAME", "lib") @patch("dopt_sensor_anomalies._find_paths.STOP_FOLDER_NAME", "lib")
@ -28,10 +66,14 @@ def test_sensor_anomalies_detection_FailElectrodeCount(path_img_with_failure_Ele
MESSAGE = "Number of counted electrodes does not match the" MESSAGE = "Number of counted electrodes does not match the"
with pytest.raises(errors.InvalidElectrodeCount, match=MESSAGE): with patch("sys.stderr", new_callable=StringIO) as mock_err:
_interface.sensor_anomalies_detection( ret = _interface.sensor_anomalies_detection(
img_path, pixels_per_metric_X, pixels_per_metric_Y img_path, pixels_per_metric_X, pixels_per_metric_Y
) )
captured = mock_err.getvalue()
assert ret != 0
assert "InvalidElectrodeCount" in captured
assert MESSAGE in captured
@patch("dopt_sensor_anomalies._find_paths.STOP_FOLDER_NAME", "lib") @patch("dopt_sensor_anomalies._find_paths.STOP_FOLDER_NAME", "lib")
@ -52,8 +94,11 @@ def test_sensor_anomalies_detection_Success(
pixels_per_metric_X: float = 0.251 pixels_per_metric_X: float = 0.251
pixels_per_metric_Y: float = 0.251 pixels_per_metric_Y: float = 0.251
_interface.sensor_anomalies_detection(img_path, pixels_per_metric_X, pixels_per_metric_Y) ret = _interface.sensor_anomalies_detection(
img_path, pixels_per_metric_X, pixels_per_metric_Y
)
assert ret == 0
assert csv_file.exists() assert csv_file.exists()
assert heatmap_file.exists() assert heatmap_file.exists()
target_folder = results_folder / "csharp_interface" target_folder = results_folder / "csharp_interface"