mirror of
https://github.com/tiqi-group/pydase.git
synced 2025-04-21 08:40:03 +02:00
removes Optional typing and unused comments
This commit is contained in:
parent
ad2ae704e9
commit
93f0627534
@ -2,7 +2,7 @@ import base64
|
|||||||
import io
|
import io
|
||||||
import logging
|
import logging
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from typing import TYPE_CHECKING, Optional
|
from typing import TYPE_CHECKING
|
||||||
from urllib.request import urlopen
|
from urllib.request import urlopen
|
||||||
|
|
||||||
import PIL.Image # type: ignore[import-untyped]
|
import PIL.Image # type: ignore[import-untyped]
|
||||||
@ -45,7 +45,7 @@ class Image(DataService):
|
|||||||
image = PIL.Image.open(urlopen(url))
|
image = PIL.Image.open(urlopen(url))
|
||||||
self._load_from_pil(image)
|
self._load_from_pil(image)
|
||||||
|
|
||||||
def load_from_base64(self, value_: bytes, format_: Optional[str] = None) -> None:
|
def load_from_base64(self, value_: bytes, format_: str | None = None) -> None:
|
||||||
if format_ is None:
|
if format_ is None:
|
||||||
format_ = self._get_image_format_from_bytes(value_)
|
format_ = self._get_image_format_from_bytes(value_)
|
||||||
if format_ is None:
|
if format_ is None:
|
||||||
|
@ -419,7 +419,7 @@ class CallbackManager:
|
|||||||
for callback in self._notification_callbacks:
|
for callback in self._notification_callbacks:
|
||||||
try:
|
try:
|
||||||
callback(parent_path, name, value)
|
callback(parent_path, name, value)
|
||||||
except Exception as e: # noqa: PERF203
|
except Exception as e:
|
||||||
logger.error(e)
|
logger.error(e)
|
||||||
|
|
||||||
def add_notification_callback(
|
def add_notification_callback(
|
||||||
|
@ -1,7 +1,7 @@
|
|||||||
import logging
|
import logging
|
||||||
import warnings
|
import warnings
|
||||||
from enum import Enum
|
from enum import Enum
|
||||||
from typing import TYPE_CHECKING, Any, Optional, get_type_hints
|
from typing import TYPE_CHECKING, Any, get_type_hints
|
||||||
|
|
||||||
import rpyc # type: ignore[import-untyped]
|
import rpyc # type: ignore[import-untyped]
|
||||||
|
|
||||||
@ -106,7 +106,7 @@ class DataService(rpyc.Service, AbstractDataService):
|
|||||||
attr_name: str,
|
attr_name: str,
|
||||||
attr: Any,
|
attr: Any,
|
||||||
value: Any,
|
value: Any,
|
||||||
index: Optional[int],
|
index: int | None,
|
||||||
path_list: list[str],
|
path_list: list[str],
|
||||||
) -> None:
|
) -> None:
|
||||||
if isinstance(attr, Enum):
|
if isinstance(attr, Enum):
|
||||||
|
@ -3,7 +3,7 @@ import logging
|
|||||||
import os
|
import os
|
||||||
from collections.abc import Callable
|
from collections.abc import Callable
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from typing import TYPE_CHECKING, Any, Optional, cast
|
from typing import TYPE_CHECKING, Any, cast
|
||||||
|
|
||||||
import pydase.units as u
|
import pydase.units as u
|
||||||
from pydase.data_service.data_service_cache import DataServiceCache
|
from pydase.data_service.data_service_cache import DataServiceCache
|
||||||
@ -97,7 +97,7 @@ class StateManager:
|
|||||||
"""
|
"""
|
||||||
|
|
||||||
def __init__(
|
def __init__(
|
||||||
self, service: "DataService", filename: Optional[str | Path] = None
|
self, service: "DataService", filename: str | Path | None = None
|
||||||
) -> None:
|
) -> None:
|
||||||
self.filename = getattr(service, "_filename", None)
|
self.filename = getattr(service, "_filename", None)
|
||||||
|
|
||||||
|
@ -7,7 +7,7 @@ from concurrent.futures import ThreadPoolExecutor
|
|||||||
from copy import deepcopy
|
from copy import deepcopy
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from types import FrameType
|
from types import FrameType
|
||||||
from typing import Any, Optional, Protocol, TypedDict
|
from typing import Any, Protocol, TypedDict
|
||||||
|
|
||||||
import uvicorn
|
import uvicorn
|
||||||
from rpyc import ForkingServer, ThreadedServer # type: ignore[import-untyped]
|
from rpyc import ForkingServer, ThreadedServer # type: ignore[import-untyped]
|
||||||
@ -169,7 +169,7 @@ class Server:
|
|||||||
web_port: int = 8001,
|
web_port: int = 8001,
|
||||||
enable_rpc: bool = True,
|
enable_rpc: bool = True,
|
||||||
enable_web: bool = True,
|
enable_web: bool = True,
|
||||||
filename: Optional[str | Path] = None,
|
filename: str | Path | None = None,
|
||||||
use_forking_server: bool = False,
|
use_forking_server: bool = False,
|
||||||
additional_servers: list[AdditionalServer] | None = None,
|
additional_servers: list[AdditionalServer] | None = None,
|
||||||
**kwargs: Any,
|
**kwargs: Any,
|
||||||
@ -360,7 +360,7 @@ class Server:
|
|||||||
for sig in HANDLED_SIGNALS:
|
for sig in HANDLED_SIGNALS:
|
||||||
signal.signal(sig, self.handle_exit)
|
signal.signal(sig, self.handle_exit)
|
||||||
|
|
||||||
def handle_exit(self, sig: int = 0, frame: Optional[FrameType] = None) -> None:
|
def handle_exit(self, sig: int = 0, frame: FrameType | None = None) -> None:
|
||||||
if self.should_exit and sig == signal.SIGINT:
|
if self.should_exit and sig == signal.SIGINT:
|
||||||
logger.warning("Received signal '%s', forcing exit...", sig)
|
logger.warning("Received signal '%s', forcing exit...", sig)
|
||||||
os._exit(1)
|
os._exit(1)
|
||||||
|
@ -53,4 +53,4 @@ def convert_to_quantity(
|
|||||||
quantity = float(value["magnitude"]) * Unit(value["unit"])
|
quantity = float(value["magnitude"]) * Unit(value["unit"])
|
||||||
else:
|
else:
|
||||||
quantity = value
|
quantity = value
|
||||||
return quantity # type: ignore[reportUnknownMemberType]
|
return quantity
|
||||||
|
@ -1,12 +1,12 @@
|
|||||||
import inspect
|
import inspect
|
||||||
import logging
|
import logging
|
||||||
from itertools import chain
|
from itertools import chain
|
||||||
from typing import Any, Optional
|
from typing import Any
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
def get_attribute_doc(attr: Any) -> Optional[str]:
|
def get_attribute_doc(attr: Any) -> str | None:
|
||||||
"""This function takes an input attribute attr and returns its documentation
|
"""This function takes an input attribute attr and returns its documentation
|
||||||
string if it's different from the documentation of its type, otherwise,
|
string if it's different from the documentation of its type, otherwise,
|
||||||
it returns None.
|
it returns None.
|
||||||
@ -54,7 +54,7 @@ def get_object_attr_from_path_list(target_obj: Any, path: list[str]) -> Any:
|
|||||||
index_str = index_str.replace("]", "")
|
index_str = index_str.replace("]", "")
|
||||||
index = int(index_str)
|
index = int(index_str)
|
||||||
target_obj = getattr(target_obj, attr)[index]
|
target_obj = getattr(target_obj, attr)[index]
|
||||||
except ValueError: # noqa: PERF203
|
except ValueError:
|
||||||
# No index, so just get the attribute
|
# No index, so just get the attribute
|
||||||
target_obj = getattr(target_obj, part)
|
target_obj = getattr(target_obj, part)
|
||||||
except AttributeError:
|
except AttributeError:
|
||||||
@ -144,7 +144,7 @@ def update_value_if_changed(
|
|||||||
logger.error("Incompatible arguments: %s, %s.", target, attr_name_or_index)
|
logger.error("Incompatible arguments: %s, %s.", target, attr_name_or_index)
|
||||||
|
|
||||||
|
|
||||||
def parse_list_attr_and_index(attr_string: str) -> tuple[str, Optional[int]]:
|
def parse_list_attr_and_index(attr_string: str) -> tuple[str, int | None]:
|
||||||
"""
|
"""
|
||||||
Parses an attribute string and extracts a potential list attribute name and its
|
Parses an attribute string and extracts a potential list attribute name and its
|
||||||
index.
|
index.
|
||||||
|
@ -2,7 +2,6 @@ import asyncio
|
|||||||
import logging
|
import logging
|
||||||
import sys
|
import sys
|
||||||
from copy import copy
|
from copy import copy
|
||||||
from typing import Optional
|
|
||||||
|
|
||||||
import socketio # type: ignore[import-untyped]
|
import socketio # type: ignore[import-untyped]
|
||||||
import uvicorn.logging
|
import uvicorn.logging
|
||||||
@ -65,7 +64,7 @@ class SocketIOHandler(logging.Handler):
|
|||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
def setup_logging(level: Optional[str | int] = None) -> None:
|
def setup_logging(level: str | int | None = None) -> None:
|
||||||
"""
|
"""
|
||||||
Configures the logging settings for the application.
|
Configures the logging settings for the application.
|
||||||
|
|
||||||
|
@ -2,7 +2,7 @@ import inspect
|
|||||||
import logging
|
import logging
|
||||||
from collections.abc import Callable
|
from collections.abc import Callable
|
||||||
from enum import Enum
|
from enum import Enum
|
||||||
from typing import Any, Optional
|
from typing import Any
|
||||||
|
|
||||||
import pydase.units as u
|
import pydase.units as u
|
||||||
from pydase.data_service.abstract_data_service import AbstractDataService
|
from pydase.data_service.abstract_data_service import AbstractDataService
|
||||||
@ -130,7 +130,7 @@ class Serializer:
|
|||||||
|
|
||||||
# Store parameters and their anotations in a dictionary
|
# Store parameters and their anotations in a dictionary
|
||||||
sig = inspect.signature(obj)
|
sig = inspect.signature(obj)
|
||||||
parameters: dict[str, Optional[str]] = {}
|
parameters: dict[str, str | None] = {}
|
||||||
|
|
||||||
for k, v in sig.parameters.items():
|
for k, v in sig.parameters.items():
|
||||||
annotation = v.annotation
|
annotation = v.annotation
|
||||||
|
Loading…
x
Reference in New Issue
Block a user