chore(tools): add a simple tool for triggering OTA update locally (#325)

This PR brings back the tools to send OTA update request and trigger OTA update directly to an ECU locally without the need of connecting to the Internet.

tools/.gitignore

@@ -0,0 +1 @@
+update_request.yaml

tools/local_ota/README.md

@@ -0,0 +1,23 @@
+# local OTA
+
+Tools for triggering OTA locally.
+
+Currently supports API version 2.
+
+## Usage
+
+1. define the `update_request.yaml` as follow:
+
+```yaml
+# adjust the settings as your needs
+- ecu_id: "autoware"
+  version: "789.x"
+  url: "https://10.0.1.1:8443/"
+  cookies: '{"test": "my-cookie"}'
+```
+
+1. make the OTA update request API call as follow:
+
+```shell
+ python3 tools/local_ota/api_v2 -i 10.0.1.10 update_request.yaml
+```

tools/local_ota/__init__.py

@@ -0,0 +1,13 @@
+# Copyright 2022 TIER IV, INC. All rights reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.

tools/local_ota/api_v2.py

@@ -0,0 +1,134 @@
+# Copyright 2022 TIER IV, INC. All rights reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""A simple tool to trigger an OTA locally, for API version 2."""
+
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import logging
+from pathlib import Path
+from typing import Iterator, List
+
+import yaml
+from pydantic import BaseModel, RootModel
+from typing_extensions import Self
+
+from otaclient_api.v2 import types
+from otaclient_api.v2.api_caller import ECUNoResponse, OTAClientCall
+from otaclient_common.typing import StrOrPath
+
+logger = logging.getLogger(__name__)
+
+DEFAULT_PORT = 50051
+
+
+class UpdateRequest(BaseModel):
+    ecu_id: str
+    version: str
+    url: str
+    cookies: str
+
+
+class UpdateRequestList(RootModel):
+    root: List[UpdateRequest]
+
+    def __iter__(self) -> Iterator[UpdateRequest]:
+        return iter(self.root)
+
+    @classmethod
+    def load_update_request_yaml(cls, fpath: StrOrPath) -> Self:
+        _raw = Path(fpath).read_text()
+        _loaded = yaml.safe_load(_raw)
+        assert isinstance(_loaded, list)
+
+        return cls.model_validate(_loaded)
+
+
+def generate_request(req_entries: UpdateRequestList) -> types.UpdateRequest:
+    return types.UpdateRequest(
+        ecu=(
+            types.UpdateRequestEcu(
+                ecu_id=entry.ecu_id,
+                version=entry.version,
+                url=entry.url,
+                cookies=entry.cookies,
+            )
+            for entry in req_entries
+        )
+    )
+
+
+async def main(
+    host: str,
+    port: int,
+    *,
+    req: types.UpdateRequest,
+    timeout: int = 3,
+) -> None:
+    try:
+        resp = await OTAClientCall.update_call(
+            "not_used",
+            host,
+            port,
+            request=req,
+            timeout=timeout,
+        )
+        logger.info(f"update response: {resp}")
+    except ECUNoResponse as e:
+        _err_msg = f"ECU doesn't response to the request on-time({timeout=}): {e}"
+        logger.error(_err_msg)
+
+
+if __name__ == "__main__":
+    logging.basicConfig(level=logging.INFO)
+
+    parser = argparse.ArgumentParser(
+        description="Calling ECU's update API, API version v2",
+        formatter_class=argparse.ArgumentDefaultsHelpFormatter,
+    )
+
+    parser.add_argument(
+        "-i",
+        "--host",
+        help="ECU listen IP address",
+        required=True,
+    )
+    parser.add_argument(
+        "-p",
+        "--port",
+        help="ECU listen port",
+        type=int,
+        default=DEFAULT_PORT,
+    )
+    parser.add_argument(
+        "request_yaml",
+        default="update_request.yaml",
+        help="request to send, described in a yaml file",
+    )
+
+    args = parser.parse_args()
+
+    req = generate_request(
+        UpdateRequestList.load_update_request_yaml(args.request_yaml)
+    )
+    logger.info(f"loaded update request: {req}")
+    asyncio.run(
+        main(
+            args.host,
+            args.port,
+            req=req,
+        )
+    )