add type annotations and docstrings to devlib

.gitignore

@@ -7,3 +7,7 @@ devlib/bin/scripts/shutils
 doc/_build/
 build/
 dist/
+.venv/
+.vscode/
+venv/
+.history/

devlib/_target_runner.py

@@ -1,4 +1,4 @@
-#    Copyright 2024 ARM Limited
+#    Copyright 2024-2025 ARM Limited
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -17,15 +17,24 @@
 Target runner and related classes are implemented here.
 """
 
-import logging
 import os
 import time
+
 from platform import machine
+from typing import Optional, cast, Protocol, TYPE_CHECKING, Union
+from typing_extensions import NotRequired, LiteralString, TypedDict
+from devlib.platform import Platform
+if TYPE_CHECKING:
+    from _typeshed import StrPath, BytesPath
+else:
+    StrPath = str
+    BytesPath = bytes
 
 from devlib.exception import (TargetStableError, HostError)
-from devlib.target import LinuxTarget
-from devlib.utils.misc import get_subprocess, which
+from devlib.target import LinuxTarget, Target
+from devlib.utils.misc import get_subprocess, which, get_logger
 from devlib.utils.ssh import SshConnection
+from devlib.utils.annotation_helpers import SubprocessCommand, SshUserConnectionSettings
 
 
 class TargetRunner:
@@ -36,16 +45,14 @@ class TargetRunner:
     (e.g., :class:`QEMUTargetRunner`).
 
     :param target: Specifies type of target per :class:`Target` based classes.
-    :type target: Target
     """
 
     def __init__(self,
-                 target):
+                 target: Target) -> None:
         self.target = target
+        self.logger = get_logger(self.__class__.__name__)
 
-        self.logger = logging.getLogger(self.__class__.__name__)
-
-    def __enter__(self):
+    def __enter__(self) -> 'TargetRunner':
         return self
 
     def __exit__(self, *_):
@@ -58,29 +65,25 @@ class SubprocessTargetRunner(TargetRunner):
 
     :param runner_cmd: The command to start runner process (e.g.,
         ``qemu-system-aarch64 -kernel Image -append "console=ttyAMA0" ...``).
-    :type runner_cmd: list(str)
 
     :param target: Specifies type of target per :class:`Target` based classes.
-    :type target: Target
 
     :param connect: Specifies if :class:`TargetRunner` should try to connect
         target after launching it, defaults to True.
-    :type connect: bool or None
 
     :param boot_timeout: Timeout for target's being ready for SSH access in
         seconds, defaults to 60.
-    :type boot_timeout: int or None
 
     :raises HostError: if it cannot execute runner command successfully.
 
     :raises TargetStableError: if Target is inaccessible.
     """
 
     def __init__(self,
-                 runner_cmd,
-                 target,
-                 connect=True,
-                 boot_timeout=60):
+                 runner_cmd: SubprocessCommand,
+                 target: Target,
+                 connect: bool = True,
+                 boot_timeout: int = 60):
         super().__init__(target=target)
 
         self.boot_timeout = boot_timeout
@@ -90,7 +93,7 @@ def __init__(self,
         try:
             self.runner_process = get_subprocess(runner_cmd)
         except Exception as ex:
-            raise HostError(f'Error while running "{runner_cmd}": {ex}') from ex
+            raise HostError(f'Error while running "{runner_cmd!r}": {ex}') from ex
 
         if connect:
             self.wait_boot_complete()
@@ -107,16 +110,16 @@ def __exit__(self, *_):
 
         self.terminate()
 
-    def wait_boot_complete(self):
+    def wait_boot_complete(self) -> None:
         """
-        Wait for target OS to finish boot up and become accessible over SSH in at most
-        ``SubprocessTargetRunner.boot_timeout`` seconds.
+        Wait for the target OS to finish booting and become accessible within
+        :attr:`boot_timeout` seconds.
 
-        :raises TargetStableError: In case of timeout.
+        :raises TargetStableError: If the target is inaccessible after the timeout.
         """
 
         start_time = time.time()
-        elapsed = 0
+        elapsed: float = 0.0
         while self.boot_timeout >= elapsed:
             try:
                 self.target.connect(timeout=self.boot_timeout - elapsed)
@@ -132,9 +135,9 @@ def wait_boot_complete(self):
         self.terminate()
         raise TargetStableError(f'Target is inaccessible for {self.boot_timeout} seconds!')
 
-    def terminate(self):
+    def terminate(self) -> None:
         """
