mirror of
https://github.com/tiqi-group/pydase.git
synced 2025-06-06 13:30:41 +02:00
chore: formatting
This commit is contained in:
parent
051e616280
commit
35f658ce4d
@ -6,7 +6,7 @@ from pydase.utils.logging import setup_logging
|
||||
setup_logging()
|
||||
|
||||
__all__ = [
|
||||
"Client",
|
||||
"DataService",
|
||||
"Server",
|
||||
"Client",
|
||||
]
|
||||
|
@ -192,7 +192,7 @@ class Client:
|
||||
async def _handle_connect(self) -> None:
|
||||
logger.debug("Connected to '%s' ...", self._url)
|
||||
serialized_object = cast(
|
||||
SerializedDataService, await self._sio.call("service_serialization")
|
||||
"SerializedDataService", await self._sio.call("service_serialization")
|
||||
)
|
||||
ProxyLoader.update_data_service_proxy(
|
||||
self.proxy, serialized_object=serialized_object
|
||||
|
@ -67,7 +67,7 @@ class ProxyClass(ProxyClassMixin, pydase.components.DeviceConnection):
|
||||
def serialize(self) -> SerializedObject:
|
||||
if self._service_representation is None:
|
||||
serialization_future = cast(
|
||||
asyncio.Future[SerializedDataService],
|
||||
"asyncio.Future[SerializedDataService]",
|
||||
asyncio.run_coroutine_threadsafe(
|
||||
self._sio.call("service_serialization"), self._loop
|
||||
),
|
||||
@ -80,7 +80,7 @@ class ProxyClass(ProxyClassMixin, pydase.components.DeviceConnection):
|
||||
self._service_representation = serialization_future.result()
|
||||
|
||||
device_connection_value = cast(
|
||||
dict[str, SerializedObject],
|
||||
"dict[str, SerializedObject]",
|
||||
pydase.components.DeviceConnection().serialize()["value"],
|
||||
)
|
||||
|
||||
@ -90,7 +90,7 @@ class ProxyClass(ProxyClassMixin, pydase.components.DeviceConnection):
|
||||
|
||||
value = {
|
||||
**cast(
|
||||
dict[str, SerializedObject],
|
||||
"dict[str, SerializedObject]",
|
||||
# need to deepcopy to not overwrite the _service_representation dict
|
||||
# when adding a prefix with add_prefix_to_full_access_path
|
||||
deepcopy(self._service_representation["value"]),
|
||||
|
@ -266,7 +266,7 @@ class ProxyLoader:
|
||||
return ProxyList(
|
||||
[
|
||||
ProxyLoader.loads_proxy(item, sio_client, loop)
|
||||
for item in cast(list[SerializedObject], serialized_object["value"])
|
||||
for item in cast("list[SerializedObject]", serialized_object["value"])
|
||||
],
|
||||
parent_path=serialized_object["full_access_path"],
|
||||
sio_client=sio_client,
|
||||
@ -283,7 +283,7 @@ class ProxyLoader:
|
||||
{
|
||||
key: ProxyLoader.loads_proxy(value, sio_client, loop)
|
||||
for key, value in cast(
|
||||
dict[str, SerializedObject], serialized_object["value"]
|
||||
"dict[str, SerializedObject]", serialized_object["value"]
|
||||
).items()
|
||||
},
|
||||
parent_path=serialized_object["full_access_path"],
|
||||
@ -300,7 +300,7 @@ class ProxyLoader:
|
||||
proxy_class._proxy_setters.clear()
|
||||
proxy_class._proxy_methods.clear()
|
||||
for key, value in cast(
|
||||
dict[str, SerializedObject], serialized_object["value"]
|
||||
"dict[str, SerializedObject]", serialized_object["value"]
|
||||
).items():
|
||||
type_handler: dict[str | None, None | Callable[..., Any]] = {
|
||||
None: None,
|
||||
@ -333,7 +333,7 @@ class ProxyLoader:
|
||||
) -> Any:
|
||||
# Custom types like Components or DataService classes
|
||||
component_class = cast(
|
||||
type, Deserializer.get_service_base_class(serialized_object["type"])
|
||||
"type", Deserializer.get_service_base_class(serialized_object["type"])
|
||||
)
|
||||
class_bases = (
|
||||
ProxyClassMixin,
|
||||
|
@ -33,8 +33,8 @@ from pydase.components.image import Image
|
||||
from pydase.components.number_slider import NumberSlider
|
||||
|
||||
__all__ = [
|
||||
"NumberSlider",
|
||||
"Image",
|
||||
"ColouredEnum",
|
||||
"DeviceConnection",
|
||||
"Image",
|
||||
"NumberSlider",
|
||||
]
|
||||
|
@ -56,13 +56,11 @@ class DataService(AbstractDataService):
|
||||
|
||||
def _is_unexpected_type_change(self, current_value: Any, new_value: Any) -> bool:
|
||||
return (
|
||||
isinstance(current_value, float)
|
||||
and not isinstance(new_value, float)
|
||||
or (
|
||||
isinstance(current_value, float) and not isinstance(new_value, float)
|
||||
) or (
|
||||
isinstance(current_value, u.Quantity)
|
||||
and not isinstance(new_value, u.Quantity)
|
||||
)
|
||||
)
|
||||
|
||||
def __warn_if_not_observable(self, value: Any, /) -> None:
|
||||
value_class = value if inspect.isclass(value) else value.__class__
|
||||
|
@ -46,13 +46,13 @@ class DataServiceCache:
|
||||
|
||||
def update_cache(self, full_access_path: str, value: Any) -> None:
|
||||
set_nested_value_by_path(
|
||||
cast(dict[str, SerializedObject], self._cache["value"]),
|
||||
cast("dict[str, SerializedObject]", self._cache["value"]),
|
||||
full_access_path,
|
||||
value,
|
||||
)
|
||||
|
||||
def get_value_dict_from_cache(self, full_access_path: str) -> SerializedObject:
|
||||
return get_nested_dict_by_path(
|
||||
cast(dict[str, SerializedObject], self._cache["value"]),
|
||||
cast("dict[str, SerializedObject]", self._cache["value"]),
|
||||
full_access_path,
|
||||
)
|
||||
|
@ -141,7 +141,7 @@ class StateManager:
|
||||
@property
|
||||
def cache_value(self) -> dict[str, SerializedObject]:
|
||||
"""Returns the "value" value of the DataService serialization."""
|
||||
return cast(dict[str, SerializedObject], self.cache_manager.cache["value"])
|
||||
return cast("dict[str, SerializedObject]", self.cache_manager.cache["value"])
|
||||
|
||||
def save_state(self) -> None:
|
||||
"""Saves the DataService's current state to a JSON file defined by
|
||||
@ -203,7 +203,7 @@ class StateManager:
|
||||
with open(self.filename) as f:
|
||||
# Load JSON data from file and update class attributes with these
|
||||
# values
|
||||
return cast(dict[str, Any], json.load(f))
|
||||
return cast("dict[str, Any]", json.load(f))
|
||||
return {}
|
||||
|
||||
def set_service_attribute_value_by_path(
|
||||
|
@ -85,7 +85,7 @@ class Deserializer:
|
||||
def deserialize_list(cls, serialized_object: SerializedObject) -> Any:
|
||||
return [
|
||||
cls.deserialize(item)
|
||||
for item in cast(list[SerializedObject], serialized_object["value"])
|
||||
for item in cast("list[SerializedObject]", serialized_object["value"])
|
||||
]
|
||||
|
||||
@classmethod
|
||||
@ -93,7 +93,7 @@ class Deserializer:
|
||||
return {
|
||||
key: cls.deserialize(value)
|
||||
for key, value in cast(
|
||||
dict[str, SerializedObject], serialized_object["value"]
|
||||
"dict[str, SerializedObject]", serialized_object["value"]
|
||||
).items()
|
||||
}
|
||||
|
||||
@ -148,7 +148,7 @@ class Deserializer:
|
||||
|
||||
# Process and add properties based on the serialized object
|
||||
for key, value in cast(
|
||||
dict[str, SerializedObject], serialized_object["value"]
|
||||
"dict[str, SerializedObject]", serialized_object["value"]
|
||||
).items():
|
||||
if value["type"] != "method":
|
||||
class_attrs[key] = cls.create_attr_property(value)
|
||||
|
@ -253,7 +253,7 @@ class Serializer:
|
||||
|
||||
for k, v in sig.parameters.items():
|
||||
default_value = cast(
|
||||
dict[str, Any], {} if v.default == inspect._empty else dump(v.default)
|
||||
"dict[str, Any]", {} if v.default == inspect._empty else dump(v.default)
|
||||
)
|
||||
default_value.pop("full_access_path", None)
|
||||
signature["parameters"][k] = {
|
||||
@ -385,7 +385,7 @@ def set_nested_value_by_path(
|
||||
current_dict, path_part, allow_append=False
|
||||
)
|
||||
current_dict = cast(
|
||||
dict[Any, SerializedObject],
|
||||
"dict[Any, SerializedObject]",
|
||||
next_level_serialized_object["value"],
|
||||
)
|
||||
|
||||
@ -426,7 +426,7 @@ def get_nested_dict_by_path(
|
||||
current_dict, path_part, allow_append=False
|
||||
)
|
||||
current_dict = cast(
|
||||
dict[Any, SerializedObject],
|
||||
"dict[Any, SerializedObject]",
|
||||
next_level_serialized_object["value"],
|
||||
)
|
||||
return get_container_item_by_key(current_dict, path_parts[-1], allow_append=False)
|
||||
@ -456,7 +456,7 @@ def get_or_create_item_in_container(
|
||||
return container[key]
|
||||
except IndexError:
|
||||
if allow_add_key and key == len(container):
|
||||
cast(list[SerializedObject], container).append(
|
||||
cast("list[SerializedObject]", container).append(
|
||||
create_empty_serialized_object()
|
||||
)
|
||||
return container[key]
|
||||
@ -541,7 +541,7 @@ def get_data_paths_from_serialized_object( # noqa: C901
|
||||
|
||||
elif serialized_dict_is_nested_object(serialized_obj):
|
||||
for key, value in cast(
|
||||
dict[str, SerializedObject], serialized_obj["value"]
|
||||
"dict[str, SerializedObject]", serialized_obj["value"]
|
||||
).items():
|
||||
# Serialized dictionaries need to have a different new_path than nested
|
||||
# classes
|
||||
@ -628,13 +628,13 @@ def add_prefix_to_full_access_path(
|
||||
|
||||
if isinstance(serialized_obj["value"], list):
|
||||
for value in serialized_obj["value"]:
|
||||
add_prefix_to_full_access_path(cast(SerializedObject, value), prefix)
|
||||
add_prefix_to_full_access_path(cast("SerializedObject", value), prefix)
|
||||
|
||||
elif isinstance(serialized_obj["value"], dict):
|
||||
for value in cast(
|
||||
dict[str, SerializedObject], serialized_obj["value"]
|
||||
"dict[str, SerializedObject]", serialized_obj["value"]
|
||||
).values():
|
||||
add_prefix_to_full_access_path(cast(SerializedObject, value), prefix)
|
||||
add_prefix_to_full_access_path(cast("SerializedObject", value), prefix)
|
||||
except (TypeError, KeyError, AttributeError):
|
||||
# passed dictionary is not a serialized object
|
||||
pass
|
||||
|
Loading…
x
Reference in New Issue
Block a user