diff --git a/.bumpversion.cfg b/.bumpversion.cfg index e60a5a0..545a844 100644 --- a/.bumpversion.cfg +++ b/.bumpversion.cfg @@ -1,6 +1,6 @@ [bumpversion] commit = true -current_version = 2.1.0 +current_version = 3.0.0 tag = true [bumpversion:file(version):setup.py] diff --git a/.github/workflows/test-and-lint.yml b/.github/workflows/test-and-lint.yml index c5ac2a4..21f1eab 100644 --- a/.github/workflows/test-and-lint.yml +++ b/.github/workflows/test-and-lint.yml @@ -4,6 +4,7 @@ on: pull_request: branches: - main + - next jobs: lint-and-test: diff --git a/inorbit_edge/__init__.py b/inorbit_edge/__init__.py index bc808c1..6df0243 100644 --- a/inorbit_edge/__init__.py +++ b/inorbit_edge/__init__.py @@ -6,7 +6,7 @@ __email__ = "support@inorbit.ai" # Do not edit this string manually, always use bumpversion # Details in CONTRIBUTING.md -__version__ = "2.1.0" +__version__ = "3.0.0" def get_module_version(): diff --git a/inorbit_edge/metrics.py b/inorbit_edge/metrics.py index 3e33c77..e79904b 100644 --- a/inorbit_edge/metrics.py +++ b/inorbit_edge/metrics.py @@ -40,8 +40,6 @@ import logging import re -from deprecated import deprecated - logger = logging.getLogger(__name__) @@ -302,17 +300,3 @@ def sync_wrapper(*args, **kwargs): return sync_wrapper return decorator - - -@deprecated( - version="2.0.2", - reason=("use with_counter_metric(), which now auto-detects async functions"), -) -def with_counter_metric_async(metric): - """Deprecated alias for :func:`with_counter_metric`. - - Prefer ``@with_counter_metric(...)``, which now detects async functions - automatically. - """ - - return with_counter_metric(metric) diff --git a/inorbit_edge/models.py b/inorbit_edge/models.py index 29627bb..6752a6a 100644 --- a/inorbit_edge/models.py +++ b/inorbit_edge/models.py @@ -1,138 +1,138 @@ -#!/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, INORBIT_REST_API_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) - * INORBIT_API_URL: The URL of the API (default is - inorbit_edge.robot.INORBIT_CLOUD_SDK_ROBOT_CONFIG_URL) - * INORBIT_REST_API_URL: The URL of the InOrbit REST API (default is - inorbit_edge.robot.INORBIT_REST_API_URL) - - Attributes: - robot_id (str): The unique ID of the robot - robot_name (str): The name of the robot - robot_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 - rest_api_endpoint (HttpUrl, optional): The URL of the InOrbit REST API. - INORBIT_REST_API_URL by default. - account_id (str, optional): The account ID of the robot owner. Required for - applying configurations to the robot. - """ - - robot_id: str - robot_name: str - robot_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 - ) - rest_api_endpoint: Optional[HttpUrl] = os.environ.get( - "INORBIT_REST_API_URL", INORBIT_REST_API_URL - ) - account_id: Optional[str] = None - - # noinspection PyMethodParameters - @field_validator("robot_id", "robot_name", "robot_key", "api_key", "account_id") - def check_whitespace(cls, value: str) -> str: - """Check if the field contains whitespace. - - This is used for the robot_id, robot_name, robot_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 +#!/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 AnyUrl, BaseModel, HttpUrl, field_validator + +# InOrbit +from inorbit_edge.robot import ( + INORBIT_CLOUD_SDK_ROBOT_CONFIG_URL, + INORBIT_DEFAULT_API_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) + * INORBIT_CONNECTION_CONFIG_URL: The URL of the MQTT configuration API + (default is inorbit_edge.robot.INORBIT_CLOUD_SDK_ROBOT_CONFIG_URL) + * INORBIT_API_URL: The URL of the InOrbit REST API (default is + inorbit_edge.robot.INORBIT_DEFAULT_API_URL) + + Attributes: + robot_id (str): The unique ID of the robot + robot_name (str): The name of the robot + robot_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 + rest_api_endpoint (HttpUrl, optional): The URL of the InOrbit REST API. + INORBIT_DEFAULT_API_URL by default. + """ + + robot_id: str + robot_name: str + robot_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_CONNECTION_CONFIG_URL", INORBIT_CLOUD_SDK_ROBOT_CONFIG_URL + ) + rest_api_endpoint: Optional[HttpUrl] = os.environ.get( + "INORBIT_API_URL", INORBIT_DEFAULT_API_URL + ) + + # noinspection PyMethodParameters + @field_validator("robot_id", "robot_name", "robot_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_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 diff --git a/inorbit_edge/robot.py b/inorbit_edge/robot.py index 1ac15fa..0f41fa2 100644 --- a/inorbit_edge/robot.py +++ b/inorbit_edge/robot.py @@ -1,76 +1,75 @@ #!/usr/bin/env python # -*- coding: utf-8 -*- import io -import zlib -from dataclasses import dataclass, field, asdict import json -from typing import Tuple, Optional, List, Dict - -from inorbit_edge import __version__ as inorbit_edge_version -from inorbit_edge.types import Pose, SpatialTolerance -from inorbit_edge.metrics import ( - attrs_from_self, - with_counter_metric, - publish_map_counter, - publish_camera_frame_counter, - publish_pose_counter, - publish_key_values_counter, - publish_system_stats_counter, - publish_odometry_counter, - publish_laser_counter, - publish_path_counter, -) -import os import logging -import paho.mqtt.client as mqtt -from PIL import Image -from urllib.parse import urlsplit -import socks +import math +import os +import re import ssl +import subprocess import threading +import time +import zlib +from dataclasses import asdict, dataclass, field +from typing import Dict, List, Optional, Tuple +from urllib.parse import urlsplit + +import certifi +import paho.mqtt.client as mqtt +import requests +import socks import yaml +from PIL import Image +from inorbit_edge import __version__ as inorbit_edge_version +from inorbit_edge.commands import ( + COMMAND_CUSTOM_COMMAND, + COMMAND_INITIAL_POSE, + COMMAND_MESSAGE, + COMMAND_NAV_GOAL, +) from inorbit_edge.inorbit_pb2 import ( + CameraMessage, + CustomCommandRosMessage, CustomDataMessage, + CustomScriptCommandMessage, + CustomScriptStatusMessage, + Echo, KeyValueCustomElement, + LaserMessage, LocationAndPoseMessage, + MapMessage, + MapRequest, OdometryDataMessage, - LaserMessage, + PathDataMessage, PathPoint, RobotPath, - PathDataMessage, - Echo, - CustomScriptCommandMessage, - CustomScriptStatusMessage, - CustomCommandRosMessage, - CameraMessage, SystemStatsMessage, - MapMessage, - MapRequest, ) -from inorbit_edge.video import CameraStreamer, Camera -from inorbit_edge.missions import MissionsModule -from inorbit_edge.commands import ( - COMMAND_INITIAL_POSE, - COMMAND_NAV_GOAL, - COMMAND_CUSTOM_COMMAND, - COMMAND_MESSAGE, +from inorbit_edge.metrics import ( + attrs_from_self, + publish_camera_frame_counter, + publish_key_values_counter, + publish_laser_counter, + publish_map_counter, + publish_odometry_counter, + publish_path_counter, + publish_pose_counter, + publish_system_stats_counter, + with_counter_metric, ) -import time -import requests -import math +from inorbit_edge.missions import MissionsModule +from inorbit_edge.types import Pose, SpatialTolerance from inorbit_edge.utils import ( + calculate_pose_delta, encode_floating_point_list, reduce_path, - calculate_pose_delta, ) -import certifi -import subprocess -import re -from deprecated import deprecated +from inorbit_edge.video import Camera, CameraStreamer INORBIT_CLOUD_SDK_ROBOT_CONFIG_URL = "https://control.inorbit.ai/cloud_sdk_robot_config" -INORBIT_REST_API_URL = "https://api.inorbit.ai" +INORBIT_DEFAULT_API_URL = "https://api.inorbit.ai" MQTT_SUBTOPIC_POSE = "ros/loc/data2" MQTT_SUBTOPIC_PATH = "ros/loc/path" @@ -299,7 +298,6 @@ def discard_next_delta(self): class RobotSession: - def __init__(self, robot_id, robot_name, api_key=None, **kwargs) -> None: """Initialize a robot session. @@ -315,9 +313,7 @@ def __init__(self, robot_id, robot_name, api_key=None, **kwargs) -> None: endpoint (str): InOrbit URL. Defaults: INORBIT_CLOUD_SDK_ROBOT_CONFIG_URL. use_ssl (bool): Configures MQTT client to use SSL. Defaults: True. rest_api_endpoint (str): The URL of the InOrbit REST API. - Defaults: INORBIT_REST_API_URL. - account_id (str): The account ID of the robot owner. Required for applying - configurations to the robot. + Defaults: INORBIT_DEFAULT_API_URL. keepalive_secs (int): Keepalive for MQTT connection (seconds). Default: 10. estimate_distance_linear (bool): Whether to publish an estimate value for linear_distance based on poses when the value is not provided on a @@ -350,10 +346,10 @@ def __init__(self, robot_id, robot_name, api_key=None, **kwargs) -> None: self.use_ssl = kwargs.get("use_ssl", True) # InOrbit REST API endpoint self.inorbit_rest_api_endpoint = kwargs.get( - "rest_api_endpoint", INORBIT_REST_API_URL + "rest_api_endpoint", INORBIT_DEFAULT_API_URL ) - # Account the robot belongs to. Used for REST API calls. - self.account_id = kwargs.get("account_id") + # Account ID, lazily fetched from REST API via get_account_id() + self._account_id: str | None = None # Use TCP transport by default. The client will use websockets # transport if the environment variable HTTP_PROXY is set. @@ -1639,36 +1635,81 @@ def publish_path( self.publish_protobuf(MQTT_SUBTOPIC_PATH, msg) + def get_account_id(self) -> str: + """Fetch and cache the account ID from the InOrbit REST API. + + Calls ``GET /user`` using the configured API key. The result is + cached so subsequent calls do not make additional HTTP requests. + + Returns: + str: The account ID associated with the API key. + + Raises: + ValueError: If no ``api_key`` is configured. + ValueError: If the API key is associated with zero or multiple accounts. + requests.HTTPError: If the REST API request fails. + """ + if self._account_id is not None: + return self._account_id + + api_key = self.api_key + if not api_key: + raise ValueError( + "An api_key is required to fetch account info from the REST API" + ) + + response = requests.get( + f"{self.inorbit_rest_api_endpoint}/user", + headers={"x-auth-inorbit-app-key": api_key}, + ) + response.raise_for_status() + + account_ids = response.json().get("accountIds") or [] + if not isinstance(account_ids, list): + raise ValueError( + "Unexpected response from the REST API: accountIds is not a list" + ) + if len(account_ids) == 0: + raise ValueError("No account IDs found for the authenticated API key") + if len(account_ids) > 1: + raise ValueError( + "Multiple account IDs found for the authenticated API key. " + "This is not supported by the Edge SDK" + ) + + account_id: str = account_ids[0] + self._account_id = account_id + self.logger.debug(f"Fetched account ID: {account_id}") + return account_id + def apply_footprint(self, spec: RobotFootprintSpec): """Creates and applies a RobotFootprint configuration at the robot level scope. - Calling this method one time applies a persistent footprint configuration. - Note that configurations can be applied at other scopes as well. - Refer to the REST APIs documentation for more information. + + Calling this method once applies a persistent footprint configuration. + The account ID is fetched automatically from the REST API. Args: spec (RobotFootprintSpec): Robot footprint configuration spec. - Will be added to the `spec` field of the RobotFootprint configuration. Raises: - ValueError: If the account ID is not set. + ValueError: If no api_key is configured or account cannot be resolved. HTTPError: If the request to the InOrbit REST API fails. - Returns: - None - References: https://api.inorbit.ai/docs/index.html """ + api_key = self.api_key + if not api_key: + raise ValueError("An api_key is required to apply footprint configuration") - if not self.account_id: - raise ValueError("Account ID is required to set robot footprint") + account_id = self.get_account_id() body = { "apiVersion": "v0.1", "kind": "RobotFootprint", "metadata": { "id": "all", - "scope": f"robot/{self.account_id}/{self.robot_id}", + "scope": f"robot/{account_id}/{self.robot_id}", }, "spec": asdict(spec), } @@ -1676,7 +1717,7 @@ def apply_footprint(self, spec: RobotFootprintSpec): res = requests.post( f"{self.inorbit_rest_api_endpoint}/configuration/apply", json=body, - headers={"x-auth-inorbit-app-key": f"{self.api_key}"}, + headers={"x-auth-inorbit-app-key": api_key}, ) res.raise_for_status() @@ -1805,11 +1846,3 @@ def free_robot_session(self, robot_id): sess = self.get_session(robot_id) sess.disconnect() del self.robot_sessions[robot_id] - - @deprecated( - version="1.7.2", - reason="use RobotSessionFactory.register_command_callback() instead", - ) - def register_command_callback(self, callback): - """Registers a callback to be called when a command is received""" - self.robot_session_factory.register_command_callback(callback) diff --git a/inorbit_edge/tests/demo/Dockerfile b/inorbit_edge/tests/demo/Dockerfile index 09c5cb5..dcfacea 100644 --- a/inorbit_edge/tests/demo/Dockerfile +++ b/inorbit_edge/tests/demo/Dockerfile @@ -7,7 +7,6 @@ # -e INORBIT_URL=... \ # -e INORBIT_API_URL=... \ # -e INORBIT_API_KEY=... \ -# -e INORBIT_ACCOUNT_ID=... \ # -e INORBIT_USE_SSL=true \ # -e INORBIT_ROBOT_ID_PREFIX=$(hostname) \ # inorbit-edge-sdk-demo diff --git a/inorbit_edge/tests/demo/README.md b/inorbit_edge/tests/demo/README.md index 0431355..62483be 100644 --- a/inorbit_edge/tests/demo/README.md +++ b/inorbit_edge/tests/demo/README.md @@ -14,10 +14,9 @@ cd /path/to/edge-sdk-python pip install -e '.[video,telemetry]' cd inorbit_edge/tests/demo -export INORBIT_URL="https://control.inorbit.ai" +export INORBIT_CONNECTION_CONFIG_URL="https://control.inorbit.ai" export INORBIT_API_URL="https://api.inorbit.ai" export INORBIT_API_KEY="foobar123" -export INORBIT_ACCOUNT_ID="account123" export INORBIT_ROBOT_ID_PREFIX="$(hostname)" # TLS to the MQTT broker defaults to on. For a local broker without TLS: INORBIT_USE_SSL=false # Optionally enable video streaming as camera "0" @@ -52,10 +51,9 @@ metrics on the host: ```bash docker run --rm -p 9464:9464 \ -v "$PWD/inorbit_edge/tests/demo:/demo:ro" \ - -e INORBIT_URL=... \ + -e INORBIT_CONNECTION_CONFIG_URL=... \ -e INORBIT_API_URL=... \ -e INORBIT_API_KEY=... \ - -e INORBIT_ACCOUNT_ID=... \ -e INORBIT_ROBOT_ID_PREFIX=$(hostname) \ inorbit-edge-sdk-demo ``` diff --git a/inorbit_edge/tests/demo/example.py b/inorbit_edge/tests/demo/example.py index c37b236..1ce9d45 100644 --- a/inorbit_edge/tests/demo/example.py +++ b/inorbit_edge/tests/demo/example.py @@ -206,9 +206,8 @@ def main(): radius=0.2, ) - inorbit_api_endpoint = _required_env("INORBIT_URL") + inorbit_api_endpoint = _required_env("INORBIT_CONNECTION_CONFIG_URL") inorbit_api_url = _required_env("INORBIT_API_URL") - inorbit_account_id = _required_env("INORBIT_ACCOUNT_ID") inorbit_api_key = _required_env("INORBIT_API_KEY") # If configured stream video as if it was a robot camera @@ -224,7 +223,6 @@ def main(): rest_api_endpoint=inorbit_api_url, api_key=inorbit_api_key, use_ssl=_mqtt_use_ssl(), - account_id=inorbit_account_id, ) robot_session_factory.register_command_callback(log_command) robot_session_factory.register_command_callback(my_command_handler) diff --git a/inorbit_edge/tests/test_metrics.py b/inorbit_edge/tests/test_metrics.py index 8f85d8a..004d396 100644 --- a/inorbit_edge/tests/test_metrics.py +++ b/inorbit_edge/tests/test_metrics.py @@ -2,7 +2,6 @@ # SPDX-License-Identifier: MIT import asyncio -import warnings import pytest @@ -80,22 +79,6 @@ def f(): assert counter.calls == [(1, {})] -def test_with_counter_metric_async_alias_emits_deprecation_warning(): - counter = _RecordingCounter() - - with warnings.catch_warnings(record=True) as captured: - warnings.simplefilter("always") - - @edge_metrics.with_counter_metric_async(counter) - async def f(): - return 1 - - asyncio.run(f()) - - assert any(issubclass(w.category, DeprecationWarning) for w in captured) - assert counter.calls == [(1, {})] - - def test_wrapped_function_preserves_name_and_docstring(): counter = _RecordingCounter() diff --git a/inorbit_edge/tests/test_models.py b/inorbit_edge/tests/test_models.py index 1543e54..3202794 100644 --- a/inorbit_edge/tests/test_models.py +++ b/inorbit_edge/tests/test_models.py @@ -125,11 +125,6 @@ def test_whitespace_validation_api_key(self, base_model): with pytest.raises(ValidationError, match=r"Whitespaces are not allowed"): RobotSessionModel(**base_model) - def test_whitespace_validation_account_id(self, base_model): - base_model["account_id"] = "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 diff --git a/inorbit_edge/tests/test_robot_session.py b/inorbit_edge/tests/test_robot_session.py index 6c3cdc4..71d3c7e 100644 --- a/inorbit_edge/tests/test_robot_session.py +++ b/inorbit_edge/tests/test_robot_session.py @@ -4,13 +4,19 @@ import math import os from unittest.mock import MagicMock + import pytest from requests import HTTPError -from inorbit_edge.robot import RobotSession, RobotFootprintSpec, RobotMap -from inorbit_edge.robot import INORBIT_CLOUD_SDK_ROBOT_CONFIG_URL, INORBIT_REST_API_URL from inorbit_edge import get_module_version -from inorbit_edge.inorbit_pb2 import MapMessage, RobotPath, PathDataMessage, PathPoint +from inorbit_edge.inorbit_pb2 import MapMessage, PathDataMessage, PathPoint, RobotPath +from inorbit_edge.robot import ( + INORBIT_CLOUD_SDK_ROBOT_CONFIG_URL, + INORBIT_DEFAULT_API_URL, + RobotFootprintSpec, + RobotMap, + RobotSession, +) def test_robot_session_init(monkeypatch, mock_sleep): @@ -124,8 +130,12 @@ def test_method_throttling(mock_sleep): def test_apply_footprint(requests_mock, mock_sleep): + requests_mock.get( + f"{INORBIT_DEFAULT_API_URL}/user", + json={"userId": "user_abc", "name": "Test User", "accountIds": ["account_123"]}, + ) adapter = requests_mock.post( - f"{INORBIT_REST_API_URL}/configuration/apply", + f"{INORBIT_DEFAULT_API_URL}/configuration/apply", json={"operationStatus": "SUCCESS"}, ) footprint = RobotFootprintSpec( @@ -138,21 +148,10 @@ def test_apply_footprint(requests_mock, mock_sleep): radius=0.2, ) - # Missing account_id - robot_session = RobotSession( - robot_id="id_123", - robot_name="name_123", - api_key="apikey_123", - ) - with pytest.raises(ValueError): - robot_session.apply_footprint(footprint) - - # Successful request robot_session = RobotSession( robot_id="id_123", robot_name="name_123", api_key="apikey_123", - account_id="account_123", ) robot_session.apply_footprint(footprint) assert adapter.called_once @@ -174,12 +173,80 @@ def test_apply_footprint(requests_mock, mock_sleep): }, } - # HTTP error - requests_mock.post(f"{INORBIT_REST_API_URL}/configuration/apply", status_code=400) + # HTTP error on apply + requests_mock.post( + f"{INORBIT_DEFAULT_API_URL}/configuration/apply", status_code=400 + ) with pytest.raises(HTTPError): robot_session.apply_footprint(footprint) +def test_get_account_id(requests_mock, mock_sleep): + requests_mock.get( + f"{INORBIT_DEFAULT_API_URL}/user", + json={"userId": "user_abc", "name": "Test User", "accountIds": ["account_123"]}, + ) + robot_session = RobotSession( + robot_id="id_123", + robot_name="name_123", + api_key="apikey_123", + ) + assert robot_session.get_account_id() == "account_123" + # Second call uses cache -- no extra HTTP request + assert robot_session.get_account_id() == "account_123" + user_requests = [h for h in requests_mock.request_history if "/user" in h.path] + assert len(user_requests) == 1 + + +def test_get_account_id_no_api_key(mock_sleep): + robot_session = RobotSession( + robot_id="id_123", + robot_name="name_123", + robot_key="robotkey_123", + ) + with pytest.raises(ValueError, match="api_key is required"): + robot_session.get_account_id() + + +def test_get_account_id_empty_accounts(requests_mock, mock_sleep): + requests_mock.get( + f"{INORBIT_DEFAULT_API_URL}/user", + json={"userId": "user_abc", "name": "Test User", "accountIds": []}, + ) + robot_session = RobotSession( + robot_id="id_123", + robot_name="name_123", + api_key="apikey_123", + ) + with pytest.raises(ValueError, match="No account IDs found"): + robot_session.get_account_id() + + +def test_get_account_id_multiple_accounts(requests_mock, mock_sleep): + requests_mock.get( + f"{INORBIT_DEFAULT_API_URL}/user", + json={"userId": "user_abc", "name": "Test", "accountIds": ["a1", "a2"]}, + ) + robot_session = RobotSession( + robot_id="id_123", + robot_name="name_123", + api_key="apikey_123", + ) + with pytest.raises(ValueError, match="Multiple account IDs"): + robot_session.get_account_id() + + +def test_get_account_id_api_error(requests_mock, mock_sleep): + requests_mock.get(f"{INORBIT_DEFAULT_API_URL}/user", status_code=401) + robot_session = RobotSession( + robot_id="id_123", + robot_name="name_123", + api_key="apikey_123", + ) + with pytest.raises(HTTPError): + robot_session.get_account_id() + + def test_robot_map_data(): # Test with good file robot_map = RobotMap( diff --git a/requirements.txt b/requirements.txt index 8482288..40359a2 100644 --- a/requirements.txt +++ b/requirements.txt @@ -6,5 +6,4 @@ pydantic>=2.6,<3.0 pysocks>=1.7,<2.0 protobuf~=5.29.6 certifi>=2024.2 -deprecated>=1.2,<2.0 rdp2~=1.1.2 diff --git a/setup.py b/setup.py index 03d804d..deb5425 100644 --- a/setup.py +++ b/setup.py @@ -8,7 +8,7 @@ from setuptools import find_packages, setup # Do not edit manually, always use bumpversion (see CONTRIBUTING.rst) -VERSION = "2.1.0" +VERSION = "3.0.0" GITHUB_ORG = "https://github.com/inorbit-ai" GITHUB_REPO = f"{GITHUB_ORG}/edge-sdk-python" @@ -48,7 +48,7 @@ ], description="InOrbit Edge SDK for Python", # Do not edit manually, always use bumpversion (see CONTRIBUTING.rst) - download_url=f"{GITHUB_REPO}/archive/refs/tags/v2.1.0.zip", + download_url=f"{GITHUB_REPO}/archive/refs/tags/v3.0.0.zip", extras_require={ "video": requirements["video"], "telemetry": requirements["telemetry"],