-        Terminate ``SubprocessTargetRunner.runner_process``.
+        Terminate the subprocess associated with this runner.
         """
 
         self.logger.debug('Killing target runner...')
@@ -147,10 +150,9 @@ class NOPTargetRunner(TargetRunner):
     Class for implementing a target runner which does nothing except providing .target attribute.
 
     :param target: Specifies type of target per :class:`Target` based classes.
-    :type target: Target
     """
 
-    def __init__(self, target):
+    def __init__(self, target: Target) -> None:
         super().__init__(target=target)
 
     def __enter__(self):
@@ -159,11 +161,63 @@ def __enter__(self):
     def __exit__(self, *_):
         pass
 
-    def terminate(self):
+    def terminate(self) -> None:
         """
         Nothing to terminate for NOP target runners.
         Defined to be compliant with other runners (e.g., ``SubprocessTargetRunner``).
         """
+        pass
+
+
+class QEMUTargetUserSettings(TypedDict, total=False):
+    kernel_image: str
+    arch: NotRequired[str]
+    cpu_type: NotRequired[str]
+    initrd_image: str
+    mem_size: NotRequired[int]
+    num_cores: NotRequired[int]
+    num_threads: NotRequired[int]
+    cmdline: NotRequired[str]
+    enable_kvm: NotRequired[bool]
+
+
+class QEMUTargetRunnerSettings(TypedDict):
+    kernel_image: str
+    arch: str
+    cpu_type: str
+    initrd_image: str
+    mem_size: int
+    num_cores: int
+    num_threads: int
+    cmdline: str
+    enable_kvm: bool
+
+
+# TODO - look into which params can be made NotRequired and Optional
+# TODO - use pydantic for dynamic type checking
+class SshConnectionSettings(TypedDict):
+    username: str
+    password: str
+    keyfile: Optional[Union[LiteralString, StrPath, BytesPath]]
+    host: str
+    port: int
+    timeout: float
+    platform: 'Platform'
+    sudo_cmd: str
+    strict_host_check: bool
+    use_scp: bool
+    poll_transfers: bool
+    start_transfer_poll_delay: int
+    total_transfer_timeout: int
+    transfer_poll_period: int
+
+
+class QEMUTargetRunnerTargetFactory(Protocol):
+    """
+    Protocol for Lambda function for creating :class:`Target` based object.
+    """
+    def __call__(self, *, connect: bool, conn_cls, connection_settings: SshConnectionSettings) -> Target:
+        ...
 
 
 class QEMUTargetRunner(SubprocessTargetRunner):
@@ -177,7 +231,7 @@ class QEMUTargetRunner(SubprocessTargetRunner):
 
         * ``arch``: Architecture type. Defaults to ``aarch64``.
 
