From f5b293d0927506c2a979a091bf0d07ecc78fa181 Mon Sep 17 00:00:00 2001 From: Dmitrii Agibov Date: Thu, 8 Sep 2022 14:24:39 +0100 Subject: MLIA-386 Simplify typing in the source code - Enable deferred annotations evaluation - Use builtin types for type hints whenever possible - Use | syntax for union types - Rename mlia.core._typing into mlia.core.typing Change-Id: I3f6ffc02fa069c589bdd9e8bddbccd504285427a --- src/mlia/backend/application.py | 18 +++++------ src/mlia/backend/common.py | 64 +++++++++++++++++-------------------- src/mlia/backend/config.py | 19 +++++------ src/mlia/backend/execution.py | 41 ++++++++++++------------ src/mlia/backend/fs.py | 7 ++-- src/mlia/backend/manager.py | 41 +++++++++++------------- src/mlia/backend/output_consumer.py | 5 +-- src/mlia/backend/proc.py | 17 +++++----- src/mlia/backend/source.py | 22 ++++++------- src/mlia/backend/system.py | 12 +++---- 10 files changed, 119 insertions(+), 127 deletions(-) (limited to 'src/mlia/backend') diff --git a/src/mlia/backend/application.py b/src/mlia/backend/application.py index 4b04324..a093afe 100644 --- a/src/mlia/backend/application.py +++ b/src/mlia/backend/application.py @@ -1,13 +1,13 @@ # SPDX-FileCopyrightText: Copyright 2022, Arm Limited and/or its affiliates. # SPDX-License-Identifier: Apache-2.0 """Application backend module.""" +from __future__ import annotations + import re from pathlib import Path from typing import Any from typing import cast -from typing import Dict from typing import List -from typing import Optional from mlia.backend.common import Backend from mlia.backend.common import ConfigurationException @@ -23,12 +23,12 @@ from mlia.backend.source import create_destination_and_install from mlia.backend.source import get_source -def get_available_application_directory_names() -> List[str]: +def get_available_application_directory_names() -> list[str]: """Return a list of directory names for all available applications.""" return [entry.name for entry in get_backend_directories("applications")] -def get_available_applications() -> List["Application"]: +def get_available_applications() -> list[Application]: """Return a list with all available applications.""" available_applications = [] for config_json in get_backend_configs("applications"): @@ -42,8 +42,8 @@ def get_available_applications() -> List["Application"]: def get_application( - application_name: str, system_name: Optional[str] = None -) -> List["Application"]: + application_name: str, system_name: str | None = None +) -> list[Application]: """Return a list of application instances with provided name.""" return [ application @@ -85,7 +85,7 @@ def remove_application(directory_name: str) -> None: remove_backend(directory_name, "applications") -def get_unique_application_names(system_name: Optional[str] = None) -> List[str]: +def get_unique_application_names(system_name: str | None = None) -> list[str]: """Extract a list of unique application names of all application available.""" return list( set( @@ -120,7 +120,7 @@ class Application(Backend): """Check if the application can run on the system passed as argument.""" return system_name in self.supported_systems - def get_details(self) -> Dict[str, Any]: + def get_details(self) -> dict[str, Any]: """Return dictionary with information about the Application instance.""" output = { "type": "application", @@ -156,7 +156,7 @@ class Application(Backend): command.params = used_params -def load_applications(config: ExtendedApplicationConfig) -> List[Application]: +def load_applications(config: ExtendedApplicationConfig) -> list[Application]: """Load application. Application configuration could contain different parameters/commands for different diff --git a/src/mlia/backend/common.py b/src/mlia/backend/common.py index e61d6b6..697c2a0 100644 --- a/src/mlia/backend/common.py +++ b/src/mlia/backend/common.py @@ -1,6 +1,8 @@ # SPDX-FileCopyrightText: Copyright 2022, Arm Limited and/or its affiliates. # SPDX-License-Identifier: Apache-2.0 """Contain all common functions for the backends.""" +from __future__ import annotations + import json import logging import re @@ -10,18 +12,12 @@ from pathlib import Path from typing import Any from typing import Callable from typing import cast -from typing import Dict from typing import Final from typing import IO from typing import Iterable -from typing import List from typing import Match from typing import NamedTuple -from typing import Optional from typing import Pattern -from typing import Tuple -from typing import Type -from typing import Union from mlia.backend.config import BackendConfig from mlia.backend.config import BaseBackendConfig @@ -74,7 +70,7 @@ def remove_backend(directory_name: str, resource_type: ResourceType) -> None: remove_resource(directory_name, resource_type) -def load_config(config: Union[None, Path, IO[bytes]]) -> BackendConfig: +def load_config(config: Path | IO[bytes] | None) -> BackendConfig: """Return a loaded json file.""" if config is None: raise Exception("Unable to read config") @@ -86,7 +82,7 @@ def load_config(config: Union[None, Path, IO[bytes]]) -> BackendConfig: return cast(BackendConfig, json.load(config)) -def parse_raw_parameter(parameter: str) -> Tuple[str, Optional[str]]: +def parse_raw_parameter(parameter: str) -> tuple[str, str | None]: """Split the parameter string in name and optional value. It manages the following cases: @@ -176,7 +172,7 @@ class Backend(ABC): def _parse_commands_and_params(self, config: BaseBackendConfig) -> None: """Parse commands and user parameters.""" - self.commands: Dict[str, Command] = {} + self.commands: dict[str, Command] = {} commands = config.get("commands") if commands: @@ -213,15 +209,15 @@ class Backend(ABC): @classmethod def _parse_params( - cls, params: Optional[UserParamsConfig], command: str - ) -> List["Param"]: + cls, params: UserParamsConfig | None, command: str + ) -> list[Param]: if not params: return [] return [cls._parse_param(p) for p in params.get(command, [])] @classmethod - def _parse_param(cls, param: UserParamConfig) -> "Param": + def _parse_param(cls, param: UserParamConfig) -> Param: """Parse a single parameter.""" name = param.get("name") if name is not None and not name: @@ -239,16 +235,14 @@ class Backend(ABC): alias=alias, ) - def _get_command_details(self) -> Dict: + def _get_command_details(self) -> dict: command_details = { command_name: command.get_details() for command_name, command in self.commands.items() } return command_details - def _get_user_param_value( - self, user_params: List[str], param: "Param" - ) -> Optional[str]: + def _get_user_param_value(self, user_params: list[str], param: Param) -> str | None: """Get the user-specified value of a parameter.""" for user_param in user_params: user_param_name, user_param_value = parse_raw_parameter(user_param) @@ -267,7 +261,7 @@ class Backend(ABC): return None @staticmethod - def _same_parameter(user_param_name_or_alias: str, param: "Param") -> bool: + def _same_parameter(user_param_name_or_alias: str, param: Param) -> bool: """Compare user parameter name with param name or alias.""" # Strip the "=" sign in the param_name. This is needed just for # comparison with the parameters passed by the user. @@ -277,10 +271,10 @@ class Backend(ABC): return user_param_name_or_alias in [param_name, param.alias] def resolved_parameters( - self, command_name: str, user_params: List[str] - ) -> List[Tuple[Optional[str], "Param"]]: + self, command_name: str, user_params: list[str] + ) -> list[tuple[str | None, Param]]: """Return list of parameters with values.""" - result: List[Tuple[Optional[str], "Param"]] = [] + result: list[tuple[str | None, Param]] = [] command = self.commands.get(command_name) if not command: return result @@ -296,9 +290,9 @@ class Backend(ABC): def build_command( self, command_name: str, - user_params: List[str], - param_resolver: Callable[[str, str, List[Tuple[Optional[str], "Param"]]], str], - ) -> List[str]: + user_params: list[str], + param_resolver: Callable[[str, str, list[tuple[str | None, Param]]], str], + ) -> list[str]: """ Return a list of executable command strings. @@ -328,11 +322,11 @@ class Param: def __init__( # pylint: disable=too-many-arguments self, - name: Optional[str], + name: str | None, description: str, - values: Optional[List[str]] = None, - default_value: Optional[str] = None, - alias: Optional[str] = None, + values: list[str] | None = None, + default_value: str | None = None, + alias: str | None = None, ) -> None: """Construct a Param instance.""" if not name and not alias: @@ -345,7 +339,7 @@ class Param: self.default_value = default_value self.alias = alias - def get_details(self) -> Dict: + def get_details(self) -> dict: """Return a dictionary with all relevant information of a Param.""" return {key: value for key, value in self.__dict__.items() if value} @@ -366,7 +360,7 @@ class Command: """Class for representing a command.""" def __init__( - self, command_strings: List[str], params: Optional[List[Param]] = None + self, command_strings: list[str], params: list[Param] | None = None ) -> None: """Construct a Command instance.""" self.command_strings = command_strings @@ -404,7 +398,7 @@ class Command: "as parameter name." ) - def get_details(self) -> Dict: + def get_details(self) -> dict: """Return a dictionary with all relevant information of a Command.""" output = { "command_strings": self.command_strings, @@ -425,9 +419,9 @@ class Command: def resolve_all_parameters( str_val: str, - param_resolver: Callable[[str, str, List[Tuple[Optional[str], Param]]], str], - command_name: Optional[str] = None, - params_values: Optional[List[Tuple[Optional[str], Param]]] = None, + param_resolver: Callable[[str, str, list[tuple[str | None, Param]]], str], + command_name: str | None = None, + params_values: list[tuple[str | None, Param]] | None = None, ) -> str: """Resolve all parameters in the string.""" if not str_val: @@ -446,7 +440,7 @@ def resolve_all_parameters( def load_application_configs( config: Any, - config_type: Type[Any], + config_type: type[Any], is_system_required: bool = True, ) -> Any: """Get one config for each system supported by the application. @@ -456,7 +450,7 @@ def load_application_configs( config with appropriate configuration. """ merged_configs = [] - supported_systems: Optional[List[NamedExecutionConfig]] = config.get( + supported_systems: list[NamedExecutionConfig] | None = config.get( "supported_systems" ) if not supported_systems: diff --git a/src/mlia/backend/config.py b/src/mlia/backend/config.py index 9a56fa9..dca53da 100644 --- a/src/mlia/backend/config.py +++ b/src/mlia/backend/config.py @@ -1,10 +1,11 @@ # SPDX-FileCopyrightText: Copyright 2022, Arm Limited and/or its affiliates. # SPDX-License-Identifier: Apache-2.0 """Contain definition of backend configuration.""" +from __future__ import annotations + from pathlib import Path from typing import Dict from typing import List -from typing import Optional from typing import TypedDict from typing import Union @@ -12,9 +13,9 @@ from typing import Union class UserParamConfig(TypedDict, total=False): """User parameter configuration.""" - name: Optional[str] + name: str | None default_value: str - values: List[str] + values: list[str] description: str alias: str @@ -25,9 +26,9 @@ UserParamsConfig = Dict[str, List[UserParamConfig]] class ExecutionConfig(TypedDict, total=False): """Execution configuration.""" - commands: Dict[str, List[str]] + commands: dict[str, list[str]] user_params: UserParamsConfig - variables: Dict[str, str] + variables: dict[str, str] class NamedExecutionConfig(ExecutionConfig): @@ -42,25 +43,25 @@ class BaseBackendConfig(ExecutionConfig, total=False): name: str description: str config_location: Path - annotations: Dict[str, Union[str, List[str]]] + annotations: dict[str, str | list[str]] class ApplicationConfig(BaseBackendConfig, total=False): """Application configuration.""" - supported_systems: List[str] + supported_systems: list[str] class ExtendedApplicationConfig(BaseBackendConfig, total=False): """Extended application configuration.""" - supported_systems: List[NamedExecutionConfig] + supported_systems: list[NamedExecutionConfig] class SystemConfig(BaseBackendConfig, total=False): """System configuration.""" - reporting: Dict[str, Dict] + reporting: dict[str, dict] BackendItemConfig = Union[ApplicationConfig, SystemConfig] diff --git a/src/mlia/backend/execution.py b/src/mlia/backend/execution.py index 5340a47..f3fe401 100644 --- a/src/mlia/backend/execution.py +++ b/src/mlia/backend/execution.py @@ -1,12 +1,11 @@ # SPDX-FileCopyrightText: Copyright 2022, Arm Limited and/or its affiliates. # SPDX-License-Identifier: Apache-2.0 """Application execution module.""" +from __future__ import annotations + import logging import re from typing import cast -from typing import List -from typing import Optional -from typing import Tuple from mlia.backend.application import Application from mlia.backend.application import get_application @@ -29,9 +28,9 @@ class ExecutionContext: # pylint: disable=too-few-public-methods def __init__( self, app: Application, - app_params: List[str], + app_params: list[str], system: System, - system_params: List[str], + system_params: list[str], ): """Init execution context.""" self.app = app @@ -41,8 +40,8 @@ class ExecutionContext: # pylint: disable=too-few-public-methods self.param_resolver = ParamResolver(self) - self.stdout: Optional[bytearray] = None - self.stderr: Optional[bytearray] = None + self.stdout: bytearray | None = None + self.stderr: bytearray | None = None class ParamResolver: @@ -54,16 +53,16 @@ class ParamResolver: @staticmethod def resolve_user_params( - cmd_name: Optional[str], + cmd_name: str | None, index_or_alias: str, - resolved_params: Optional[List[Tuple[Optional[str], Param]]], + resolved_params: list[tuple[str | None, Param]] | None, ) -> str: """Resolve user params.""" if not cmd_name or resolved_params is None: raise ConfigurationException("Unable to resolve user params") - param_value: Optional[str] = None - param: Optional[Param] = None + param_value: str | None = None + param: Param | None = None if index_or_alias.isnumeric(): i = int(index_or_alias) @@ -176,8 +175,8 @@ class ParamResolver: def param_matcher( self, param_name: str, - cmd_name: Optional[str], - resolved_params: Optional[List[Tuple[Optional[str], Param]]], + cmd_name: str | None, + resolved_params: list[tuple[str | None, Param]] | None, ) -> str: """Regexp to resolve a param from the param_name.""" # this pattern supports parameter names like "application.commands.run:0" and @@ -224,8 +223,8 @@ class ParamResolver: def param_resolver( self, param_name: str, - cmd_name: Optional[str] = None, - resolved_params: Optional[List[Tuple[Optional[str], Param]]] = None, + cmd_name: str | None = None, + resolved_params: list[tuple[str | None, Param]] | None = None, ) -> str: """Resolve parameter value based on current execution context.""" # Note: 'software.*' is included for backward compatibility. @@ -253,15 +252,15 @@ class ParamResolver: def __call__( self, param_name: str, - cmd_name: Optional[str] = None, - resolved_params: Optional[List[Tuple[Optional[str], Param]]] = None, + cmd_name: str | None = None, + resolved_params: list[tuple[str | None, Param]] | None = None, ) -> str: """Resolve provided parameter.""" return self.param_resolver(param_name, cmd_name, resolved_params) def validate_parameters( - backend: Backend, command_names: List[str], params: List[str] + backend: Backend, command_names: list[str], params: list[str] ) -> None: """Check parameters passed to backend.""" for param in params: @@ -301,7 +300,7 @@ def get_application_by_name_and_system( def get_application_and_system( application_name: str, system_name: str -) -> Tuple[Application, System]: +) -> tuple[Application, System]: """Return application and system by provided names.""" system = get_system(system_name) if not system: @@ -314,9 +313,9 @@ def get_application_and_system( def run_application( application_name: str, - application_params: List[str], + application_params: list[str], system_name: str, - system_params: List[str], + system_params: list[str], ) -> ExecutionContext: """Run application on the provided system.""" application, system = get_application_and_system(application_name, system_name) diff --git a/src/mlia/backend/fs.py b/src/mlia/backend/fs.py index 9fb53b1..3fce19c 100644 --- a/src/mlia/backend/fs.py +++ b/src/mlia/backend/fs.py @@ -1,11 +1,12 @@ # SPDX-FileCopyrightText: Copyright 2022, Arm Limited and/or its affiliates. # SPDX-License-Identifier: Apache-2.0 """Module to host all file system related functions.""" +from __future__ import annotations + import re import shutil from pathlib import Path from typing import Literal -from typing import Optional from mlia.utils.filesystem import get_mlia_resources @@ -58,7 +59,7 @@ def remove_resource(resource_directory: str, resource_type: ResourceType) -> Non shutil.rmtree(resource_location) -def remove_directory(directory_path: Optional[Path]) -> None: +def remove_directory(directory_path: Path | None) -> None: """Remove directory.""" if not directory_path or not directory_path.is_dir(): raise Exception("No directory path provided") @@ -66,7 +67,7 @@ def remove_directory(directory_path: Optional[Path]) -> None: shutil.rmtree(directory_path) -def recreate_directory(directory_path: Optional[Path]) -> None: +def recreate_directory(directory_path: Path | None) -> None: """Recreate directory.""" if not directory_path: raise Exception("No directory path provided") diff --git a/src/mlia/backend/manager.py b/src/mlia/backend/manager.py index 8d8246d..c8fe0f7 100644 --- a/src/mlia/backend/manager.py +++ b/src/mlia/backend/manager.py @@ -1,17 +1,14 @@ # SPDX-FileCopyrightText: Copyright 2022, Arm Limited and/or its affiliates. # SPDX-License-Identifier: Apache-2.0 """Module for backend integration.""" +from __future__ import annotations + import logging from abc import ABC from abc import abstractmethod from dataclasses import dataclass from pathlib import Path -from typing import Dict -from typing import List from typing import Literal -from typing import Optional -from typing import Set -from typing import Tuple from mlia.backend.application import get_available_applications from mlia.backend.application import install_application @@ -58,7 +55,7 @@ def get_system_name(backend: str, device_type: str) -> str: return _SUPPORTED_SYSTEMS[backend][device_type] -def is_supported(backend: str, device_type: Optional[str] = None) -> bool: +def is_supported(backend: str, device_type: str | None = None) -> bool: """Check if the backend (and optionally device type) is supported.""" if device_type is None: return backend in _SUPPORTED_SYSTEMS @@ -70,17 +67,17 @@ def is_supported(backend: str, device_type: Optional[str] = None) -> bool: return False -def supported_backends() -> List[str]: +def supported_backends() -> list[str]: """Get a list of all backends supported by the backend manager.""" return list(_SUPPORTED_SYSTEMS.keys()) -def get_all_system_names(backend: str) -> List[str]: +def get_all_system_names(backend: str) -> list[str]: """Get all systems supported by the backend.""" return list(_SUPPORTED_SYSTEMS.get(backend, {}).values()) -def get_all_application_names(backend: str) -> List[str]: +def get_all_application_names(backend: str) -> list[str]: """Get all applications supported by the backend.""" app_set = { app @@ -124,8 +121,8 @@ class ExecutionParams: application: str system: str - application_params: List[str] - system_params: List[str] + application_params: list[str] + system_params: list[str] class LogWriter(OutputConsumer): @@ -153,7 +150,7 @@ class GenericInferenceOutputParser(Base64OutputConsumer): } @property - def result(self) -> Dict: + def result(self) -> dict: """Merge the raw results and map the names to the right output names.""" merged_result = {} for raw_result in self.parsed_output: @@ -172,7 +169,7 @@ class GenericInferenceOutputParser(Base64OutputConsumer): """Return true if all expected data has been parsed.""" return set(self.result.keys()) == set(self._map.values()) - def missed_keys(self) -> Set[str]: + def missed_keys(self) -> set[str]: """Return a set of the keys that have not been found in the output.""" return set(self._map.values()) - set(self.result.keys()) @@ -184,12 +181,12 @@ class BackendRunner: """Init BackendRunner instance.""" @staticmethod - def get_installed_systems() -> List[str]: + def get_installed_systems() -> list[str]: """Get list of the installed systems.""" return [system.name for system in get_available_systems()] @staticmethod - def get_installed_applications(system: Optional[str] = None) -> List[str]: + def get_installed_applications(system: str | None = None) -> list[str]: """Get list of the installed application.""" return [ app.name @@ -205,7 +202,7 @@ class BackendRunner: """Return true if requested system installed.""" return system in self.get_installed_systems() - def systems_installed(self, systems: List[str]) -> bool: + def systems_installed(self, systems: list[str]) -> bool: """Check if all provided systems are installed.""" if not systems: return False @@ -213,7 +210,7 @@ class BackendRunner: installed_systems = self.get_installed_systems() return all(system in installed_systems for system in systems) - def applications_installed(self, applications: List[str]) -> bool: + def applications_installed(self, applications: list[str]) -> bool: """Check if all provided applications are installed.""" if not applications: return False @@ -221,7 +218,7 @@ class BackendRunner: installed_apps = self.get_installed_applications() return all(app in installed_apps for app in applications) - def all_installed(self, systems: List[str], apps: List[str]) -> bool: + def all_installed(self, systems: list[str], apps: list[str]) -> bool: """Check if all provided artifacts are installed.""" return self.systems_installed(systems) and self.applications_installed(apps) @@ -247,7 +244,7 @@ class BackendRunner: return ctx @staticmethod - def _params(name: str, params: List[str]) -> List[str]: + def _params(name: str, params: list[str]) -> list[str]: return [p for item in [(name, param) for param in params] for p in item] @@ -259,7 +256,7 @@ class GenericInferenceRunner(ABC): self.backend_runner = backend_runner def run( - self, model_info: ModelInfo, output_consumers: List[OutputConsumer] + self, model_info: ModelInfo, output_consumers: list[OutputConsumer] ) -> None: """Run generic inference for the provided device/model.""" execution_params = self.get_execution_params(model_info) @@ -284,7 +281,7 @@ class GenericInferenceRunner(ABC): ) @staticmethod - def consume_output(output: bytearray, consumers: List[OutputConsumer]) -> bytearray: + def consume_output(output: bytearray, consumers: list[OutputConsumer]) -> bytearray: """ Pass program's output to the consumers and filter it. @@ -320,7 +317,7 @@ class GenericInferenceRunnerEthosU(GenericInferenceRunner): @staticmethod def resolve_system_and_app( device_info: DeviceInfo, backend: str - ) -> Tuple[str, str]: + ) -> tuple[str, str]: """Find appropriate system and application for the provided device/backend.""" try: system_name = get_system_name(backend, device_info.device_type) diff --git a/src/mlia/backend/output_consumer.py b/src/mlia/backend/output_consumer.py index bac4186..3c3b132 100644 --- a/src/mlia/backend/output_consumer.py +++ b/src/mlia/backend/output_consumer.py @@ -1,10 +1,11 @@ # SPDX-FileCopyrightText: Copyright 2022, Arm Limited and/or its affiliates. # SPDX-License-Identifier: Apache-2.0 """Output consumers module.""" +from __future__ import annotations + import base64 import json import re -from typing import List from typing import Protocol from typing import runtime_checkable @@ -37,7 +38,7 @@ class Base64OutputConsumer(OutputConsumer): def __init__(self) -> None: """Set up the regular expression to extract tagged strings.""" self._regex = re.compile(rf"<{self.TAG_NAME}>(.*)") - self.parsed_output: List = [] + self.parsed_output: list = [] def feed(self, line: str) -> bool: """ diff --git a/src/mlia/backend/proc.py b/src/mlia/backend/proc.py index 911d672..7b3e92a 100644 --- a/src/mlia/backend/proc.py +++ b/src/mlia/backend/proc.py @@ -5,6 +5,8 @@ This module contains all classes and functions for dealing with Linux processes. """ +from __future__ import annotations + import datetime import logging import shlex @@ -13,9 +15,6 @@ import tempfile import time from pathlib import Path from typing import Any -from typing import List -from typing import Optional -from typing import Tuple from sh import Command from sh import CommandNotFound @@ -38,12 +37,12 @@ class ShellCommand: self, cmd: str, *args: str, - _cwd: Optional[Path] = None, + _cwd: Path | None = None, _tee: bool = True, _bg: bool = True, _out: Any = None, _err: Any = None, - _search_paths: Optional[List[Path]] = None, + _search_paths: list[Path] | None = None, ) -> RunningCommand: """Run the shell command with the given arguments. @@ -72,7 +71,7 @@ class ShellCommand: return command(_out=out, _err=err, _tee=_tee, _bg=_bg, _bg_exc=False) @classmethod - def get_stdout_stderr_paths(cls, cmd: str) -> Tuple[Path, Path]: + def get_stdout_stderr_paths(cls, cmd: str) -> tuple[Path, Path]: """Construct and returns the paths of stdout/stderr files.""" timestamp = datetime.datetime.now().timestamp() base_path = Path(tempfile.mkdtemp(prefix="mlia-", suffix=f"{timestamp}")) @@ -88,7 +87,7 @@ class ShellCommand: return stdout, stderr -def parse_command(command: str, shell: str = "bash") -> List[str]: +def parse_command(command: str, shell: str = "bash") -> list[str]: """Parse command.""" cmd, *args = shlex.split(command, posix=True) @@ -130,13 +129,13 @@ def run_and_wait( terminate_on_error: bool = False, out: Any = None, err: Any = None, -) -> Tuple[int, bytearray, bytearray]: +) -> tuple[int, bytearray, bytearray]: """ Run command and wait while it is executing. Returns a tuple: (exit_code, stdout, stderr) """ - running_cmd: Optional[RunningCommand] = None + running_cmd: RunningCommand | None = None try: running_cmd = execute_command(command, cwd, bg=True, out=out, err=err) return running_cmd.exit_code, running_cmd.stdout, running_cmd.stderr diff --git a/src/mlia/backend/source.py b/src/mlia/backend/source.py index f80a774..c951eae 100644 --- a/src/mlia/backend/source.py +++ b/src/mlia/backend/source.py @@ -1,6 +1,8 @@ # SPDX-FileCopyrightText: Copyright 2022, Arm Limited and/or its affiliates. # SPDX-License-Identifier: Apache-2.0 """Contain source related classes and functions.""" +from __future__ import annotations + import os import shutil import tarfile @@ -8,8 +10,6 @@ from abc import ABC from abc import abstractmethod from pathlib import Path from tarfile import TarFile -from typing import Optional -from typing import Union from mlia.backend.common import BACKEND_CONFIG_FILE from mlia.backend.common import ConfigurationException @@ -24,11 +24,11 @@ class Source(ABC): """Source class.""" @abstractmethod - def name(self) -> Optional[str]: + def name(self) -> str | None: """Get source name.""" @abstractmethod - def config(self) -> Optional[BackendConfig]: + def config(self) -> BackendConfig | None: """Get configuration file content.""" @abstractmethod @@ -52,7 +52,7 @@ class DirectorySource(Source): """Return name of source.""" return self.directory_path.name - def config(self) -> Optional[BackendConfig]: + def config(self) -> BackendConfig | None: """Return configuration file content.""" if not is_backend_directory(self.directory_path): raise ConfigurationException("No configuration file found") @@ -84,9 +84,9 @@ class TarArchiveSource(Source): """Create the TarArchiveSource class.""" assert isinstance(archive_path, Path) self.archive_path = archive_path - self._config: Optional[BackendConfig] = None - self._has_top_level_folder: Optional[bool] = None - self._name: Optional[str] = None + self._config: BackendConfig | None = None + self._has_top_level_folder: bool | None = None + self._name: str | None = None def _read_archive_content(self) -> None: """Read various information about archive.""" @@ -125,14 +125,14 @@ class TarArchiveSource(Source): content = archive.extractfile(config_entry) self._config = load_config(content) - def config(self) -> Optional[BackendConfig]: + def config(self) -> BackendConfig | None: """Return configuration file content.""" if self._config is None: self._read_archive_content() return self._config - def name(self) -> Optional[str]: + def name(self) -> str | None: """Return name of the source.""" if self._name is None: self._read_archive_content() @@ -171,7 +171,7 @@ class TarArchiveSource(Source): ) -def get_source(source_path: Path) -> Union[TarArchiveSource, DirectorySource]: +def get_source(source_path: Path) -> TarArchiveSource | DirectorySource: """Return appropriate source instance based on provided source path.""" if source_path.is_file(): return TarArchiveSource(source_path) diff --git a/src/mlia/backend/system.py b/src/mlia/backend/system.py index ff85bf3..0e51ab2 100644 --- a/src/mlia/backend/system.py +++ b/src/mlia/backend/system.py @@ -1,12 +1,12 @@ # SPDX-FileCopyrightText: Copyright 2022, Arm Limited and/or its affiliates. # SPDX-License-Identifier: Apache-2.0 """System backend module.""" +from __future__ import annotations + from pathlib import Path from typing import Any from typing import cast -from typing import Dict from typing import List -from typing import Tuple from mlia.backend.common import Backend from mlia.backend.common import ConfigurationException @@ -33,7 +33,7 @@ class System(Backend): def _setup_reporting(self, config: SystemConfig) -> None: self.reporting = config.get("reporting") - def run(self, command: str) -> Tuple[int, bytearray, bytearray]: + def run(self, command: str) -> tuple[int, bytearray, bytearray]: """ Run command on the system. @@ -63,7 +63,7 @@ class System(Backend): return super().__eq__(other) and self.name == other.name - def get_details(self) -> Dict[str, Any]: + def get_details(self) -> dict[str, Any]: """Return a dictionary with all relevant information of a System.""" output = { "type": "system", @@ -76,12 +76,12 @@ class System(Backend): return output -def get_available_systems_directory_names() -> List[str]: +def get_available_systems_directory_names() -> list[str]: """Return a list of directory names for all avialable systems.""" return [entry.name for entry in get_backend_directories("systems")] -def get_available_systems() -> List[System]: +def get_available_systems() -> list[System]: """Return a list with all available systems.""" available_systems = [] for config_json in get_backend_configs("systems"): -- cgit v1.2.1