refine(v3.7.x): refactor proxy_info.yaml and ecu_info.yaml config files parsing

otaclient/__init__.py

@@ -12,7 +12,10 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-
-from otaclient._version import version, __version__
+try:
+    from otaclient._version import version, __version__
+except ImportError:
+    # unknown version
+    version = __version__ = "0.0.0"
 
 __all__ = ["version", "__version__"]

otaclient/_utils/typing.py

@@ -0,0 +1,48 @@
+# 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.
+
+
+from __future__ import annotations
+from pathlib import Path
+from typing import Any, Callable, TypeVar, Union
+from typing_extensions import Annotated, ParamSpec
+
+from pydantic import Field
+
+P = ParamSpec("P")
+T = TypeVar("T")
+RT = TypeVar("RT")
+
+StrOrPath = Union[str, Path]
+
+# pydantic helpers
+
+NetworkPort = Annotated[int, Field(ge=1, le=65535)]
+
+
+def gen_strenum_validator(enum_type: type[T]) -> Callable[[T | str | Any], T]:
+    """A before validator generator that converts input value into enum
+    before passing it to pydantic validator.
+
+    NOTE(20240129): as upto pydantic v2.5.3, (str, Enum) field cannot
+                    pass strict validation if input is str.
+    """
+
+    def _inner(value: T | str | Any) -> T:
+        assert isinstance(
+            value, (enum_type, str)
+        ), f"{value=} should be {enum_type} or str type"
+        return enum_type(value)
+
+    return _inner

otaclient/app/boot_control/_cboot.py

@@ -177,9 +177,7 @@ def _init_dev_info(self):
         self.current_slot: str = Nvbootctrl.get_current_slot()
         self.current_rootfs_dev: str = CMDHelperFuncs.get_current_rootfs_dev()
         # NOTE: boot dev is always emmc device now
-        self.current_boot_dev: str = (
-            f"/dev/{Nvbootctrl.EMMC_DEV}p{Nvbootctrl.SLOTID_PARTID_MAP[self.current_slot]}"
-        )
+        self.current_boot_dev: str = f"/dev/{Nvbootctrl.EMMC_DEV}p{Nvbootctrl.SLOTID_PARTID_MAP[self.current_slot]}"
 
         self.standby_slot: str = Nvbootctrl.CURRENT_STANDBY_FLIP[self.current_slot]
         standby_partid = Nvbootctrl.SLOTID_PARTID_MAP[self.standby_slot]

otaclient/app/boot_control/configs.py

@@ -13,39 +13,13 @@
 # limitations under the License.
 
 
+from __future__ import annotations
 from dataclasses import dataclass, field
-from enum import Enum
 from typing import Dict
 
 from ..configs import BaseConfig
 
-
-class BootloaderType(Enum):
-    """Bootloaders that supported by otaclient.
-
-    grub: generic x86_64 platform with grub
-    cboot: ADLink rqx-580, rqx-58g, with BSP 32.5.x
-        (theoretically other Nvidia jetson xavier devices using cboot are also supported)
-    rpi_boot: raspberry pi 4 with eeprom version newer than 2020-10-28
-    """
-
-    UNSPECIFIED = "unspecified"
-    GRUB = "grub"
-    CBOOT = "cboot"
-    RPI_BOOT = "rpi_boot"
-
-    @classmethod
-    def parse_str(cls, _input: str) -> "BootloaderType":
-        res = cls.UNSPECIFIED
-        try:  # input is enum key(capitalized)
-            res = BootloaderType[_input]
-        except KeyError:
-            pass
-        try:  # input is enum value(uncapitalized)
-            res = BootloaderType(_input)
-        except ValueError:
-            pass
-        return res
+from otaclient.configs.ecu_info import BootloaderType
 
 
 @dataclass

otaclient/app/boot_control/selecter.py

@@ -13,12 +13,13 @@
 # limitations under the License.
 
 
+from __future__ import annotations
 import logging
 import platform
 from pathlib import Path
-from typing import Type
+from typing_extensions import deprecated
 
-from .configs import BootloaderType, cboot_cfg, rpi_boot_cfg
+from .configs import BootloaderType
 from ._errors import BootControlError
 from .protocol import BootControllerProtocol
 
@@ -27,7 +28,8 @@
 logger = logging.getLogger(__name__)
 
 