-        * ``cpu_types``: List of CPU ids for QEMU. The list only contains ``cortex-a72`` by
+        * ``cpu_type``: List of CPU ids for QEMU. The list only contains ``cortex-a72`` by
             default. This parameter is valid for Arm architectures only.
 
         * ``initrd_image``: This points to the location of initrd image (e.g.,
@@ -197,36 +251,37 @@ class QEMUTargetRunner(SubprocessTargetRunner):
         * ``enable_kvm``: Specifies if KVM will be used as accelerator in QEMU or not.
             Enabled by default if host architecture matches with target's for improving
             QEMU performance.
-    :type qemu_settings: Dict
 
     :param connection_settings: the dictionary to store connection settings
         of ``Target.connection_settings``, defaults to None.
-    :type connection_settings: Dict or None
 
     :param make_target: Lambda function for creating :class:`Target` based object.
-    :type make_target: func or None
 
     :Variable positional arguments: Forwarded to :class:`TargetRunner`.
 
     :raises FileNotFoundError: if QEMU executable, kernel or initrd image cannot be found.
     """
 
     def __init__(self,
-                 qemu_settings,
-                 connection_settings=None,
-                 make_target=LinuxTarget,
-                 **args):
+                 qemu_settings: QEMUTargetUserSettings,
+                 connection_settings: Optional[SshUserConnectionSettings] = None,
+                 make_target: QEMUTargetRunnerTargetFactory = cast(QEMUTargetRunnerTargetFactory, LinuxTarget),
+                 **kwargs) -> None:
 
-        self.connection_settings = {
+        default_connection_settings = {
             'host': '127.0.0.1',
             'port': 8022,
             'username': 'root',
             'password': 'root',
             'strict_host_check': False,
         }
-        self.connection_settings = {**self.connection_settings, **(connection_settings or {})}
+        # TODO - use pydantic for dynamic type checking. that can avoid casting and ensure runtime type compatibility
+        self.connection_settings: SshConnectionSettings = cast(SshConnectionSettings, {
+            **default_connection_settings,
+            **(connection_settings or {})
+        })
 
-        qemu_args = {
+        qemu_default_args = {
             'arch': 'aarch64',
             'cpu_type': 'cortex-a72',
             'mem_size': 512,
@@ -235,7 +290,8 @@ def __init__(self,
             'cmdline': 'console=ttyAMA0',
             'enable_kvm': True,
         }
-        qemu_args = {**qemu_args, **qemu_settings}
+        # TODO - same as above, use pydantic.
+        qemu_args: QEMUTargetRunnerSettings = cast(QEMUTargetRunnerSettings, {**qemu_default_args, **qemu_settings})
 
         qemu_executable = f'qemu-system-{qemu_args["arch"]}'
         qemu_path = which(qemu_executable)
@@ -281,4 +337,4 @@ def __init__(self,
 
         super().__init__(runner_cmd=qemu_cmd,
                          target=target,
-                         **args)
+                         **kwargs)

devlib/collector/__init__.py

@@ -1,4 +1,4 @@
-#    Copyright 2015 ARM Limited
+#    Copyright 2015-2025 ARM Limited
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -16,27 +16,63 @@
 import logging
 
 from devlib.utils.types import caseless_string
+from devlib.utils.misc import get_logger
+from typing import TYPE_CHECKING, Optional, List
+if TYPE_CHECKING:
+    from devlib.target import Target
+
 
 class CollectorBase(object):
+    """
+    The Collector API provide a consistent way of collecting arbitrary data from
+    a target. Data is collected via an instance of a class derived from :class:`CollectorBase`.
 
-    def __init__(self, target):
+    :param target: The devlib Target from which data will be collected.
+    """
+    def __init__(self, target: 'Target'):
         self.target = target
-        self.logger = logging.getLogger(self.__class__.__name__)
-        self.output_path = None
-
-    def reset(self):
+        self.logger: logging.Logger = get_logger(self.__class__.__name__)
+        self.output_path: Optional[str] = None
+
+    def reset(self) -> None:
+        """
+        This can be used to configure a collector for collection. This must be invoked
+        before :meth:`start()` is called to begin collection.
+        """
         pass
 
-    def start(self):
+    def start(self) -> None:
+        """
+        Starts collecting from the target.
+        """
         pass
 
     def stop(self):
+        """
+        Stops collecting from target. Must be called after
+        :func:`start()`.
+        """
         pass
 
-    def set_output(self, output_path):
+    def set_output(self, output_path: str) -> None:
+        """
+        Configure the output path for the particular collector. This will be either
+        a directory or file path which will be used when storing the data. Please see
+        the individual Collector documentation for more information.
+
+        :param output_path: The path (file or directory) to which data will be saved.
+        """
         self.output_path = output_path
 
-    def get_data(self):
+    def get_data(self) -> 'CollectorOutput':
+        """
+        The collected data will be return via the previously specified output_path.
+        This method will return a :class:`CollectorOutput` object which is a subclassed
+        list object containing individual ``CollectorOutputEntry`` objects with details
+        about the individual output entry.
+
+        :raises RuntimeError: If ``output_path`` has not been set.
+        """
         return CollectorOutput()
 
     def __enter__(self):
@@ -47,18 +83,26 @@ def __enter__(self):
     def __exit__(self, exc_type, exc_value, traceback):
         self.stop()
 
+
 class CollectorOutputEntry(object):
+    """
+    This object is designed to allow for the output of a collector to be processed
+    generically. The object will behave as a regular string containing the path to
+    underlying output path and can be used directly in ``os.path`` operations.
 
-    path_kinds = ['file', 'directory']
+    :param path: The file path of the collected output data.
+    :param path_kind: The type of output. Must be one of ``file`` or ``directory``.
+    """
+    path_kinds: List[str] = ['file', 'directory']
 
-    def __init__(self, path, path_kind):
-        self.path = path
+    def __init__(self, path: str, path_kind: str):
+        self.path = path  # path for the corresponding output item
 
         path_kind = caseless_string(path_kind)
         if path_kind not in self.path_kinds:
             msg = '{} is not a valid path_kind [{}]'
             raise ValueError(msg.format(path_kind, ' '.join(self.path_kinds)))
-        self.path_kind = path_kind
+        self.path_kind = path_kind  # file or directory
 
     def __str__(self):
         return self.path