From 84ee504ba56b5de976930babc0dd233d6803b4ff Mon Sep 17 00:00:00 2001 From: Jordan Yates Date: Tue, 16 Jun 2026 14:35:15 +1000 Subject: [PATCH 1/3] api_client: regenerate client Regenerate client from the API schema as of 18/06/2026. Signed-off-by: Jordan Yates --- ...device_application_updates_by_device_id.py | 191 +++++++++++++++ ..._device_application_update_by_device_id.py | 208 ++++++++++++++++ ...t_device_application_state_by_device_id.py | 166 +++++++++++++ ...ation_update_by_device_id_and_update_id.py | 181 ++++++++++++++ ...device_application_updates_by_device_id.py | 227 ++++++++++++++++++ src/infuse_iot/api_client/models/__init__.py | 10 + .../models/device_application_state.py | 106 ++++++++ .../models/device_application_update.py | 164 +++++++++++++ .../device_application_update_and_message.py | 186 ++++++++++++++ .../device_application_update_status.py | 11 + .../models/new_device_application_update.py | 61 +++++ 11 files changed, 1511 insertions(+) create mode 100644 src/infuse_iot/api_client/api/device/cancel_pending_device_application_updates_by_device_id.py create mode 100644 src/infuse_iot/api_client/api/device/create_device_application_update_by_device_id.py create mode 100644 src/infuse_iot/api_client/api/device/get_device_application_state_by_device_id.py create mode 100644 src/infuse_iot/api_client/api/device/get_device_application_update_by_device_id_and_update_id.py create mode 100644 src/infuse_iot/api_client/api/device/get_device_application_updates_by_device_id.py create mode 100644 src/infuse_iot/api_client/models/device_application_state.py create mode 100644 src/infuse_iot/api_client/models/device_application_update.py create mode 100644 src/infuse_iot/api_client/models/device_application_update_and_message.py create mode 100644 src/infuse_iot/api_client/models/device_application_update_status.py create mode 100644 src/infuse_iot/api_client/models/new_device_application_update.py diff --git a/src/infuse_iot/api_client/api/device/cancel_pending_device_application_updates_by_device_id.py b/src/infuse_iot/api_client/api/device/cancel_pending_device_application_updates_by_device_id.py new file mode 100644 index 0000000..f568319 --- /dev/null +++ b/src/infuse_iot/api_client/api/device/cancel_pending_device_application_updates_by_device_id.py @@ -0,0 +1,191 @@ +from http import HTTPStatus +from typing import Any, cast +from urllib.parse import quote + +import httpx + +from ... import errors +from ...client import AuthenticatedClient, Client +from ...models.device_application_update import DeviceApplicationUpdate +from ...models.error import Error +from ...types import Response + + +def _get_kwargs( + device_id: str, +) -> dict[str, Any]: + + _kwargs: dict[str, Any] = { + "method": "delete", + "url": "/device/deviceId/{device_id}/application/updates".format( + device_id=quote(str(device_id), safe=""), + ), + } + + return _kwargs + + +def _parse_response( + *, client: AuthenticatedClient | Client, response: httpx.Response +) -> Any | DeviceApplicationUpdate | Error | None: + if response.status_code == 200: + response_200 = DeviceApplicationUpdate.from_dict(response.json()) + + return response_200 + + if response.status_code == 204: + response_204 = cast(Any, None) + return response_204 + + if response.status_code == 403: + response_403 = Error.from_dict(response.json()) + + return response_403 + + if response.status_code == 404: + response_404 = Error.from_dict(response.json()) + + return response_404 + + if response.status_code == 500: + response_500 = Error.from_dict(response.json()) + + return response_500 + + if client.raise_on_unexpected_status: + raise errors.UnexpectedStatus(response.status_code, response.content) + else: + return None + + +def _build_response( + *, client: AuthenticatedClient | Client, response: httpx.Response +) -> Response[Any | DeviceApplicationUpdate | Error]: + return Response( + status_code=HTTPStatus(response.status_code), + content=response.content, + headers=response.headers, + parsed=_parse_response(client=client, response=response), + ) + + +def sync_detailed( + device_id: str, + *, + client: AuthenticatedClient | Client, +) -> Response[Any | DeviceApplicationUpdate | Error]: + """Cancel pending device application update by DeviceID. + + Cancel a pending device application update by DeviceID. If an RPC has already been sent to the + device for the pending update, this will not cancel the update on the device, but it will prevent + any further attempts to update the device application until a new update is created. + + Args: + device_id (str): + + Raises: + errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True. + httpx.TimeoutException: If the request takes longer than Client.timeout. + + Returns: + Response[Any | DeviceApplicationUpdate | Error] + """ + + kwargs = _get_kwargs( + device_id=device_id, + ) + + response = client.get_httpx_client().request( + **kwargs, + ) + + return _build_response(client=client, response=response) + + +def sync( + device_id: str, + *, + client: AuthenticatedClient | Client, +) -> Any | DeviceApplicationUpdate | Error | None: + """Cancel pending device application update by DeviceID. + + Cancel a pending device application update by DeviceID. If an RPC has already been sent to the + device for the pending update, this will not cancel the update on the device, but it will prevent + any further attempts to update the device application until a new update is created. + + Args: + device_id (str): + + Raises: + errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True. + httpx.TimeoutException: If the request takes longer than Client.timeout. + + Returns: + Any | DeviceApplicationUpdate | Error + """ + + return sync_detailed( + device_id=device_id, + client=client, + ).parsed + + +async def asyncio_detailed( + device_id: str, + *, + client: AuthenticatedClient | Client, +) -> Response[Any | DeviceApplicationUpdate | Error]: + """Cancel pending device application update by DeviceID. + + Cancel a pending device application update by DeviceID. If an RPC has already been sent to the + device for the pending update, this will not cancel the update on the device, but it will prevent + any further attempts to update the device application until a new update is created. + + Args: + device_id (str): + + Raises: + errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True. + httpx.TimeoutException: If the request takes longer than Client.timeout. + + Returns: + Response[Any | DeviceApplicationUpdate | Error] + """ + + kwargs = _get_kwargs( + device_id=device_id, + ) + + response = await client.get_async_httpx_client().request(**kwargs) + + return _build_response(client=client, response=response) + + +async def asyncio( + device_id: str, + *, + client: AuthenticatedClient | Client, +) -> Any | DeviceApplicationUpdate | Error | None: + """Cancel pending device application update by DeviceID. + + Cancel a pending device application update by DeviceID. If an RPC has already been sent to the + device for the pending update, this will not cancel the update on the device, but it will prevent + any further attempts to update the device application until a new update is created. + + Args: + device_id (str): + + Raises: + errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True. + httpx.TimeoutException: If the request takes longer than Client.timeout. + + Returns: + Any | DeviceApplicationUpdate | Error + """ + + return ( + await asyncio_detailed( + device_id=device_id, + client=client, + ) + ).parsed diff --git a/src/infuse_iot/api_client/api/device/create_device_application_update_by_device_id.py b/src/infuse_iot/api_client/api/device/create_device_application_update_by_device_id.py new file mode 100644 index 0000000..f2f4859 --- /dev/null +++ b/src/infuse_iot/api_client/api/device/create_device_application_update_by_device_id.py @@ -0,0 +1,208 @@ +from http import HTTPStatus +from typing import Any +from urllib.parse import quote + +import httpx + +from ... import errors +from ...client import AuthenticatedClient, Client +from ...models.device_application_state import DeviceApplicationState +from ...models.device_application_update import DeviceApplicationUpdate +from ...models.error import Error +from ...models.new_device_application_update import NewDeviceApplicationUpdate +from ...types import Response + + +def _get_kwargs( + device_id: str, + *, + body: NewDeviceApplicationUpdate, +) -> dict[str, Any]: + headers: dict[str, Any] = {} + + _kwargs: dict[str, Any] = { + "method": "post", + "url": "/device/deviceId/{device_id}/application/updates".format( + device_id=quote(str(device_id), safe=""), + ), + } + + _kwargs["json"] = body.to_dict() + + headers["Content-Type"] = "application/json" + + _kwargs["headers"] = headers + return _kwargs + + +def _parse_response( + *, client: AuthenticatedClient | Client, response: httpx.Response +) -> DeviceApplicationState | DeviceApplicationUpdate | Error | None: + if response.status_code == 200: + response_200 = DeviceApplicationState.from_dict(response.json()) + + return response_200 + + if response.status_code == 201: + response_201 = DeviceApplicationUpdate.from_dict(response.json()) + + return response_201 + + if response.status_code == 400: + response_400 = Error.from_dict(response.json()) + + return response_400 + + if response.status_code == 403: + response_403 = Error.from_dict(response.json()) + + return response_403 + + if response.status_code == 404: + response_404 = Error.from_dict(response.json()) + + return response_404 + + if response.status_code == 409: + response_409 = Error.from_dict(response.json()) + + return response_409 + + if response.status_code == 500: + response_500 = Error.from_dict(response.json()) + + return response_500 + + if client.raise_on_unexpected_status: + raise errors.UnexpectedStatus(response.status_code, response.content) + else: + return None + + +def _build_response( + *, client: AuthenticatedClient | Client, response: httpx.Response +) -> Response[DeviceApplicationState | DeviceApplicationUpdate | Error]: + return Response( + status_code=HTTPStatus(response.status_code), + content=response.content, + headers=response.headers, + parsed=_parse_response(client=client, response=response), + ) + + +def sync_detailed( + device_id: str, + *, + client: AuthenticatedClient | Client, + body: NewDeviceApplicationUpdate, +) -> Response[DeviceApplicationState | DeviceApplicationUpdate | Error]: + """Create a device application update by DeviceID + + Args: + device_id (str): + body (NewDeviceApplicationUpdate): + + Raises: + errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True. + httpx.TimeoutException: If the request takes longer than Client.timeout. + + Returns: + Response[DeviceApplicationState | DeviceApplicationUpdate | Error] + """ + + kwargs = _get_kwargs( + device_id=device_id, + body=body, + ) + + response = client.get_httpx_client().request( + **kwargs, + ) + + return _build_response(client=client, response=response) + + +def sync( + device_id: str, + *, + client: AuthenticatedClient | Client, + body: NewDeviceApplicationUpdate, +) -> DeviceApplicationState | DeviceApplicationUpdate | Error | None: + """Create a device application update by DeviceID + + Args: + device_id (str): + body (NewDeviceApplicationUpdate): + + Raises: + errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True. + httpx.TimeoutException: If the request takes longer than Client.timeout. + + Returns: + DeviceApplicationState | DeviceApplicationUpdate | Error + """ + + return sync_detailed( + device_id=device_id, + client=client, + body=body, + ).parsed + + +async def asyncio_detailed( + device_id: str, + *, + client: AuthenticatedClient | Client, + body: NewDeviceApplicationUpdate, +) -> Response[DeviceApplicationState | DeviceApplicationUpdate | Error]: + """Create a device application update by DeviceID + + Args: + device_id (str): + body (NewDeviceApplicationUpdate): + + Raises: + errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True. + httpx.TimeoutException: If the request takes longer than Client.timeout. + + Returns: + Response[DeviceApplicationState | DeviceApplicationUpdate | Error] + """ + + kwargs = _get_kwargs( + device_id=device_id, + body=body, + ) + + response = await client.get_async_httpx_client().request(**kwargs) + + return _build_response(client=client, response=response) + + +async def asyncio( + device_id: str, + *, + client: AuthenticatedClient | Client, + body: NewDeviceApplicationUpdate, +) -> DeviceApplicationState | DeviceApplicationUpdate | Error | None: + """Create a device application update by DeviceID + + Args: + device_id (str): + body (NewDeviceApplicationUpdate): + + Raises: + errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True. + httpx.TimeoutException: If the request takes longer than Client.timeout. + + Returns: + DeviceApplicationState | DeviceApplicationUpdate | Error + """ + + return ( + await asyncio_detailed( + device_id=device_id, + client=client, + body=body, + ) + ).parsed diff --git a/src/infuse_iot/api_client/api/device/get_device_application_state_by_device_id.py b/src/infuse_iot/api_client/api/device/get_device_application_state_by_device_id.py new file mode 100644 index 0000000..358a628 --- /dev/null +++ b/src/infuse_iot/api_client/api/device/get_device_application_state_by_device_id.py @@ -0,0 +1,166 @@ +from http import HTTPStatus +from typing import Any +from urllib.parse import quote + +import httpx + +from ... import errors +from ...client import AuthenticatedClient, Client +from ...models.device_application_state import DeviceApplicationState +from ...models.error import Error +from ...types import Response + + +def _get_kwargs( + device_id: str, +) -> dict[str, Any]: + + _kwargs: dict[str, Any] = { + "method": "get", + "url": "/device/deviceId/{device_id}/application/state".format( + device_id=quote(str(device_id), safe=""), + ), + } + + return _kwargs + + +def _parse_response( + *, client: AuthenticatedClient | Client, response: httpx.Response +) -> DeviceApplicationState | Error | None: + if response.status_code == 200: + response_200 = DeviceApplicationState.from_dict(response.json()) + + return response_200 + + if response.status_code == 404: + response_404 = Error.from_dict(response.json()) + + return response_404 + + if response.status_code == 500: + response_500 = Error.from_dict(response.json()) + + return response_500 + + if client.raise_on_unexpected_status: + raise errors.UnexpectedStatus(response.status_code, response.content) + else: + return None + + +def _build_response( + *, client: AuthenticatedClient | Client, response: httpx.Response +) -> Response[DeviceApplicationState | Error]: + return Response( + status_code=HTTPStatus(response.status_code), + content=response.content, + headers=response.headers, + parsed=_parse_response(client=client, response=response), + ) + + +def sync_detailed( + device_id: str, + *, + client: AuthenticatedClient | Client, +) -> Response[DeviceApplicationState | Error]: + """Get device application state by DeviceID + + Args: + device_id (str): + + Raises: + errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True. + httpx.TimeoutException: If the request takes longer than Client.timeout. + + Returns: + Response[DeviceApplicationState | Error] + """ + + kwargs = _get_kwargs( + device_id=device_id, + ) + + response = client.get_httpx_client().request( + **kwargs, + ) + + return _build_response(client=client, response=response) + + +def sync( + device_id: str, + *, + client: AuthenticatedClient | Client, +) -> DeviceApplicationState | Error | None: + """Get device application state by DeviceID + + Args: + device_id (str): + + Raises: + errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True. + httpx.TimeoutException: If the request takes longer than Client.timeout. + + Returns: + DeviceApplicationState | Error + """ + + return sync_detailed( + device_id=device_id, + client=client, + ).parsed + + +async def asyncio_detailed( + device_id: str, + *, + client: AuthenticatedClient | Client, +) -> Response[DeviceApplicationState | Error]: + """Get device application state by DeviceID + + Args: + device_id (str): + + Raises: + errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True. + httpx.TimeoutException: If the request takes longer than Client.timeout. + + Returns: + Response[DeviceApplicationState | Error] + """ + + kwargs = _get_kwargs( + device_id=device_id, + ) + + response = await client.get_async_httpx_client().request(**kwargs) + + return _build_response(client=client, response=response) + + +async def asyncio( + device_id: str, + *, + client: AuthenticatedClient | Client, +) -> DeviceApplicationState | Error | None: + """Get device application state by DeviceID + + Args: + device_id (str): + + Raises: + errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True. + httpx.TimeoutException: If the request takes longer than Client.timeout. + + Returns: + DeviceApplicationState | Error + """ + + return ( + await asyncio_detailed( + device_id=device_id, + client=client, + ) + ).parsed diff --git a/src/infuse_iot/api_client/api/device/get_device_application_update_by_device_id_and_update_id.py b/src/infuse_iot/api_client/api/device/get_device_application_update_by_device_id_and_update_id.py new file mode 100644 index 0000000..15c217f --- /dev/null +++ b/src/infuse_iot/api_client/api/device/get_device_application_update_by_device_id_and_update_id.py @@ -0,0 +1,181 @@ +from http import HTTPStatus +from typing import Any +from urllib.parse import quote +from uuid import UUID + +import httpx + +from ... import errors +from ...client import AuthenticatedClient, Client +from ...models.device_application_update_and_message import DeviceApplicationUpdateAndMessage +from ...models.error import Error +from ...types import Response + + +def _get_kwargs( + device_id: str, + update_id: UUID, +) -> dict[str, Any]: + + _kwargs: dict[str, Any] = { + "method": "get", + "url": "/device/deviceId/{device_id}/application/updates/{update_id}".format( + device_id=quote(str(device_id), safe=""), + update_id=quote(str(update_id), safe=""), + ), + } + + return _kwargs + + +def _parse_response( + *, client: AuthenticatedClient | Client, response: httpx.Response +) -> DeviceApplicationUpdateAndMessage | Error | None: + if response.status_code == 200: + response_200 = DeviceApplicationUpdateAndMessage.from_dict(response.json()) + + return response_200 + + if response.status_code == 404: + response_404 = Error.from_dict(response.json()) + + return response_404 + + if response.status_code == 500: + response_500 = Error.from_dict(response.json()) + + return response_500 + + if client.raise_on_unexpected_status: + raise errors.UnexpectedStatus(response.status_code, response.content) + else: + return None + + +def _build_response( + *, client: AuthenticatedClient | Client, response: httpx.Response +) -> Response[DeviceApplicationUpdateAndMessage | Error]: + return Response( + status_code=HTTPStatus(response.status_code), + content=response.content, + headers=response.headers, + parsed=_parse_response(client=client, response=response), + ) + + +def sync_detailed( + device_id: str, + update_id: UUID, + *, + client: AuthenticatedClient | Client, +) -> Response[DeviceApplicationUpdateAndMessage | Error]: + """Get a device application update with downlink message by DeviceID and UpdateID + + Args: + device_id (str): + update_id (UUID): + + Raises: + errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True. + httpx.TimeoutException: If the request takes longer than Client.timeout. + + Returns: + Response[DeviceApplicationUpdateAndMessage | Error] + """ + + kwargs = _get_kwargs( + device_id=device_id, + update_id=update_id, + ) + + response = client.get_httpx_client().request( + **kwargs, + ) + + return _build_response(client=client, response=response) + + +def sync( + device_id: str, + update_id: UUID, + *, + client: AuthenticatedClient | Client, +) -> DeviceApplicationUpdateAndMessage | Error | None: + """Get a device application update with downlink message by DeviceID and UpdateID + + Args: + device_id (str): + update_id (UUID): + + Raises: + errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True. + httpx.TimeoutException: If the request takes longer than Client.timeout. + + Returns: + DeviceApplicationUpdateAndMessage | Error + """ + + return sync_detailed( + device_id=device_id, + update_id=update_id, + client=client, + ).parsed + + +async def asyncio_detailed( + device_id: str, + update_id: UUID, + *, + client: AuthenticatedClient | Client, +) -> Response[DeviceApplicationUpdateAndMessage | Error]: + """Get a device application update with downlink message by DeviceID and UpdateID + + Args: + device_id (str): + update_id (UUID): + + Raises: + errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True. + httpx.TimeoutException: If the request takes longer than Client.timeout. + + Returns: + Response[DeviceApplicationUpdateAndMessage | Error] + """ + + kwargs = _get_kwargs( + device_id=device_id, + update_id=update_id, + ) + + response = await client.get_async_httpx_client().request(**kwargs) + + return _build_response(client=client, response=response) + + +async def asyncio( + device_id: str, + update_id: UUID, + *, + client: AuthenticatedClient | Client, +) -> DeviceApplicationUpdateAndMessage | Error | None: + """Get a device application update with downlink message by DeviceID and UpdateID + + Args: + device_id (str): + update_id (UUID): + + Raises: + errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True. + httpx.TimeoutException: If the request takes longer than Client.timeout. + + Returns: + DeviceApplicationUpdateAndMessage | Error + """ + + return ( + await asyncio_detailed( + device_id=device_id, + update_id=update_id, + client=client, + ) + ).parsed diff --git a/src/infuse_iot/api_client/api/device/get_device_application_updates_by_device_id.py b/src/infuse_iot/api_client/api/device/get_device_application_updates_by_device_id.py new file mode 100644 index 0000000..1c3d4c0 --- /dev/null +++ b/src/infuse_iot/api_client/api/device/get_device_application_updates_by_device_id.py @@ -0,0 +1,227 @@ +from http import HTTPStatus +from typing import Any +from urllib.parse import quote + +import httpx + +from ... import errors +from ...client import AuthenticatedClient, Client +from ...models.device_application_update import DeviceApplicationUpdate +from ...models.device_application_update_status import DeviceApplicationUpdateStatus +from ...models.error import Error +from ...types import UNSET, Response, Unset + + +def _get_kwargs( + device_id: str, + *, + status: DeviceApplicationUpdateStatus | Unset = UNSET, + limit: int | Unset = 100, + offset: int | Unset = 0, +) -> dict[str, Any]: + + params: dict[str, Any] = {} + + json_status: str | Unset = UNSET + if not isinstance(status, Unset): + json_status = status.value + + params["status"] = json_status + + params["limit"] = limit + + params["offset"] = offset + + params = {k: v for k, v in params.items() if v is not UNSET and v is not None} + + _kwargs: dict[str, Any] = { + "method": "get", + "url": "/device/deviceId/{device_id}/application/updates".format( + device_id=quote(str(device_id), safe=""), + ), + "params": params, + } + + return _kwargs + + +def _parse_response( + *, client: AuthenticatedClient | Client, response: httpx.Response +) -> Error | list[DeviceApplicationUpdate] | None: + if response.status_code == 200: + response_200 = [] + _response_200 = response.json() + for response_200_item_data in _response_200: + response_200_item = DeviceApplicationUpdate.from_dict(response_200_item_data) + + response_200.append(response_200_item) + + return response_200 + + if response.status_code == 404: + response_404 = Error.from_dict(response.json()) + + return response_404 + + if response.status_code == 500: + response_500 = Error.from_dict(response.json()) + + return response_500 + + if client.raise_on_unexpected_status: + raise errors.UnexpectedStatus(response.status_code, response.content) + else: + return None + + +def _build_response( + *, client: AuthenticatedClient | Client, response: httpx.Response +) -> Response[Error | list[DeviceApplicationUpdate]]: + return Response( + status_code=HTTPStatus(response.status_code), + content=response.content, + headers=response.headers, + parsed=_parse_response(client=client, response=response), + ) + + +def sync_detailed( + device_id: str, + *, + client: AuthenticatedClient | Client, + status: DeviceApplicationUpdateStatus | Unset = UNSET, + limit: int | Unset = 100, + offset: int | Unset = 0, +) -> Response[Error | list[DeviceApplicationUpdate]]: + """Get device application updates by DeviceID + + Args: + device_id (str): + status (DeviceApplicationUpdateStatus | Unset): Status of device application update + limit (int | Unset): Default: 100. + offset (int | Unset): Default: 0. + + Raises: + errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True. + httpx.TimeoutException: If the request takes longer than Client.timeout. + + Returns: + Response[Error | list[DeviceApplicationUpdate]] + """ + + kwargs = _get_kwargs( + device_id=device_id, + status=status, + limit=limit, + offset=offset, + ) + + response = client.get_httpx_client().request( + **kwargs, + ) + + return _build_response(client=client, response=response) + + +def sync( + device_id: str, + *, + client: AuthenticatedClient | Client, + status: DeviceApplicationUpdateStatus | Unset = UNSET, + limit: int | Unset = 100, + offset: int | Unset = 0, +) -> Error | list[DeviceApplicationUpdate] | None: + """Get device application updates by DeviceID + + Args: + device_id (str): + status (DeviceApplicationUpdateStatus | Unset): Status of device application update + limit (int | Unset): Default: 100. + offset (int | Unset): Default: 0. + + Raises: + errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True. + httpx.TimeoutException: If the request takes longer than Client.timeout. + + Returns: + Error | list[DeviceApplicationUpdate] + """ + + return sync_detailed( + device_id=device_id, + client=client, + status=status, + limit=limit, + offset=offset, + ).parsed + + +async def asyncio_detailed( + device_id: str, + *, + client: AuthenticatedClient | Client, + status: DeviceApplicationUpdateStatus | Unset = UNSET, + limit: int | Unset = 100, + offset: int | Unset = 0, +) -> Response[Error | list[DeviceApplicationUpdate]]: + """Get device application updates by DeviceID + + Args: + device_id (str): + status (DeviceApplicationUpdateStatus | Unset): Status of device application update + limit (int | Unset): Default: 100. + offset (int | Unset): Default: 0. + + Raises: + errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True. + httpx.TimeoutException: If the request takes longer than Client.timeout. + + Returns: + Response[Error | list[DeviceApplicationUpdate]] + """ + + kwargs = _get_kwargs( + device_id=device_id, + status=status, + limit=limit, + offset=offset, + ) + + response = await client.get_async_httpx_client().request(**kwargs) + + return _build_response(client=client, response=response) + + +async def asyncio( + device_id: str, + *, + client: AuthenticatedClient | Client, + status: DeviceApplicationUpdateStatus | Unset = UNSET, + limit: int | Unset = 100, + offset: int | Unset = 0, +) -> Error | list[DeviceApplicationUpdate] | None: + """Get device application updates by DeviceID + + Args: + device_id (str): + status (DeviceApplicationUpdateStatus | Unset): Status of device application update + limit (int | Unset): Default: 100. + offset (int | Unset): Default: 0. + + Raises: + errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True. + httpx.TimeoutException: If the request takes longer than Client.timeout. + + Returns: + Error | list[DeviceApplicationUpdate] + """ + + return ( + await asyncio_detailed( + device_id=device_id, + client=client, + status=status, + limit=limit, + offset=offset, + ) + ).parsed diff --git a/src/infuse_iot/api_client/models/__init__.py b/src/infuse_iot/api_client/models/__init__.py index 1287f8b..f9efa95 100644 --- a/src/infuse_iot/api_client/models/__init__.py +++ b/src/infuse_iot/api_client/models/__init__.py @@ -50,6 +50,10 @@ from .derive_device_key_body import DeriveDeviceKeyBody from .device import Device from .device_and_state import DeviceAndState +from .device_application_state import DeviceApplicationState +from .device_application_update import DeviceApplicationUpdate +from .device_application_update_and_message import DeviceApplicationUpdateAndMessage +from .device_application_update_status import DeviceApplicationUpdateStatus from .device_entry_update_status import DeviceEntryUpdateStatus from .device_id_field import DeviceIdField from .device_kv_entry import DeviceKVEntry @@ -84,6 +88,7 @@ from .new_application import NewApplication from .new_board import NewBoard from .new_device import NewDevice +from .new_device_application_update import NewDeviceApplicationUpdate from .new_device_kv_entry_update import NewDeviceKVEntryUpdate from .new_device_kv_entry_update_decoded import NewDeviceKVEntryUpdateDecoded from .new_device_state import NewDeviceState @@ -155,6 +160,10 @@ "DeriveDeviceKeyBody", "Device", "DeviceAndState", + "DeviceApplicationState", + "DeviceApplicationUpdate", + "DeviceApplicationUpdateAndMessage", + "DeviceApplicationUpdateStatus", "DeviceEntryUpdateStatus", "DeviceIdField", "DeviceKVEntry", @@ -189,6 +198,7 @@ "NewApplication", "NewBoard", "NewDevice", + "NewDeviceApplicationUpdate", "NewDeviceKVEntryUpdate", "NewDeviceKVEntryUpdateDecoded", "NewDeviceState", diff --git a/src/infuse_iot/api_client/models/device_application_state.py b/src/infuse_iot/api_client/models/device_application_state.py new file mode 100644 index 0000000..0b4d524 --- /dev/null +++ b/src/infuse_iot/api_client/models/device_application_state.py @@ -0,0 +1,106 @@ +from __future__ import annotations + +import datetime +from collections.abc import Mapping +from typing import TYPE_CHECKING, Any, TypeVar + +from attrs import define as _attrs_define +from attrs import field as _attrs_field +from dateutil.parser import isoparse + +from ..types import UNSET, Unset + +if TYPE_CHECKING: + from ..models.application_release_version import ApplicationReleaseVersion + + +T = TypeVar("T", bound="DeviceApplicationState") + + +@_attrs_define +class DeviceApplicationState: + """ + Attributes: + application_id (int): ID of application + version (ApplicationReleaseVersion): + last_reported_time (datetime.datetime): Time of report from device with this application state (if reported by + device) + board_target_crc (int | Unset): CRC16 of board target string for application release + release_id (str | Unset): ID of associated release for the application state (if known) + """ + + application_id: int + version: ApplicationReleaseVersion + last_reported_time: datetime.datetime + board_target_crc: int | Unset = UNSET + release_id: str | Unset = UNSET + additional_properties: dict[str, Any] = _attrs_field(init=False, factory=dict) + + def to_dict(self) -> dict[str, Any]: + application_id = self.application_id + + version = self.version.to_dict() + + last_reported_time = self.last_reported_time.isoformat() + + board_target_crc = self.board_target_crc + + release_id = self.release_id + + field_dict: dict[str, Any] = {} + field_dict.update(self.additional_properties) + field_dict.update( + { + "applicationId": application_id, + "version": version, + "lastReportedTime": last_reported_time, + } + ) + if board_target_crc is not UNSET: + field_dict["boardTargetCrc"] = board_target_crc + if release_id is not UNSET: + field_dict["releaseId"] = release_id + + return field_dict + + @classmethod + def from_dict(cls: type[T], src_dict: Mapping[str, Any]) -> T: + from ..models.application_release_version import ApplicationReleaseVersion + + d = dict(src_dict) + application_id = d.pop("applicationId") + + version = ApplicationReleaseVersion.from_dict(d.pop("version")) + + last_reported_time = isoparse(d.pop("lastReportedTime")) + + board_target_crc = d.pop("boardTargetCrc", UNSET) + + release_id = d.pop("releaseId", UNSET) + + device_application_state = cls( + application_id=application_id, + version=version, + last_reported_time=last_reported_time, + board_target_crc=board_target_crc, + release_id=release_id, + ) + + device_application_state.additional_properties = d + return device_application_state + + @property + def additional_keys(self) -> list[str]: + return list(self.additional_properties.keys()) + + def __getitem__(self, key: str) -> Any: + return self.additional_properties[key] + + def __setitem__(self, key: str, value: Any) -> None: + self.additional_properties[key] = value + + def __delitem__(self, key: str) -> None: + del self.additional_properties[key] + + def __contains__(self, key: str) -> bool: + return key in self.additional_properties diff --git a/src/infuse_iot/api_client/models/device_application_update.py b/src/infuse_iot/api_client/models/device_application_update.py new file mode 100644 index 0000000..33c3777 --- /dev/null +++ b/src/infuse_iot/api_client/models/device_application_update.py @@ -0,0 +1,164 @@ +from __future__ import annotations + +import datetime +from collections.abc import Mapping +from typing import Any, TypeVar +from uuid import UUID + +from attrs import define as _attrs_define +from attrs import field as _attrs_field +from dateutil.parser import isoparse + +from ..models.device_application_update_status import DeviceApplicationUpdateStatus +from ..types import UNSET, Unset + +T = TypeVar("T", bound="DeviceApplicationUpdate") + + +@_attrs_define +class DeviceApplicationUpdate: + """ + Attributes: + release_id (str): ID of application release to update device to + id (UUID): ID of update + status (DeviceApplicationUpdateStatus): Status of device application update + attempt_count (int): Number of attempts made to update the device + created_at (datetime.datetime): + updated_at (datetime.datetime): + last_error (str | Unset): Last error message if update failed + last_attempt_at (datetime.datetime | Unset): Time of last attempt + downlink_message_id (UUID | Unset): ID of latest downlink message for the update (if sent) + completed_at (datetime.datetime | Unset): Time the update was completed (if completed) + """ + + release_id: str + id: UUID + status: DeviceApplicationUpdateStatus + attempt_count: int + created_at: datetime.datetime + updated_at: datetime.datetime + last_error: str | Unset = UNSET + last_attempt_at: datetime.datetime | Unset = UNSET + downlink_message_id: UUID | Unset = UNSET + completed_at: datetime.datetime | Unset = UNSET + additional_properties: dict[str, Any] = _attrs_field(init=False, factory=dict) + + def to_dict(self) -> dict[str, Any]: + release_id = self.release_id + + id = str(self.id) + + status = self.status.value + + attempt_count = self.attempt_count + + created_at = self.created_at.isoformat() + + updated_at = self.updated_at.isoformat() + + last_error = self.last_error + + last_attempt_at: str | Unset = UNSET + if not isinstance(self.last_attempt_at, Unset): + last_attempt_at = self.last_attempt_at.isoformat() + + downlink_message_id: str | Unset = UNSET + if not isinstance(self.downlink_message_id, Unset): + downlink_message_id = str(self.downlink_message_id) + + completed_at: str | Unset = UNSET + if not isinstance(self.completed_at, Unset): + completed_at = self.completed_at.isoformat() + + field_dict: dict[str, Any] = {} + field_dict.update(self.additional_properties) + field_dict.update( + { + "releaseId": release_id, + "id": id, + "status": status, + "attemptCount": attempt_count, + "createdAt": created_at, + "updatedAt": updated_at, + } + ) + if last_error is not UNSET: + field_dict["lastError"] = last_error + if last_attempt_at is not UNSET: + field_dict["lastAttemptAt"] = last_attempt_at + if downlink_message_id is not UNSET: + field_dict["downlinkMessageId"] = downlink_message_id + if completed_at is not UNSET: + field_dict["completedAt"] = completed_at + + return field_dict + + @classmethod + def from_dict(cls: type[T], src_dict: Mapping[str, Any]) -> T: + d = dict(src_dict) + release_id = d.pop("releaseId") + + id = UUID(d.pop("id")) + + status = DeviceApplicationUpdateStatus(d.pop("status")) + + attempt_count = d.pop("attemptCount") + + created_at = isoparse(d.pop("createdAt")) + + updated_at = isoparse(d.pop("updatedAt")) + + last_error = d.pop("lastError", UNSET) + + _last_attempt_at = d.pop("lastAttemptAt", UNSET) + last_attempt_at: datetime.datetime | Unset + if isinstance(_last_attempt_at, Unset): + last_attempt_at = UNSET + else: + last_attempt_at = isoparse(_last_attempt_at) + + _downlink_message_id = d.pop("downlinkMessageId", UNSET) + downlink_message_id: UUID | Unset + if isinstance(_downlink_message_id, Unset): + downlink_message_id = UNSET + else: + downlink_message_id = UUID(_downlink_message_id) + + _completed_at = d.pop("completedAt", UNSET) + completed_at: datetime.datetime | Unset + if isinstance(_completed_at, Unset): + completed_at = UNSET + else: + completed_at = isoparse(_completed_at) + + device_application_update = cls( + release_id=release_id, + id=id, + status=status, + attempt_count=attempt_count, + created_at=created_at, + updated_at=updated_at, + last_error=last_error, + last_attempt_at=last_attempt_at, + downlink_message_id=downlink_message_id, + completed_at=completed_at, + ) + + device_application_update.additional_properties = d + return device_application_update + + @property + def additional_keys(self) -> list[str]: + return list(self.additional_properties.keys()) + + def __getitem__(self, key: str) -> Any: + return self.additional_properties[key] + + def __setitem__(self, key: str, value: Any) -> None: + self.additional_properties[key] = value + + def __delitem__(self, key: str) -> None: + del self.additional_properties[key] + + def __contains__(self, key: str) -> bool: + return key in self.additional_properties diff --git a/src/infuse_iot/api_client/models/device_application_update_and_message.py b/src/infuse_iot/api_client/models/device_application_update_and_message.py new file mode 100644 index 0000000..6362087 --- /dev/null +++ b/src/infuse_iot/api_client/models/device_application_update_and_message.py @@ -0,0 +1,186 @@ +from __future__ import annotations + +import datetime +from collections.abc import Mapping +from typing import TYPE_CHECKING, Any, TypeVar +from uuid import UUID + +from attrs import define as _attrs_define +from attrs import field as _attrs_field +from dateutil.parser import isoparse + +from ..models.device_application_update_status import DeviceApplicationUpdateStatus +from ..types import UNSET, Unset + +if TYPE_CHECKING: + from ..models.downlink_message import DownlinkMessage + + +T = TypeVar("T", bound="DeviceApplicationUpdateAndMessage") + + +@_attrs_define +class DeviceApplicationUpdateAndMessage: + """ + Attributes: + release_id (str): ID of application release to update device to + id (UUID): ID of update + status (DeviceApplicationUpdateStatus): Status of device application update + attempt_count (int): Number of attempts made to update the device + created_at (datetime.datetime): + updated_at (datetime.datetime): + last_error (str | Unset): Last error message if update failed + last_attempt_at (datetime.datetime | Unset): Time of last attempt + downlink_message_id (UUID | Unset): ID of latest downlink message for the update (if sent) + completed_at (datetime.datetime | Unset): Time the update was completed (if completed) + downlink_message (DownlinkMessage | Unset): + """ + + release_id: str + id: UUID + status: DeviceApplicationUpdateStatus + attempt_count: int + created_at: datetime.datetime + updated_at: datetime.datetime + last_error: str | Unset = UNSET + last_attempt_at: datetime.datetime | Unset = UNSET + downlink_message_id: UUID | Unset = UNSET + completed_at: datetime.datetime | Unset = UNSET + downlink_message: DownlinkMessage | Unset = UNSET + additional_properties: dict[str, Any] = _attrs_field(init=False, factory=dict) + + def to_dict(self) -> dict[str, Any]: + release_id = self.release_id + + id = str(self.id) + + status = self.status.value + + attempt_count = self.attempt_count + + created_at = self.created_at.isoformat() + + updated_at = self.updated_at.isoformat() + + last_error = self.last_error + + last_attempt_at: str | Unset = UNSET + if not isinstance(self.last_attempt_at, Unset): + last_attempt_at = self.last_attempt_at.isoformat() + + downlink_message_id: str | Unset = UNSET + if not isinstance(self.downlink_message_id, Unset): + downlink_message_id = str(self.downlink_message_id) + + completed_at: str | Unset = UNSET + if not isinstance(self.completed_at, Unset): + completed_at = self.completed_at.isoformat() + + downlink_message: dict[str, Any] | Unset = UNSET + if not isinstance(self.downlink_message, Unset): + downlink_message = self.downlink_message.to_dict() + + field_dict: dict[str, Any] = {} + field_dict.update(self.additional_properties) + field_dict.update( + { + "releaseId": release_id, + "id": id, + "status": status, + "attemptCount": attempt_count, + "createdAt": created_at, + "updatedAt": updated_at, + } + ) + if last_error is not UNSET: + field_dict["lastError"] = last_error + if last_attempt_at is not UNSET: + field_dict["lastAttemptAt"] = last_attempt_at + if downlink_message_id is not UNSET: + field_dict["downlinkMessageId"] = downlink_message_id + if completed_at is not UNSET: + field_dict["completedAt"] = completed_at + if downlink_message is not UNSET: + field_dict["downlinkMessage"] = downlink_message + + return field_dict + + @classmethod + def from_dict(cls: type[T], src_dict: Mapping[str, Any]) -> T: + from ..models.downlink_message import DownlinkMessage + + d = dict(src_dict) + release_id = d.pop("releaseId") + + id = UUID(d.pop("id")) + + status = DeviceApplicationUpdateStatus(d.pop("status")) + + attempt_count = d.pop("attemptCount") + + created_at = isoparse(d.pop("createdAt")) + + updated_at = isoparse(d.pop("updatedAt")) + + last_error = d.pop("lastError", UNSET) + + _last_attempt_at = d.pop("lastAttemptAt", UNSET) + last_attempt_at: datetime.datetime | Unset + if isinstance(_last_attempt_at, Unset): + last_attempt_at = UNSET + else: + last_attempt_at = isoparse(_last_attempt_at) + + _downlink_message_id = d.pop("downlinkMessageId", UNSET) + downlink_message_id: UUID | Unset + if isinstance(_downlink_message_id, Unset): + downlink_message_id = UNSET + else: + downlink_message_id = UUID(_downlink_message_id) + + _completed_at = d.pop("completedAt", UNSET) + completed_at: datetime.datetime | Unset + if isinstance(_completed_at, Unset): + completed_at = UNSET + else: + completed_at = isoparse(_completed_at) + + _downlink_message = d.pop("downlinkMessage", UNSET) + downlink_message: DownlinkMessage | Unset + if isinstance(_downlink_message, Unset): + downlink_message = UNSET + else: + downlink_message = DownlinkMessage.from_dict(_downlink_message) + + device_application_update_and_message = cls( + release_id=release_id, + id=id, + status=status, + attempt_count=attempt_count, + created_at=created_at, + updated_at=updated_at, + last_error=last_error, + last_attempt_at=last_attempt_at, + downlink_message_id=downlink_message_id, + completed_at=completed_at, + downlink_message=downlink_message, + ) + + device_application_update_and_message.additional_properties = d + return device_application_update_and_message + + @property + def additional_keys(self) -> list[str]: + return list(self.additional_properties.keys()) + + def __getitem__(self, key: str) -> Any: + return self.additional_properties[key] + + def __setitem__(self, key: str, value: Any) -> None: + self.additional_properties[key] = value + + def __delitem__(self, key: str) -> None: + del self.additional_properties[key] + + def __contains__(self, key: str) -> bool: + return key in self.additional_properties diff --git a/src/infuse_iot/api_client/models/device_application_update_status.py b/src/infuse_iot/api_client/models/device_application_update_status.py new file mode 100644 index 0000000..bd66277 --- /dev/null +++ b/src/infuse_iot/api_client/models/device_application_update_status.py @@ -0,0 +1,11 @@ +from enum import Enum + + +class DeviceApplicationUpdateStatus(str, Enum): + CANCELLED = "cancelled" + FAILED = "failed" + PENDING = "pending" + SUCCESS = "success" + + def __str__(self) -> str: + return str(self.value) diff --git a/src/infuse_iot/api_client/models/new_device_application_update.py b/src/infuse_iot/api_client/models/new_device_application_update.py new file mode 100644 index 0000000..840b7b5 --- /dev/null +++ b/src/infuse_iot/api_client/models/new_device_application_update.py @@ -0,0 +1,61 @@ +from __future__ import annotations + +from collections.abc import Mapping +from typing import Any, TypeVar + +from attrs import define as _attrs_define +from attrs import field as _attrs_field + +T = TypeVar("T", bound="NewDeviceApplicationUpdate") + + +@_attrs_define +class NewDeviceApplicationUpdate: + """ + Attributes: + release_id (str): ID of application release to update device to + """ + + release_id: str + additional_properties: dict[str, Any] = _attrs_field(init=False, factory=dict) + + def to_dict(self) -> dict[str, Any]: + release_id = self.release_id + + field_dict: dict[str, Any] = {} + field_dict.update(self.additional_properties) + field_dict.update( + { + "releaseId": release_id, + } + ) + + return field_dict + + @classmethod + def from_dict(cls: type[T], src_dict: Mapping[str, Any]) -> T: + d = dict(src_dict) + release_id = d.pop("releaseId") + + new_device_application_update = cls( + release_id=release_id, + ) + + new_device_application_update.additional_properties = d + return new_device_application_update + + @property + def additional_keys(self) -> list[str]: + return list(self.additional_properties.keys()) + + def __getitem__(self, key: str) -> Any: + return self.additional_properties[key] + + def __setitem__(self, key: str, value: Any) -> None: + self.additional_properties[key] = value + + def __delitem__(self, key: str) -> None: + del self.additional_properties[key] + + def __contains__(self, key: str) -> bool: + return key in self.additional_properties From a9f5278bf94b3d56df3bff858bc5a022bd218618 Mon Sep 17 00:00:00 2001 From: Jordan Yates Date: Tue, 16 Jun 2026 14:40:42 +1000 Subject: [PATCH 2/3] tools: cloud: release ID in `device info` Display the current release ID (if known) as part of `infuse cloud device info`. Signed-off-by: Jordan Yates --- src/infuse_iot/tools/cloud.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/src/infuse_iot/tools/cloud.py b/src/infuse_iot/tools/cloud.py index 012ff5a..0ad5487 100644 --- a/src/infuse_iot/tools/cloud.py +++ b/src/infuse_iot/tools/cloud.py @@ -31,6 +31,7 @@ ) from infuse_iot.api_client.api.coap import get_coap_files from infuse_iot.api_client.api.device import ( + get_device_application_state_by_device_id, get_device_by_device_id, get_device_kv_entries_by_device_id, get_device_last_route_by_device_id, @@ -210,6 +211,7 @@ def info(self, client: Client): board = get_board_by_id.sync(client=client, id=info.board_id) state = get_device_state_by_id.sync(client=client, id=info.id) route = get_device_last_route_by_device_id.sync(client=client, device_id=id_str) + app = get_device_application_state_by_device_id.sync(client=client, device_id=id_str) logger_states = get_device_logger_states_by_device_id.sync(client=client, device_id=id_str) table: list[tuple[str, Any]] = [ @@ -235,6 +237,8 @@ def info(self, client: Client): table += [("Application ID", f"0x{state.application_id:08x}")] if v: table += [("Version", f"{v.major}.{v.minor}.{v.revision}+{v.build_num:08x}")] + if isinstance(app, models.DeviceApplicationState): + table += [("Release ID", app.release_id if app.release_id else "N/A")] if isinstance(route, models.UplinkRoute): table += [ ("~~~Latest Route~~~", ""), From 6e0344ec255673e414ab2d563b91805738213d94 Mon Sep 17 00:00:00 2001 From: Jordan Yates Date: Thu, 18 Jun 2026 15:20:17 +1000 Subject: [PATCH 3/3] tools: cloud: device: dfu: added Add subcommands the schedule and query device firmware updates. Signed-off-by: Jordan Yates --- src/infuse_iot/tools/cloud.py | 53 +++++++++++++++++++++++++++++++++++ 1 file changed, 53 insertions(+) diff --git a/src/infuse_iot/tools/cloud.py b/src/infuse_iot/tools/cloud.py index 0ad5487..addfb72 100644 --- a/src/infuse_iot/tools/cloud.py +++ b/src/infuse_iot/tools/cloud.py @@ -31,7 +31,9 @@ ) from infuse_iot.api_client.api.coap import get_coap_files from infuse_iot.api_client.api.device import ( + create_device_application_update_by_device_id, get_device_application_state_by_device_id, + get_device_application_updates_by_device_id, get_device_by_device_id, get_device_kv_entries_by_device_id, get_device_last_route_by_device_id, @@ -186,11 +188,19 @@ def add_parser(cls, parser): info_parser = tool_parser.add_parser("info", help="General device information") info_parser.set_defaults(command_fn=cls.info) info_parser.add_argument("--id", type=str, required=True, help="Infuse-IoT device ID") + kv_parser = tool_parser.add_parser("kv_state", help="Key-Value device state") kv_parser.set_defaults(command_fn=cls.kv_state) kv_parser.add_argument("--id", type=str, required=True, help="Infuse-IoT device ID") kv_parser.add_argument("--schedules", action="store_true", help="Display task schedules") + dfu_parser = tool_parser.add_parser("dfu", help="Manage device firmware upgrades") + dfu_parser.set_defaults(command_fn=cls.dfu) + dfu_parser.add_argument("--id", type=str, required=True, help="Infuse-IoT device ID") + dfu_action = dfu_parser.add_mutually_exclusive_group(required=True) + dfu_action.add_argument("--schedule", type=str, help="Release ID to upgrade to") + dfu_action.add_argument("--status", action="store_true", help="Check DFU status") + def run(self): with self.client() as client: self.args.command_fn(self, client) @@ -312,6 +322,49 @@ def kv_state(self, client: Client): print(tabulate(table)) + def dfu(self, client: Client): + id_int = int(self.args.id, 0) + id_str = f"{id_int:016x}" + + if self.args.schedule: + body = models.NewDeviceApplicationUpdate(self.args.schedule) + rsp = create_device_application_update_by_device_id.sync(client=client, device_id=id_str, body=body) + + if rsp is None: + sys.exit("Create application updates: No response") + elif isinstance(rsp, models.Error): + sys.exit(f"<{rsp.code}>: {rsp.message}") + elif isinstance(rsp, models.DeviceApplicationState): + print(f"Device already on release {self.args.schedule}") + elif isinstance(rsp, models.DeviceApplicationUpdate): + print(f"DFU scheduled with ID {rsp.id}") + else: + raise NotImplementedError(f"Unknown response ({rsp})") + elif self.args.status: + updates = get_device_application_updates_by_device_id.sync( + client=client, + device_id=id_str, + ) + if updates is None: + sys.exit("Get application updates: No response") + elif isinstance(updates, models.Error): + sys.exit(f"<{updates.code}>: {updates.message}") + + for update in updates: + print( + tabulate( + [ + ["To Release", update.release_id], + ["Status", str(update.status)], + ["Attempts", str(update.attempt_count)], + ["Last Attempt", str(update.last_attempt_at)], + ["Completed", str(update.completed_at)], + ] + ) + ) + else: + raise NotImplementedError("Unknown DFU subcommand") + class Coap(CloudSubCommand): @classmethod