aboutsummaryrefslogtreecommitdiff
path: root/src/mlia/backend
diff options
context:
space:
mode:
authorDmitrii Agibov <dmitrii.agibov@arm.com>2022-09-08 14:24:39 +0100
committerDmitrii Agibov <dmitrii.agibov@arm.com>2022-09-09 17:21:48 +0100
commitf5b293d0927506c2a979a091bf0d07ecc78fa181 (patch)
tree4de585b7cb6ed34da8237063752270189a730a41 /src/mlia/backend
parentcde0c6ee140bd108849bff40467d8f18ffc332ef (diff)
downloadmlia-f5b293d0927506c2a979a091bf0d07ecc78fa181.tar.gz
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
Diffstat (limited to 'src/mlia/backend')
-rw-r--r--src/mlia/backend/application.py18
-rw-r--r--src/mlia/backend/common.py64
-rw-r--r--src/mlia/backend/config.py19
-rw-r--r--src/mlia/backend/execution.py41
-rw-r--r--src/mlia/backend/fs.py7
-rw-r--r--src/mlia/backend/manager.py41
-rw-r--r--src/mlia/backend/output_consumer.py5
-rw-r--r--src/mlia/backend/proc.py17
-rw-r--r--src/mlia/backend/source.py22
-rw-r--r--src/mlia/backend/system.py12
10 files changed, 119 insertions, 127 deletions
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.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"):