Adds models from common connector code into edge-sdk (#53)

inorbit_edge/models.py

@@ -0,0 +1,126 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+# License: MIT License
+# Copyright 2024 InOrbit, Inc.
+
+# Standard
+import os
+from typing import Optional
+
+# Third-party
+from pydantic import BaseModel, AnyUrl, field_validator, HttpUrl
+
+# InOrbit
+from inorbit_edge.robot import INORBIT_CLOUD_SDK_ROBOT_CONFIG_URL
+
+
+class CameraConfig(BaseModel):
+    """A class representing a camera configuration model.
+
+    This class inherits from the `BaseModel` class. None values should be interpreted as
+    "use the default values".
+
+    Attributes:
+        video_url (AnyUrl): The URL of the video feed from the camera
+        rate (int, optional): The rate at which frames are captured from the camera
+        quality (int, optional): The quality of the captured frames from the camera
+        scaling (float, optional): The scaling factor for the frames from the camera
+    """
+
+    video_url: AnyUrl
+    quality: Optional[int] = None
+    rate: Optional[int] = None
+    scaling: Optional[float] = None
+
+    # noinspection PyMethodParameters
+    @field_validator("quality")
+    def check_quality_range(cls, quality: Optional[float]) -> Optional[float]:
+        """Check if the quality is between 1 and 100.
+
+        This is used for quality.
+
+        Args:
+            quality (int | None): The quality value to be checked
+
+        Raises:
+            ValueError: If the value is not between 1 and 100
+
+        Returns:
+            int | None: The given value if it is between 1 and 100, or None if the input
+                        value was None
+        """
+
+        if quality is not None and not (1 <= quality <= 100):
+            raise ValueError("Must be between 1 and 100")
+        return quality
+
+    # noinspection PyMethodParameters
+    @field_validator("rate", "scaling")
+    def check_positive(cls, value: Optional[float]) -> Optional[float]:
+        """Check if an argument is positive and non-zero.
+
+        This is used for rate and scaling values.
+
+        Args:
+            value (float | None): The value to be checked
+
+        Raises:
+            ValueError: If the value is less than or equal to zero
+
+        Returns:
+            float | None : The given value if it is positive and non-zero, or None if
+                           input value was None
+        """
+        if value is not None and value <= 0:
+            raise ValueError("Must be positive and non-zero")
+        return value
+
+
+class RobotSessionModel(BaseModel):
+    """A class representing InOrbit robot session.
+
+    This class inherits from the `BaseModel` class.
+
+    The following environment variables will be read during instantiation:
+
+    * INORBIT_API_KEY (required): The InOrbit API key
+    * INORBIT_USE_SSL: If SSL should be used (default is true)
+
+    Attributes:
+        robot_id (str): The unique ID of the robot
+        robot_name (str): The name of the robot
+        robot_api_key (str | None, optional): The robot key for InOrbit cloud services
+        api_key (str | None, optional): The InOrbit API token
+        use_ssl (bool, optional): If SSL is used for the InOrbit API connection
+        endpoint (HttpUrl, optional): The URL of the API or inorbit_edge's
+                                      INORBIT_CLOUD_SDK_ROBOT_CONFIG_URL by default
+    """
+
+    robot_id: str
+    robot_name: str
+    robot_api_key: Optional[str] = None
+    api_key: Optional[str] = os.getenv("INORBIT_API_KEY")
+    use_ssl: bool = os.environ.get("INORBIT_USE_SSL", "true").lower() == "true"
+    endpoint: HttpUrl = os.environ.get(
+        "INORBIT_API_URL", INORBIT_CLOUD_SDK_ROBOT_CONFIG_URL
+    )
+
+    # noinspection PyMethodParameters
+    @field_validator("robot_id", "robot_name", "robot_api_key", "api_key")
+    def check_whitespace(cls, value: str) -> str:
+        """Check if the field contains whitespace.
+
+        This is used for the robot_id, robot_name, robot_api_key, and api_key.
+
+        Args:
+            value (str): The field to be checked
+
+        Raises:
+            ValueError: If the field contains whitespace
+
+        Returns:
+            str: The given value if it does not contain whitespaces
+        """
+        if value and any(char.isspace() for char in value):
+            raise ValueError("Whitespaces are not allowed")
+        return value

inorbit_edge/robot.py

@@ -126,7 +126,8 @@ def __init__(self, robot_id, robot_name, api_key=None, **kwargs) -> None:
         self.robot_name = kwargs.get("robot_name", robot_name)
         # The agent version is generated based on the InOrbit Edge SDK version
         self.agent_version = "{}.edgesdk_py".format(inorbit_edge_version)
-        self.endpoint = kwargs.get("endpoint", INORBIT_CLOUD_SDK_ROBOT_CONFIG_URL)
+        # Cast to string to support URL objects
+        self.endpoint = str(kwargs.get("endpoint", INORBIT_CLOUD_SDK_ROBOT_CONFIG_URL))
         # Track robot's current pose
         self._last_pose = None
         # Unique names of configs

inorbit_edge/tests/demo/README.md

@@ -17,7 +17,7 @@ export INORBIT_API_KEY="foobar123"
 # https://api.inorbit.ai/docs/index.html#operation/generateRobotKey)
 export INORBIT_ROBOT_CONFIG_FILE=`pwd`/robots_config_example.yaml
 # Disable SSL for local development only
-export INORBIT_API_USE_SSL="true"
+export INORBIT_USE_SSL="true"
 # Optionally enable video streaming as camera "0"
 export INORBIT_VIDEO_URL=/dev/video0
 

inorbit_edge/tests/demo/example.py

@@ -132,7 +132,7 @@ def my_command_handler(robot_id, command_name, args, options):
 if __name__ == "__main__":
     inorbit_api_endpoint = os.environ.get("INORBIT_URL")
     inorbit_api_url = os.environ.get("INORBIT_API_URL")
-    inorbit_api_use_ssl = os.environ.get("INORBIT_API_USE_SSL")
+    inorbit_api_use_ssl = os.environ.get("INORBIT_USE_SSL")
     inorbit_api_key = os.environ.get("INORBIT_API_KEY")
 
     # For InOrbit Connect (https://connect.inorbit.ai/) certified robots,

inorbit_edge/tests/test_models.py

@@ -0,0 +1,156 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+# License: MIT License
+# Copyright 2024 InOrbit, Inc.
+
+# Standard
+import importlib
+import os
+import re
+import sys
+from unittest import mock
+
+# Third-party
+import pytest
+from pydantic import ValidationError
+
+# InOrbit
+from inorbit_edge.models import RobotSessionModel, CameraConfig
+from inorbit_edge.robot import INORBIT_CLOUD_SDK_ROBOT_CONFIG_URL
+
+
+class TestCameraConfig:
+    def test_quality_validation(self):
+        # Test with valid quality parameter
+        camera1 = CameraConfig(video_url="https://test.com/", quality=50)
+        assert camera1.quality == 50
+
+        # Test with None quality parameter
+        camera2 = CameraConfig(video_url="https://test.com/", quality=None)
+        assert camera2.quality is None
+
+        # Test outside range quality parameter
+        with pytest.raises(ValueError, match="Must be between 1 and 100"):
+            CameraConfig(video_url="https://test.com/", quality=-10)
+
+        with pytest.raises(ValueError, match="Must be between 1 and 100"):
+            CameraConfig(video_url="https://test.com/", quality=150)
+
+    def test_rate_validation(self):
+        # Test with valid rate parameter
+        camera1 = CameraConfig(video_url="https://test.com/", rate=2)
+        assert camera1.rate == 2
+
+        # Test with None rate parameter
+        camera2 = CameraConfig(video_url="https://test.com/", rate=None)
+        assert camera2.rate is None
+
+        # Test with non-positive rate parameter
+        with pytest.raises(ValueError, match="Must be positive and non-zero"):
+            CameraConfig(video_url="https://test.com/", rate=0)
+
+    def test_scaling_validation(self):
+        # Test with valid scaling parameter
+        camera3 = CameraConfig(video_url="https://test.com/", scaling=1.5)
+        assert camera3.scaling == 1.5
+
+        # Test with None scaling parameter
+        camera4 = CameraConfig(video_url="https://test.com/", scaling=None)
+        assert camera4.scaling is None
+
+        # Test with negative scaling parameter
+        with pytest.raises(ValueError, match="Must be positive and non-zero"):
+            CameraConfig(video_url="https://test.com/", scaling=-1.5)
+
+    def test_video_url_validation(self):
+        # Test missing URL
+        error = (
+            "1 validation error for CameraConfig\nvideo_url\n  Field required "
+            "[type=missing, input_value={}, input_type=dict]\n    For further "
+            "information visit https://errors.pydantic.dev/2.7/v/missing"
+        )
+        with pytest.raises(ValidationError, match=re.escape(error)):
+            CameraConfig()
+
+        # Test invalid URL
+        error = (
+            "1 validation error for CameraConfig\nvideo_url\n  Input should be a "
+            "valid URL, relative URL without a base [type=url_parsing, "
+            "input_value='invalid_video_url', input_type=str]\n    For further "
+            "information visit https://errors.pydantic.dev/2.7/v/url_parsing"
+        )
+        with pytest.raises(ValidationError, match=re.escape(error)):
+            CameraConfig(video_url="invalid_video_url")
+
+        # Test valid URL
+        camera = CameraConfig(video_url="https://test.com/")
+        assert str(camera.video_url) == "https://test.com/"
+
+
+class TestRobotSessionModel:
+
+    @pytest.fixture
+    def base_model(self):
+        return {
+            "robot_id": "123",
+            "robot_name": "test_robot",
+            "robot_api_key": "valid_robot_api_key",
+            "api_key": "valid_api_key",
+        }
+
+    def test_model_creation(self, base_model):
+        model = RobotSessionModel(**base_model)
+        assert model.robot_id == base_model["robot_id"]
+        assert model.robot_name == base_model["robot_name"]
+        assert model.robot_api_key == base_model["robot_api_key"]
+        assert model.api_key == base_model["api_key"]
+        assert model.use_ssl is True
+        assert str(model.endpoint) == INORBIT_CLOUD_SDK_ROBOT_CONFIG_URL
+
+    def test_whitespace_validation_robot_id(self, base_model):
+        base_model["robot_id"] = "123 "
+        with pytest.raises(ValidationError, match=r"Whitespaces are not allowed"):
+            RobotSessionModel(**base_model)
+
+    def test_whitespace_validation_robot_name(self, base_model):
+        base_model["robot_name"] = "test robot"
+        with pytest.raises(ValidationError, match=r"Whitespaces are not allowed"):
+            RobotSessionModel(**base_model)
+
+    def test_whitespace_validation_robot_api_key(self, base_model):
+        base_model["robot_api_key"] = "abc def"
+        with pytest.raises(ValidationError, match=r"Whitespaces are not allowed"):
+            RobotSessionModel(**base_model)
+
+    def test_whitespace_validation_api_key(self, base_model):
+        base_model["api_key"] = "abc def"
+        with pytest.raises(ValidationError, match=r"Whitespaces are not allowed"):
+            RobotSessionModel(**base_model)
+
+    @mock.patch.dict(os.environ, {"INORBIT_API_KEY": "env_valid_key"})
+    def test_reads_api_key_from_environment_variable(self, base_model):
+        # Re-import after Mock
+        importlib.reload(sys.modules["inorbit_edge.models"])
+        from inorbit_edge.models import RobotSessionModel
+
+        init_input = {
+            "robot_id": "123",
+            "robot_name": "test_robot",
+            "robot_api_key": "valid_robot_api_key",
+        }
+        model = RobotSessionModel(**init_input)
+        assert model.api_key == "env_valid_key"
+
+    @mock.patch.dict(os.environ, {"INORBIT_USE_SSL": "false"})
+    def test_reads_use_ssl_from_environment_variable(self, base_model):
+        # Re-import after Mock
+        importlib.reload(sys.modules["inorbit_edge.models"])
+        from inorbit_edge.models import RobotSessionModel
+
+        init_input = {
+            "robot_id": "123",
+            "robot_name": "test_robot",
+            "robot_api_key": "valid_robot_api_key",
+        }
+        model = RobotSessionModel(**init_input)
+        assert model.use_ssl is False

inorbit_edge/video.py

@@ -58,7 +58,8 @@ class OpenCVCamera(Camera):
     """Camera implementation backed up by OpenCV"""
 
     def __init__(self, video_url, rate=10, scaling=0.3, quality=35):
-        self.video_url = video_url
+        # Cast to string to support URL objects
+        self.video_url = str(video_url)
         self.capture = None
         self.capture_mutex = threading.Lock()
         self.capture_thread = None