-def detect_bootloader(raise_on_unknown=True) -> BootloaderType:
+@deprecated("bootloader type SHOULD be set in ecu_info.yaml instead of detecting")
+def detect_bootloader() -> BootloaderType:
     """Detect which bootloader we are running with by some evidence.
 
     Evidence:
@@ -42,37 +44,38 @@ def detect_bootloader(raise_on_unknown=True) -> BootloaderType:
     machine, arch = platform.machine(), platform.processor()
     if machine == "x86_64" or arch == "x86_64":
         return BootloaderType.GRUB
+
     # distinguish between rpi and jetson xavier device
     if machine == "aarch64" or arch == "aarch64":
+        from .configs import cboot_cfg, rpi_boot_cfg
+
         # evidence: jetson xvaier device has a special file which reveals the
         #           tegra chip id
         if Path(cboot_cfg.TEGRA_CHIP_ID_PATH).is_file():
             return BootloaderType.CBOOT
+
         # evidence: rpi device has a special file which reveals the rpi model
         rpi_model_file = Path(rpi_boot_cfg.RPI_MODEL_FILE)
         if rpi_model_file.is_file():
             if (_model_str := read_str_from_file(rpi_model_file)).find(
                 rpi_boot_cfg.RPI_MODEL_HINT
             ) != -1:
                 return BootloaderType.RPI_BOOT
-            else:
-                logger.error(
-                    f"detect unsupported raspberry pi platform({_model_str=}), only {rpi_boot_cfg.RPI_MODEL_HINT} is supported"
-                )
 
-    # failed to detect bootloader
-    if raise_on_unknown:
-        raise ValueError(f"unsupported platform({machine=}, {arch=}) detected, abort")
-    return BootloaderType.UNSPECIFIED
+            logger.error(
+                f"detect unsupported raspberry pi platform({_model_str=}), "
+                f"only {rpi_boot_cfg.RPI_MODEL_HINT} is supported"
+            )
+    raise ValueError(f"unsupported platform({machine=}, {arch=}) detected, abort")
 
 
 def get_boot_controller(
     bootloader_type: BootloaderType,
-) -> Type[BootControllerProtocol]:
+) -> type[BootControllerProtocol]:
     # if ecu_info doesn't specify the bootloader type,
-    # we try to detect by ourselves
-    if bootloader_type is BootloaderType.UNSPECIFIED:
-        bootloader_type = detect_bootloader(raise_on_unknown=True)
+    # we try to detect by ourselves.
+    if bootloader_type is BootloaderType.AUTO_DETECT:
+        bootloader_type = detect_bootloader()
     logger.info(f"use boot_controller for {bootloader_type=}")
 
     if bootloader_type == BootloaderType.GRUB:

otaclient/app/configs.py

@@ -17,6 +17,11 @@
 from logging import INFO
 from typing import Dict, Tuple
 
+from otaclient.configs.ecu_info import ecu_info
+from otaclient.configs.proxy_info import proxy_info
+
+ecu_info, proxy_info = ecu_info, proxy_info  # to prevent static check warnings
+
 
 class CreateStandbyMechanism(Enum):
     LEGACY = 0  # deprecated and removed

otaclient/app/log_setting.py

@@ -17,26 +17,13 @@
 
 import atexit
 import logging
-import os
-import yaml
+import requests
 from queue import Queue
 from threading import Event, Thread
 from urllib.parse import urljoin
 
-import requests
-
 import otaclient
-from .configs import config as cfg
-
-
-# NOTE: EcuInfo imports this log_setting so independent get_ecu_id are required.
-def get_ecu_id():
-    try:
-        with open(cfg.ECU_INFO_FILE) as f:
-            ecu_info = yaml.load(f, Loader=yaml.SafeLoader)
-            return ecu_info["ecu_id"]
-    except Exception:
-        return "autoware"
+from .configs import config as cfg, ecu_info, proxy_info
 
 
 class _LogTeeHandler(logging.Handler):
@@ -99,25 +86,15 @@ def configure_logging():
         _logger = logging.getLogger(_module_name)
         _logger.setLevel(_log_level)
 
-    # NOTE(20240306): for only god knows reason, although in proxy_info.yaml,
-    #   the logging_server field is assigned with an URL, and otaclient
-    #   expects an URL, the run.sh from autoware_ecu_system_setup pass in
-    #   HTTP_LOGGING_SERVER with URL schema being removed!?
-    # NOTE: I will do a quick fix here as I don't want to touch autoware_ecu_system_setup
-    #   for now, leave it in the future.
-    if iot_logger_url := os.environ.get("HTTP_LOGGING_SERVER"):
-        # special treatment for not-a-URL passed in by run.sh
-        # note that we only support http, the proxy_info.yaml should be properly setup.
-        if not (iot_logger_url.startswith("http") or iot_logger_url.startswith("HTTP")):
-            iot_logger_url = f"http://{iot_logger_url.strip('/')}"
+    if iot_logger_url := str(proxy_info.logging_server):
         iot_logger_url = f"{iot_logger_url.strip('/')}/"
 
         ch = _LogTeeHandler()
         fmt = logging.Formatter(fmt=cfg.LOG_FORMAT)
         ch.setFormatter(fmt)
 
         # star the logging thread
-        log_upload_endpoint = urljoin(iot_logger_url, get_ecu_id())
+        log_upload_endpoint = urljoin(iot_logger_url, ecu_info.ecu_id)
         ch.start_upload_thread(log_upload_endpoint)
 
         # NOTE: "otaclient" logger will be the root logger for all loggers name