diff --git a/blueye/sdk/camera.py b/blueye/sdk/camera.py index 71ce2e3c..bec66861 100644 --- a/blueye/sdk/camera.py +++ b/blueye/sdk/camera.py @@ -18,6 +18,22 @@ logger = logging.getLogger(__name__) +_RESOLUTION_TO_HEIGHT = { + blueye.protocol.Resolution.RESOLUTION_VGA_480P: 480, + blueye.protocol.Resolution.RESOLUTION_HD_720P: 720, + blueye.protocol.Resolution.RESOLUTION_FULLHD_1080P: 1080, + blueye.protocol.Resolution.RESOLUTION_QHD_2K: 1440, + blueye.protocol.Resolution.RESOLUTION_UHD_4K: 2160, +} +_HEIGHT_TO_RESOLUTION = {height: res for res, height in _RESOLUTION_TO_HEIGHT.items()} + +_FRAMERATE_TO_FPS = { + blueye.protocol.Framerate.FRAMERATE_FPS_25: 25, + blueye.protocol.Framerate.FRAMERATE_FPS_30: 30, + blueye.protocol.Framerate.FRAMERATE_FPS_60: 60, +} +_FPS_TO_FRAMERATE = {fps: framerate for framerate, fps in _FRAMERATE_TO_FPS.items()} + class Tilt: """Handles the camera tilt functionality for the Blueye drone.""" @@ -831,6 +847,17 @@ def __getattr__(self, name): _VERSION_GATED_PARAMS = { "recording_bitrate": "5.0.0", "recording_codec": "5.0.0", + "brightness": "5.0.0", + "contrast": "5.0.0", + "saturation": "5.0.0", + "gamma": "5.0.0", + "sharpness": "5.0.0", + "backlight_compensation": "5.0.0", + "denoise": "5.0.0", + "ehdr_enabled": "5.0.0", + "ehdr_exposure_min_number": "5.0.0", + "ehdr_exposure_max_number": "5.0.0", + "mtu_size": "5.1.0", } def __setattr__(self, name, value): @@ -1061,40 +1088,374 @@ def set_hue(self, hue: int): hue = deprecated_property("get_hue", "set_hue") + def get_gain(self) -> float: + """Get the camera gain. + + Only available on Pioneer/Pro/X1/X3. + + Returns: + The camera ISO gain. + """ + self._update_camera_parameters() + return self._camera_parameters.gain + + def set_gain(self, gain: float): + """Set the camera gain. + + Only available on Pioneer/Pro/X1/X3. + + Args: + gain (float): Set the ISO gain (0..1). + """ + if self._camera_parameters is None: + self._update_camera_parameters() + self._camera_parameters.gain = gain + self._parent_drone._req_rep_client.set_camera_parameters(self._camera_parameters) + + def get_brightness(self) -> int: + """Get the camera brightness. + + Only available on Ultra. + + Returns: + The camera brightness. + """ + self._update_camera_parameters() + return self._camera_parameters.brightness + + def set_brightness(self, brightness: int): + """Set the camera brightness. + + Only available on Ultra. + + Args: + brightness (int): Set the brightness (-10..10), 0 is the default. + + Raises: + RuntimeError: If the connected drone is running a Blunux version older than 5.0.0. + """ + self._parent_drone._verify_required_blunux_version("5.0.0") + if self._camera_parameters is None: + self._update_camera_parameters() + self._camera_parameters.brightness = brightness + self._parent_drone._req_rep_client.set_camera_parameters(self._camera_parameters) + + def get_contrast(self) -> int: + """Get the camera contrast. + + Only available on Ultra. + + Returns: + The camera contrast. + """ + self._update_camera_parameters() + return self._camera_parameters.contrast + + def set_contrast(self, contrast: int): + """Set the camera contrast. + + Only available on Ultra. + + Args: + contrast (int): Set the contrast (-50..50), 0 is the default. + + Raises: + RuntimeError: If the connected drone is running a Blunux version older than 5.0.0. + """ + self._parent_drone._verify_required_blunux_version("5.0.0") + if self._camera_parameters is None: + self._update_camera_parameters() + self._camera_parameters.contrast = contrast + self._parent_drone._req_rep_client.set_camera_parameters(self._camera_parameters) + + def get_saturation(self) -> int: + """Get the camera saturation. + + Only available on Ultra. + + Returns: + The camera saturation. + """ + self._update_camera_parameters() + return self._camera_parameters.saturation + + def set_saturation(self, saturation: int): + """Set the camera saturation. + + Only available on Ultra. + + Args: + saturation (int): Set the saturation (0..50), 8 is the default. + + Raises: + RuntimeError: If the connected drone is running a Blunux version older than 5.0.0. + """ + self._parent_drone._verify_required_blunux_version("5.0.0") + if self._camera_parameters is None: + self._update_camera_parameters() + self._camera_parameters.saturation = saturation + self._parent_drone._req_rep_client.set_camera_parameters(self._camera_parameters) + + def get_gamma(self) -> int: + """Get the camera gamma. + + Only available on Ultra. + + Returns: + The camera gamma. + """ + self._update_camera_parameters() + return self._camera_parameters.gamma + + def set_gamma(self, gamma: int): + """Set the camera gamma. + + Only available on Ultra. + + Args: + gamma (int): Set the gamma (4..79), 22 is the default. + + Raises: + RuntimeError: If the connected drone is running a Blunux version older than 5.0.0. + """ + self._parent_drone._verify_required_blunux_version("5.0.0") + if self._camera_parameters is None: + self._update_camera_parameters() + self._camera_parameters.gamma = gamma + self._parent_drone._req_rep_client.set_camera_parameters(self._camera_parameters) + + def get_sharpness(self) -> int: + """Get the camera sharpness. + + Only available on Ultra. + + Returns: + The camera sharpness. + """ + self._update_camera_parameters() + return self._camera_parameters.sharpness + + def set_sharpness(self, sharpness: int): + """Set the camera sharpness. + + Only available on Ultra. + + Args: + sharpness (int): Set the sharpness (-20..20), -20 is the default. + + Raises: + RuntimeError: If the connected drone is running a Blunux version older than 5.0.0. + """ + self._parent_drone._verify_required_blunux_version("5.0.0") + if self._camera_parameters is None: + self._update_camera_parameters() + self._camera_parameters.sharpness = sharpness + self._parent_drone._req_rep_client.set_camera_parameters(self._camera_parameters) + + def get_backlight_compensation(self) -> int: + """Get the camera backlight compensation. + + Only available on Ultra. + + Returns: + The camera backlight compensation. + """ + self._update_camera_parameters() + return self._camera_parameters.backlight_compensation + + def set_backlight_compensation(self, compensation: int): + """Set the camera backlight compensation. + + Only available on Ultra. + + Args: + compensation (int): Set the backlight compensation (-150..150), 10 is the default. + + Raises: + RuntimeError: If the connected drone is running a Blunux version older than 5.0.0. + """ + self._parent_drone._verify_required_blunux_version("5.0.0") + if self._camera_parameters is None: + self._update_camera_parameters() + self._camera_parameters.backlight_compensation = compensation + self._parent_drone._req_rep_client.set_camera_parameters(self._camera_parameters) + + def get_denoise(self) -> int: + """Get the camera noise reduction. + + Only available on Ultra. + + Returns: + The camera noise reduction. + """ + self._update_camera_parameters() + return self._camera_parameters.denoise + + def set_denoise(self, denoise: int): + """Set the camera noise reduction. + + Only available on Ultra. + + Args: + denoise (int): Set the noise reduction (-20..20), -20 is the default. + + Raises: + RuntimeError: If the connected drone is running a Blunux version older than 5.0.0. + """ + self._parent_drone._verify_required_blunux_version("5.0.0") + if self._camera_parameters is None: + self._update_camera_parameters() + self._camera_parameters.denoise = denoise + self._parent_drone._req_rep_client.set_camera_parameters(self._camera_parameters) + + def is_ehdr_enabled(self) -> bool: + """Get the state of the eHDR mode. + + Only available on Ultra. + + Returns: + The current state of the eHDR mode. + """ + self._update_camera_parameters() + return self._camera_parameters.ehdr_enabled + + def enable_ehdr(self, enable_ehdr: bool): + """Enable or disable the eHDR mode. + + Only available on Ultra. + + Args: + enable_ehdr (bool): True to enable eHDR mode, False to disable it. Enabled by default. + + Raises: + RuntimeError: If the connected drone is running a Blunux version older than 5.0.0. + """ + self._parent_drone._verify_required_blunux_version("5.0.0") + if self._camera_parameters is None: + self._update_camera_parameters() + self._camera_parameters.ehdr_enabled = enable_ehdr + self._parent_drone._req_rep_client.set_camera_parameters(self._camera_parameters) + + def get_ehdr_exposure_min_number(self) -> int: + """Get the minimum number of eHDR frames. + + Only available on Ultra. + + Returns: + The minimum number of eHDR frames. + """ + self._update_camera_parameters() + return self._camera_parameters.ehdr_exposure_min_number + + def set_ehdr_exposure_min_number(self, number: int): + """Set the minimum number of eHDR frames. + + Only available on Ultra. + + Args: + number (int): Set the minimum number of eHDR frames (1..4), 1 is the default. + + Raises: + RuntimeError: If the connected drone is running a Blunux version older than 5.0.0. + """ + self._parent_drone._verify_required_blunux_version("5.0.0") + if self._camera_parameters is None: + self._update_camera_parameters() + self._camera_parameters.ehdr_exposure_min_number = number + self._parent_drone._req_rep_client.set_camera_parameters(self._camera_parameters) + + def get_ehdr_exposure_max_number(self) -> int: + """Get the maximum number of eHDR frames. + + Only available on Ultra. + + Returns: + The maximum number of eHDR frames. + """ + self._update_camera_parameters() + return self._camera_parameters.ehdr_exposure_max_number + + def set_ehdr_exposure_max_number(self, number: int): + """Set the maximum number of eHDR frames. + + Only available on Ultra. + + Args: + number (int): Set the maximum number of eHDR frames (1..4), 2 is the default. Setting + it higher than 2 can reduce the frame rate. + + Raises: + RuntimeError: If the connected drone is running a Blunux version older than 5.0.0. + """ + self._parent_drone._verify_required_blunux_version("5.0.0") + if self._camera_parameters is None: + self._update_camera_parameters() + self._camera_parameters.ehdr_exposure_max_number = number + self._parent_drone._req_rep_client.set_camera_parameters(self._camera_parameters) + def get_resolution(self) -> int: """Get the camera resolution. + Deprecated: + Drones running Blunux 4.4 or newer take the resolution from the stream and recording + resolution fields. Use + [`get_stream_resolution`][blueye.sdk.camera.Camera.get_stream_resolution] and + [`get_recording_resolution`][blueye.sdk.camera.Camera.get_recording_resolution] + instead. + + Raises: + RuntimeError: If the drone reports a resolution the SDK does not recognize. + Returns: The camera resolution. """ + warnings.warn( + "`Camera.get_resolution` is deprecated, use `Camera.get_stream_resolution` or " + "`Camera.get_recording_resolution` instead", + DeprecationWarning, + stacklevel=2, + ) self._update_camera_parameters() - if self._camera_parameters.resolution == blueye.protocol.Resolution.RESOLUTION_HD_720P: - return 720 - elif ( - self._camera_parameters.resolution == blueye.protocol.Resolution.RESOLUTION_FULLHD_1080P - ): - return 1080 + try: + return _RESOLUTION_TO_HEIGHT[self._camera_parameters.resolution] + except KeyError: + raise RuntimeError( + "Drone reported an unsupported resolution: " + f"{self._camera_parameters.resolution.name}" + ) from None def set_resolution(self, resolution: int): """Set the camera resolution. + Deprecated: + Drones running Blunux 4.4 or newer ignore this field when camera parameters are set, + so calling this method has no effect on them. Use + [`set_stream_resolution`][blueye.sdk.camera.Camera.set_stream_resolution] and + [`set_recording_resolution`][blueye.sdk.camera.Camera.set_recording_resolution] + instead. + Args: - resolution (int): Set the camera in vertical pixels. Valid values are 720 or 1080. + resolution (int): Set the camera in vertical pixels. Valid values are 480, 720, 1080, + 1440 or 2160. Raises: - ValueError: If the resolution is not 720 or 1080. - """ - if resolution not in (720, 1080): + ValueError: If the resolution is not one of the valid values. + """ + warnings.warn( + "`Camera.set_resolution` is deprecated and is ignored by drones running Blunux 4.4 or " + "newer, use `Camera.set_stream_resolution` or `Camera.set_recording_resolution` " + "instead", + DeprecationWarning, + stacklevel=2, + ) + if resolution not in _HEIGHT_TO_RESOLUTION: raise ValueError( - f"{resolution} is not a valid resolution. Valid values are 720 or 1080" + f"{resolution} is not a valid resolution. " + "Valid values are 480, 720, 1080, 1440 or 2160" ) if self._camera_parameters is None: self._update_camera_parameters() - if resolution == 720: - self._camera_parameters.resolution = blueye.protocol.Resolution.RESOLUTION_HD_720P - elif resolution == 1080: - self._camera_parameters.resolution = blueye.protocol.Resolution.RESOLUTION_FULLHD_1080P - + self._camera_parameters.resolution = _HEIGHT_TO_RESOLUTION[resolution] self._parent_drone._req_rep_client.set_camera_parameters(self._camera_parameters) resolution = deprecated_property("get_resolution", "set_resolution") @@ -1186,33 +1547,38 @@ def set_recording_resolution(self, resolution: blueye.protocol.Resolution): def get_framerate(self) -> int: """Get the camera frame rate. + Raises: + RuntimeError: If the drone reports a frame rate the SDK does not recognize. + Returns: The camera frame rate. """ self._update_camera_parameters() - if self._camera_parameters.framerate == blueye.protocol.Framerate.FRAMERATE_FPS_25: - return 25 - elif self._camera_parameters.framerate == blueye.protocol.Framerate.FRAMERATE_FPS_30: - return 30 + try: + return _FRAMERATE_TO_FPS[self._camera_parameters.framerate] + except KeyError: + raise RuntimeError( + "Drone reported an unsupported framerate: " + f"{self._camera_parameters.framerate.name}" + ) from None def set_framerate(self, framerate: int): """Set the camera frame rate. Args: - framerate (int): Set the camera frame rate in frames per second. - Valid values are 25 or 30. + framerate (int): Set the camera frame rate in frames per second. Valid values are 25, + 30 or 60. 25 fps is only supported on Pioneer/Pro/X1/X3, and 60 fps only on the + Ultra at 1440p or lower. If the requested frame rate is not supported at the + current resolution the drone reduces it while respecting the resolution. Raises: - ValueError: If the framerate is not 25 or 30. + ValueError: If the frame rate is not 25, 30 or 60. """ - if framerate not in (25, 30): - raise ValueError(f"{framerate} is not a valid framerate. Valid values are 25 or 30") + if framerate not in _FPS_TO_FRAMERATE: + raise ValueError(f"{framerate} is not a valid framerate. Valid values are 25, 30 or 60") if self._camera_parameters is None: self._update_camera_parameters() - if framerate == 25: - self._camera_parameters.framerate = blueye.protocol.Framerate.FRAMERATE_FPS_25 - elif framerate == 30: - self._camera_parameters.framerate = blueye.protocol.Framerate.FRAMERATE_FPS_30 + self._camera_parameters.framerate = _FPS_TO_FRAMERATE[framerate] self._parent_drone._req_rep_client.set_camera_parameters(self._camera_parameters) framerate = deprecated_property("get_framerate", "set_framerate") @@ -1307,6 +1673,31 @@ def set_streaming_protocol(self, protocol: blueye.protocol.StreamingProtocol): streaming_protocol = deprecated_property("get_streaming_protocol", "set_streaming_protocol") + def get_mtu_size(self) -> int: + """Get the network MTU size used for video streaming. + + Returns: + The MTU size in bytes (0 if the drone default is used). + """ + self._update_camera_parameters() + return self._camera_parameters.mtu_size + + def set_mtu_size(self, mtu_size: int): + """Set the network MTU size used for video streaming. + + Args: + mtu_size (int): Set the MTU size in bytes (68..65535). The Blueye app allows values + between 500 and 1460. A value of 0 means the drone uses its default of 1400. + + Raises: + RuntimeError: If the connected drone is running a Blunux version older than 5.1.0. + """ + self._parent_drone._verify_required_blunux_version("5.1.0") + if self._camera_parameters is None: + self._update_camera_parameters() + self._camera_parameters.mtu_size = mtu_size + self._parent_drone._req_rep_client.set_camera_parameters(self._camera_parameters) + def get_record_time(self) -> Optional[int]: """Get the duration of the current camera recording. diff --git a/tests/conftest.py b/tests/conftest.py index 7cbb3329..53909166 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,3 +1,5 @@ +import time + import blueye.protocol as bp import pytest @@ -13,6 +15,23 @@ def real_drone(): return blueye.sdk.Drone() +@pytest.fixture(scope="class") +def drone_model(real_drone) -> bp.Model: + """Fixture that reports the model of the connected drone + + Used to skip integration tests for camera parameters the connected hardware does not + support. Returns MODEL_UNSPECIFIED if no drone info is received, which is also what + drones older than the introduction of the model field report. + """ + deadline = time.monotonic() + 3 + while time.monotonic() < deadline: + drone_info = real_drone.telemetry.get(bp.DroneInfoTel) + if drone_info is not None: + return drone_info.drone_info.model + time.sleep(0.1) + return bp.Model.MODEL_UNSPECIFIED + + @pytest.fixture def mocked_requests(requests_mock): import json @@ -60,7 +79,24 @@ def mocked_ctrl_client(mocker): @pytest.fixture def mocked_telemetry_client(mocker): - return mocker.patch("blueye.sdk.drone.TelemetryClient", autospec=True) + """Patch the telemetry subscriber so unit tests never read telemetry off the network. + + Without this the drone built by `mocked_drone` opens a real ZMQ subscriber against the + drone IP, and any drone on the same network fills its state within a few hundred + milliseconds. That makes every "returns None when no telemetry has been received" test + fail at random, depending on which one happens to run late enough to receive something. + + The mock keeps a real dict for `_state` and looks messages up in it, raising KeyError for + the ones that are missing, so tests can keep seeding telemetry by assigning to + `drone._telemetry_watcher._state[SomeTel]`. + """ + patched = mocker.patch("blueye.sdk.drone.TelemetryClient", autospec=True) + instance = patched.return_value + instance._state = {} + # Read the attribute on every call rather than closing over the dict, because the real + # get() looks up self._state and several tests rebind _state to a fresh dict. + instance.get.side_effect = lambda key: instance._state[key] + return patched @pytest.fixture @@ -79,6 +115,7 @@ def mocked_drone( mocker, mocked_requests, mocked_ctrl_client, + mocked_telemetry_client, mocked_watchdog_publisher, mocked_req_rep_client, ): diff --git a/tests/test_camera.py b/tests/test_camera.py index bf2a07d4..73b640f1 100644 --- a/tests/test_camera.py +++ b/tests/test_camera.py @@ -84,6 +84,76 @@ def test_recording_resolution_invalid_type(mocked_camera): mocked_camera.set_recording_resolution("invalid_resolution") +@pytest.mark.parametrize( + "enum_value, expected", + [ + (bp.Resolution.RESOLUTION_VGA_480P, 480), + (bp.Resolution.RESOLUTION_HD_720P, 720), + (bp.Resolution.RESOLUTION_FULLHD_1080P, 1080), + (bp.Resolution.RESOLUTION_QHD_2K, 1440), + (bp.Resolution.RESOLUTION_UHD_4K, 2160), + ], +) +@pytest.mark.filterwarnings("ignore::DeprecationWarning") +def test_resolution_getter(mocked_camera, enum_value, expected): + mocked_camera._parent_drone._req_rep_client.get_camera_parameters.return_value = ( + bp.CameraParameters(resolution=enum_value) + ) + assert mocked_camera.get_resolution() == expected + + +@pytest.mark.parametrize( + "value, expected_enum", + [ + (480, bp.Resolution.RESOLUTION_VGA_480P), + (720, bp.Resolution.RESOLUTION_HD_720P), + (1080, bp.Resolution.RESOLUTION_FULLHD_1080P), + (1440, bp.Resolution.RESOLUTION_QHD_2K), + (2160, bp.Resolution.RESOLUTION_UHD_4K), + ], +) +@pytest.mark.filterwarnings("ignore::DeprecationWarning") +def test_resolution_setter(mocked_camera, value, expected_enum): + mocked_camera._camera_parameters = bp.CameraParameters() + mocked_camera.set_resolution(value) + assert mocked_camera._camera_parameters.resolution == expected_enum + mocked_camera._parent_drone._req_rep_client.set_camera_parameters.assert_called_once_with( + mocked_camera._camera_parameters + ) + + +@pytest.mark.filterwarnings("ignore::DeprecationWarning") +def test_resolution_setter_invalid_value(mocked_camera): + with pytest.raises(ValueError): + mocked_camera.set_resolution(600) + + +def test_streaming_protocol_getter(mocked_camera): + mocked_camera._parent_drone._req_rep_client.get_camera_parameters.return_value = ( + bp.CameraParameters(streaming_protocol=bp.StreamingProtocol.STREAMING_PROTOCOL_RTSP_H264) + ) + assert ( + mocked_camera.get_streaming_protocol() == bp.StreamingProtocol.STREAMING_PROTOCOL_RTSP_H264 + ) + + +def test_streaming_protocol_setter(mocked_camera): + mocked_camera._camera_parameters = bp.CameraParameters() + mocked_camera.set_streaming_protocol(bp.StreamingProtocol.STREAMING_PROTOCOL_RTSP_MJPEG) + assert ( + mocked_camera._camera_parameters.streaming_protocol + == bp.StreamingProtocol.STREAMING_PROTOCOL_RTSP_MJPEG + ) + mocked_camera._parent_drone._req_rep_client.set_camera_parameters.assert_called_once_with( + mocked_camera._camera_parameters + ) + + +def test_streaming_protocol_invalid_type(mocked_camera): + with pytest.raises(ValueError): + mocked_camera.set_streaming_protocol("invalid_protocol") + + def test_old_drones_use_resolution_field(mocked_camera): # Set the version to a value that does not support separate recording resolution mocked_camera._parent_drone.software_version_short = "4.3" @@ -150,3 +220,259 @@ def test_configure_passes_timeout(mocked_camera): mocked_camera._parent_drone._req_rep_client.set_camera_parameters.call_args[1]["timeout"] == 5.0 ) + + +@pytest.fixture +def mocked_ultra_camera(mocked_drone: Drone): + """A camera on a drone running a Blunux version that supports all camera parameters.""" + from blueye.sdk.camera import Camera + + mocked_drone.software_version_short = "5.1.0" + return Camera(mocked_drone) + + +def test_get_resolution_warns_deprecation(mocked_camera): + mocked_camera._parent_drone._req_rep_client.get_camera_parameters.return_value = ( + bp.CameraParameters(resolution=bp.Resolution.RESOLUTION_HD_720P) + ) + with pytest.warns(DeprecationWarning, match="get_stream_resolution"): + assert mocked_camera.get_resolution() == 720 + + +def test_set_resolution_warns_deprecation(mocked_camera): + mocked_camera._camera_parameters = bp.CameraParameters() + with pytest.warns(DeprecationWarning, match="set_stream_resolution"): + mocked_camera.set_resolution(720) + + +@pytest.mark.filterwarnings("ignore::DeprecationWarning") +def test_resolution_getter_raises_on_unknown_resolution(mocked_camera): + mocked_camera._parent_drone._req_rep_client.get_camera_parameters.return_value = ( + bp.CameraParameters(resolution=bp.Resolution.RESOLUTION_UNSPECIFIED) + ) + with pytest.raises(RuntimeError): + mocked_camera.get_resolution() + + +@pytest.mark.parametrize( + "enum_value, expected", + [ + (bp.Framerate.FRAMERATE_FPS_25, 25), + (bp.Framerate.FRAMERATE_FPS_30, 30), + (bp.Framerate.FRAMERATE_FPS_60, 60), + ], +) +def test_framerate_getter(mocked_camera, enum_value, expected): + mocked_camera._parent_drone._req_rep_client.get_camera_parameters.return_value = ( + bp.CameraParameters(framerate=enum_value) + ) + assert mocked_camera.get_framerate() == expected + + +@pytest.mark.parametrize( + "value, expected_enum", + [ + (25, bp.Framerate.FRAMERATE_FPS_25), + (30, bp.Framerate.FRAMERATE_FPS_30), + (60, bp.Framerate.FRAMERATE_FPS_60), + ], +) +def test_framerate_setter(mocked_camera, value, expected_enum): + mocked_camera._camera_parameters = bp.CameraParameters() + mocked_camera.set_framerate(value) + assert mocked_camera._camera_parameters.framerate == expected_enum + mocked_camera._parent_drone._req_rep_client.set_camera_parameters.assert_called_once_with( + mocked_camera._camera_parameters + ) + + +def test_framerate_setter_invalid_value(mocked_camera): + with pytest.raises(ValueError): + mocked_camera.set_framerate(24) + + +def test_framerate_getter_raises_on_unknown_framerate(mocked_camera): + mocked_camera._parent_drone._req_rep_client.get_camera_parameters.return_value = ( + bp.CameraParameters(framerate=bp.Framerate.FRAMERATE_UNSPECIFIED) + ) + with pytest.raises(RuntimeError): + mocked_camera.get_framerate() + + +# (getter, setter, CameraParameters field name, test value) for the Ultra-only image parameters +ULTRA_IMAGE_PARAMETERS = [ + ("get_brightness", "set_brightness", "brightness", 5), + ("get_contrast", "set_contrast", "contrast", 20), + ("get_saturation", "set_saturation", "saturation", 10), + ("get_gamma", "set_gamma", "gamma", 30), + ("get_sharpness", "set_sharpness", "sharpness", -10), + ("get_backlight_compensation", "set_backlight_compensation", "backlight_compensation", 100), + ("get_denoise", "set_denoise", "denoise", -5), + ("is_ehdr_enabled", "enable_ehdr", "ehdr_enabled", True), + ( + "get_ehdr_exposure_min_number", + "set_ehdr_exposure_min_number", + "ehdr_exposure_min_number", + 2, + ), + ( + "get_ehdr_exposure_max_number", + "set_ehdr_exposure_max_number", + "ehdr_exposure_max_number", + 3, + ), +] + + +@pytest.mark.parametrize("getter, setter, field_name, value", ULTRA_IMAGE_PARAMETERS) +def test_ultra_image_parameter_getter(mocked_ultra_camera, getter, setter, field_name, value): + mocked_ultra_camera._parent_drone._req_rep_client.get_camera_parameters.return_value = ( + bp.CameraParameters(**{field_name: value}) + ) + assert getattr(mocked_ultra_camera, getter)() == value + + +@pytest.mark.parametrize("getter, setter, field_name, value", ULTRA_IMAGE_PARAMETERS) +def test_ultra_image_parameter_setter(mocked_ultra_camera, getter, setter, field_name, value): + mocked_ultra_camera._camera_parameters = bp.CameraParameters() + getattr(mocked_ultra_camera, setter)(value) + assert getattr(mocked_ultra_camera._camera_parameters, field_name) == value + mocked_ultra_camera._parent_drone._req_rep_client.set_camera_parameters.assert_called_once_with( + mocked_ultra_camera._camera_parameters + ) + + +@pytest.mark.parametrize("getter, setter, field_name, value", ULTRA_IMAGE_PARAMETERS) +def test_ultra_image_parameter_setter_requires_blunux_5( + mocked_camera, getter, setter, field_name, value +): + # The mocked_camera fixture runs Blunux 4.4.1 + with pytest.raises(RuntimeError): + getattr(mocked_camera, setter)(value) + + +def test_gain_getter(mocked_camera): + mocked_camera._parent_drone._req_rep_client.get_camera_parameters.return_value = ( + bp.CameraParameters(gain=0.5) + ) + assert mocked_camera.get_gain() == pytest.approx(0.5) + + +def test_gain_setter(mocked_camera): + mocked_camera._camera_parameters = bp.CameraParameters() + mocked_camera.set_gain(0.5) + assert mocked_camera._camera_parameters.gain == pytest.approx(0.5) + mocked_camera._parent_drone._req_rep_client.set_camera_parameters.assert_called_once_with( + mocked_camera._camera_parameters + ) + + +def test_mtu_size_getter(mocked_ultra_camera): + mocked_ultra_camera._parent_drone._req_rep_client.get_camera_parameters.return_value = ( + bp.CameraParameters(mtu_size=1400) + ) + assert mocked_ultra_camera.get_mtu_size() == 1400 + + +def test_mtu_size_setter(mocked_ultra_camera): + mocked_ultra_camera._camera_parameters = bp.CameraParameters() + mocked_ultra_camera.set_mtu_size(1200) + assert mocked_ultra_camera._camera_parameters.mtu_size == 1200 + mocked_ultra_camera._parent_drone._req_rep_client.set_camera_parameters.assert_called_once_with( + mocked_ultra_camera._camera_parameters + ) + + +def test_mtu_size_setter_requires_blunux_5_1(mocked_camera): + # The mocked_camera fixture runs Blunux 4.4.1 + with pytest.raises(RuntimeError): + mocked_camera.set_mtu_size(1200) + + +def test_configure_rejects_ultra_image_parameters_on_old_drone(mocked_camera): + """Batched changes should honour the same Blunux version requirements as the setters.""" + mocked_camera._parent_drone._req_rep_client.get_camera_parameters.return_value = ( + bp.CameraParameters() + ) + with pytest.raises(RuntimeError): + with mocked_camera.configure() as params: + params.brightness = 5 + mocked_camera._parent_drone._req_rep_client.set_camera_parameters.assert_not_called() + + +def test_configure_rejects_mtu_size_on_blunux_5_0(mocked_drone: Drone): + from blueye.sdk.camera import Camera + + mocked_drone.software_version_short = "5.0.0" + camera = Camera(mocked_drone) + camera._parent_drone._req_rep_client.get_camera_parameters.return_value = bp.CameraParameters() + with pytest.raises(RuntimeError): + with camera.configure() as params: + params.mtu_size = 1200 + camera._parent_drone._req_rep_client.set_camera_parameters.assert_not_called() + + +def test_recording_bitrate_getter(mocked_ultra_camera): + mocked_ultra_camera._parent_drone._req_rep_client.get_camera_parameters.return_value = ( + bp.CameraParameters(recording_bitrate=24_000_000) + ) + assert mocked_ultra_camera.get_recording_bitrate() == 24_000_000 + + +def test_recording_bitrate_setter(mocked_ultra_camera): + mocked_ultra_camera._camera_parameters = bp.CameraParameters(recording_bitrate=0) + + mocked_ultra_camera.set_recording_bitrate(24_000_000) + + mocked_ultra_camera._parent_drone._req_rep_client.set_camera_parameters.assert_called_once_with( + bp.CameraParameters(recording_bitrate=24_000_000) + ) + assert mocked_ultra_camera._camera_parameters.recording_bitrate == 24_000_000 + + +def test_recording_bitrate_setter_requires_blunux_5(mocked_camera): + with pytest.raises(RuntimeError): + mocked_camera.set_recording_bitrate(24_000_000) + + +def test_parameter_getter_refreshes_from_the_drone(mocked_ultra_camera): + """A getter must ask the drone, never answer from the cache. + + Every getter method calls _update_camera_parameters() first, so a value + that changed since the last request - because another client set it, or + because the drone resolved an "automatic" request to a concrete number - is + reported as it is now and not as it was last written. + """ + mocked_ultra_camera._camera_parameters = bp.CameraParameters(recording_bitrate=0) + mocked_ultra_camera._parent_drone._req_rep_client.get_camera_parameters.return_value = ( + bp.CameraParameters(recording_bitrate=14_000_000) + ) + + assert mocked_ultra_camera.get_recording_bitrate() == 14_000_000 + mocked_ultra_camera._parent_drone._req_rep_client.get_camera_parameters.assert_called_once() + + +def test_setter_on_a_cold_camera_fetches_current_parameters_first(mocked_ultra_camera): + """A setter on a camera that has not talked to the drone yet must fetch the + current parameters before it sends. + + set_camera_parameters carries the whole CameraParameters struct, so without + the lazy fetch the first setter after construction would send a + default-constructed one and zero every field the caller never touched - + resolution, framerate and codec among them. + """ + assert mocked_ultra_camera._camera_parameters is None + mocked_ultra_camera._parent_drone._req_rep_client.get_camera_parameters.return_value = ( + bp.CameraParameters( + stream_resolution=bp.Resolution.RESOLUTION_FULLHD_1080P, + framerate=bp.Framerate.FRAMERATE_FPS_30, + ) + ) + + mocked_ultra_camera.set_recording_bitrate(24_000_000) + + mocked_ultra_camera._parent_drone._req_rep_client.get_camera_parameters.assert_called_once() + sent = mocked_ultra_camera._parent_drone._req_rep_client.set_camera_parameters.call_args[0][0] + assert sent.recording_bitrate == 24_000_000 + assert sent.stream_resolution == bp.Resolution.RESOLUTION_FULLHD_1080P + assert sent.framerate == bp.Framerate.FRAMERATE_FPS_30 diff --git a/tests/test_integration.py b/tests/test_integration.py index a340d600..3445a982 100644 --- a/tests/test_integration.py +++ b/tests/test_integration.py @@ -1,6 +1,16 @@ from time import time +import blueye.protocol as bp import pytest +from packaging import version + +ALL_RESOLUTIONS = [ + bp.Resolution.RESOLUTION_VGA_480P, + bp.Resolution.RESOLUTION_HD_720P, + bp.Resolution.RESOLUTION_FULLHD_1080P, + bp.Resolution.RESOLUTION_QHD_2K, + bp.Resolution.RESOLUTION_UHD_4K, +] def polling_assert_with_timeout(getter, value_to_wait_for, timeout): @@ -46,7 +56,13 @@ def test_camera_record_time(self, real_drone): real_drone.camera.set_recording(True) polling_assert_with_timeout(real_drone.camera.get_record_time, 1, 3) - def test_camera_bitrate(self, real_drone): + def test_camera_bitrate(self, real_drone, drone_model): + if drone_model == bp.Model.MODEL_X3_ULTRA: + pytest.xfail( + "The Ultra applies the stream bitrate but always reports it as 0. The drone " + "fills h264_bitrate from the camera control node, and the Ultra camera does not " + "encode H264 - the RTSP server does, and its bitrate never reaches the reply" + ) _ = real_drone.camera.get_bitrate() real_drone.camera.set_bitrate(2000000) polling_assert_with_timeout(real_drone.camera.get_bitrate, 2000000, 1) @@ -67,23 +83,78 @@ def test_camera_whitebalance(self, real_drone): real_drone.camera.set_whitebalance(3400) polling_assert_with_timeout(real_drone.camera.get_whitebalance, 3400, 1) - def test_camera_hue(self, real_drone): + def test_camera_hue(self, real_drone, drone_model): + if drone_model == bp.Model.MODEL_X3_ULTRA: + pytest.skip( + "Hue is only available on Pioneer/Pro/X1/X3, the Ultra camera has no hue control" + ) _ = real_drone.camera.get_hue() real_drone.camera.set_hue(20) polling_assert_with_timeout(real_drone.camera.get_hue, 20, 1) real_drone.camera.set_hue(30) polling_assert_with_timeout(real_drone.camera.get_hue, 30, 1) + @pytest.mark.filterwarnings("ignore::DeprecationWarning") def test_camera_resolution(self, real_drone): + if version.parse(real_drone.software_version_short) >= version.parse("4.4"): + pytest.xfail( + "Drones running Blunux 4.4 or newer ignore the deprecated resolution field when " + "camera parameters are set, use the stream/recording resolution methods instead" + ) _ = real_drone.camera.get_resolution() real_drone.camera.set_resolution(720) polling_assert_with_timeout(real_drone.camera.get_resolution, 720, 1) real_drone.camera.set_resolution(1080) polling_assert_with_timeout(real_drone.camera.get_resolution, 1080, 1) - def test_camera_framerate(self, real_drone): + @pytest.mark.parametrize("resolution", ALL_RESOLUTIONS, ids=lambda r: r.name) + def test_camera_stream_resolution(self, real_drone, resolution): + original_resolution = real_drone.camera.get_stream_resolution() + try: + real_drone.camera.set_stream_resolution(resolution) + polling_assert_with_timeout(real_drone.camera.get_stream_resolution, resolution, 3) + finally: + real_drone.camera.set_stream_resolution(original_resolution) + + @pytest.mark.parametrize("resolution", ALL_RESOLUTIONS, ids=lambda r: r.name) + def test_camera_recording_resolution(self, real_drone, resolution): + original_resolution = real_drone.camera.get_recording_resolution() + try: + real_drone.camera.set_recording_resolution(resolution) + polling_assert_with_timeout(real_drone.camera.get_recording_resolution, resolution, 3) + finally: + real_drone.camera.set_recording_resolution(original_resolution) + + def test_camera_framerate(self, real_drone, drone_model): + if drone_model == bp.Model.MODEL_X3_ULTRA: + pytest.skip( + "The Ultra only applies 25 fps to the recording pipeline, and the reported frame " + "rate comes from the stream pipeline, which stays at 30. See " + "test_camera_framerate_60_fps for the frame rate the Ultra does support" + ) _ = real_drone.camera.get_framerate() real_drone.camera.set_framerate(25) polling_assert_with_timeout(real_drone.camera.get_framerate, 25, 1) real_drone.camera.set_framerate(30) polling_assert_with_timeout(real_drone.camera.get_framerate, 30, 1) + + def test_camera_framerate_60_fps(self, real_drone, drone_model): + """60 fps is only supported on the Ultra, and only at 1440p or lower. + + The drone caps the frame rate against the highest of the stream and recording + resolution, so both must be lowered before requesting 60 fps. + """ + if drone_model != bp.Model.MODEL_X3_ULTRA: + pytest.skip("60 fps is only supported on the Ultra") + original_stream_resolution = real_drone.camera.get_stream_resolution() + original_recording_resolution = real_drone.camera.get_recording_resolution() + original_framerate = real_drone.camera.get_framerate() + try: + real_drone.camera.set_stream_resolution(bp.Resolution.RESOLUTION_FULLHD_1080P) + real_drone.camera.set_recording_resolution(bp.Resolution.RESOLUTION_FULLHD_1080P) + real_drone.camera.set_framerate(60) + polling_assert_with_timeout(real_drone.camera.get_framerate, 60, 3) + finally: + real_drone.camera.set_framerate(original_framerate) + real_drone.camera.set_stream_resolution(original_stream_resolution) + real_drone.camera.set_recording_resolution(original_recording_resolution) diff --git a/tests/test_sdk.py b/tests/test_sdk.py index 0ad9ebf8..5cb2f1ba 100644 --- a/tests/test_sdk.py +++ b/tests/test_sdk.py @@ -1,6 +1,6 @@ import json from time import time -from unittest.mock import Mock, PropertyMock +from unittest.mock import Mock, NonCallableMock, PropertyMock import blueye.protocol as bp import pytest @@ -33,6 +33,20 @@ def test_angle_conversion(self, mocked_drone, old_angle, new_angle): assert pose["yaw"] == new_angle +def test_mocked_drone_is_isolated_from_live_telemetry(mocked_drone): + """Unit tests must never read telemetry off the network. + + `mocked_drone` used to build a real TelemetryClient, which subscribes to the drone IP. With + a drone on the same network that fills the state within a few hundred milliseconds, and the + tests asserting that a getter returns None when no telemetry has been received start failing + at random. + """ + # A spec'd mock passes isinstance() against the class it specs, so check for the mock + # itself rather than asserting the watcher is not a TelemetryClient. + assert isinstance(mocked_drone._telemetry_watcher, NonCallableMock) + assert mocked_drone.telemetry.get(bp.ControlModeTel) is None + + def test_zmq_connection_error(mocked_drone): mocked_drone._req_rep_client.ping.side_effect = bp.exceptions.ResponseTimeout with pytest.raises(ConnectionError):