Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

More typing issues #293

Merged
merged 11 commits into from
May 21, 2024
32 changes: 0 additions & 32 deletions src/ansys/optislang/core/nodes.py
Original file line number Diff line number Diff line change
Expand Up @@ -35,7 +35,6 @@

from ansys.optislang.core.managers import CriteriaManager, ParameterManager, ResponseManager
from ansys.optislang.core.node_types import NodeType
from ansys.optislang.core.osl_server import OslServer
from ansys.optislang.core.project_parametric import Design

PROJECT_COMMANDS_RETURN_STATES = {
Expand Down Expand Up @@ -1574,37 +1573,6 @@ def get_type_hint(self) -> str: # pragma: no cover
"""
pass

@staticmethod
@abstractmethod
def create_slot(
osl_server: OslServer,
node: Node,
name: str,
type_: SlotType,
type_hint: Optional[str] = None,
) -> Slot: # pragma: no cover
"""Create instance of new slot.

Parameters
----------
osl_server: OslServer
Object providing access to the optiSLang server.
node : Node
Node to which slot belongs to.
name : str
Slot name.
type_ : SlotType
Slot type.
type_hint : Optional[str], optional
Slot's expected data type, by default None.

Returns
-------
Slot
Instance of InputSlot, OutputSlot, InnerInputSlot or InnerOutputSlot class.
"""
pass


class InputSlot(Slot):
"""Provides for creating and operating on input slots."""
Expand Down
9 changes: 3 additions & 6 deletions src/ansys/optislang/core/tcp/managers.py
Original file line number Diff line number Diff line change
Expand Up @@ -212,7 +212,7 @@ def modify_criterion_property(
self.__osl_server.set_criterion_property(
uid=self.__uid,
criterion_name=criterion_name,
name=self.__class__.__PROPERTY_MAPPING.get(property_name, property_name),
name=self.__PROPERTY_MAPPING.get(property_name, property_name),
value=property_value,
)

Expand Down Expand Up @@ -304,10 +304,7 @@ def add_parameter(self, parameter: Parameter) -> None:
Raised when the timeout float value expires.
"""
parameter_manager, container = self.__get_parameter_container()
if (
self.__class__.__get_parameter_idx(container=container, parameter_name=parameter.name)
is not None
):
if self.__get_parameter_idx(container=container, parameter_name=parameter.name) is not None:
raise NameError(
f"Parameter `{parameter.name}` already exists, choose another name"
" or modify this parameter instead."
Expand Down Expand Up @@ -421,7 +418,7 @@ def modify_parameter_property(
idx = self.__get_parameter_idx(container=container, parameter_name=parameter_name)
if idx is not None:
container[idx][
self.__class__.__PROPERTY_MAPPING.get(property_name, property_name)
self.__PROPERTY_MAPPING.get(property_name, property_name)
] = property_value
parameter_manager["parameter_container"] = container
self.__osl_server.set_actor_property(
Expand Down
75 changes: 31 additions & 44 deletions src/ansys/optislang/core/tcp/nodes.py
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,7 @@
import logging
from pathlib import Path
import time
from typing import TYPE_CHECKING, Any, Dict, Iterable, List, Optional, Tuple, Type, Union
from typing import TYPE_CHECKING, Any, Dict, Iterable, List, Optional, Tuple, Type, Union, cast

from deprecated.sphinx import deprecated

Expand Down Expand Up @@ -357,7 +357,7 @@ def get_input_slots(self, name: Optional[str] = None) -> Tuple[TcpInputSlotProxy
TimeoutError
Raised when the timeout float value expires.
"""
return self._get_slots(type_=SlotType.INPUT, name=name)
return cast(Tuple[TcpInputSlotProxy, ...], self._get_slots(type_=SlotType.INPUT, name=name))

def get_name(self) -> str:
"""Get the name of the node.
Expand Down Expand Up @@ -405,7 +405,9 @@ def get_output_slots(self, name: Optional[str] = None) -> Tuple[TcpOutputSlotPro
TimeoutError
Raised when the timeout float value expires.
"""
return self._get_slots(type_=SlotType.OUTPUT, name=name)
return cast(
Tuple[TcpOutputSlotProxy, ...], self._get_slots(type_=SlotType.OUTPUT, name=name)
)

def get_parent(self) -> TcpNodeProxy:
"""Get the instance of the parent node.
Expand Down Expand Up @@ -912,7 +914,7 @@ def _find_ancestor_line(
"is_parametric_system": "ParameterManager" in node.get("properties", {}),
}
)
__class__._find_ancestor_line( # type: ignore
TcpNodeProxy._find_ancestor_line(
tree=node,
ancestor_line=ancestor_line,
node_uid=node_uid,
Expand Down Expand Up @@ -970,7 +972,7 @@ def _find_nodes_with_name(
if node["kind"] == "system" and (
current_depth < max_search_depth or max_search_depth == -1
):
__class__._find_nodes_with_name( # type: ignore
TcpNodeProxy._find_nodes_with_name(
name=name,
tree=node,
properties_dicts_list=properties_dicts_list,
Expand Down Expand Up @@ -1027,7 +1029,7 @@ def _find_node_with_uid(
if node["kind"] == "system" and (
current_depth < max_search_depth or max_search_depth == -1
):
__class__._find_node_with_uid( # type: ignore
TcpNodeProxy._find_node_with_uid(
uid=uid,
tree=node,
properties_dicts_list=properties_dicts_list,
Expand Down Expand Up @@ -1072,7 +1074,7 @@ def _find_parent_node_info(tree: dict, parent_info: dict, node_uid: str) -> dict
"kind": node["kind"],
"is_parametric_system": "ParameterManager" in node.get("properties", {}),
}
__class__._find_parent_node_info( # type: ignore
TcpNodeProxy._find_parent_node_info(
tree=node,
parent_info=new_parent_info,
node_uid=node_uid,
Expand Down Expand Up @@ -1626,6 +1628,12 @@ def create_node(
mop_node_type,
node_type,
) = self._get_subtypes(addin_type=type_.subtype)

if not isinstance(design_flow, (str, DesignFlow)):
raise TypeError(f"Design flow type: `{type(design_flow)}` is not supported.")
if isinstance(design_flow, str):
design_flow = DesignFlow.from_str(design_flow)

uid = self._osl_server.create_node(
type_=type_.id,
name=name,
Expand All @@ -1634,7 +1642,7 @@ def create_node(
mop_node_type=mop_node_type,
node_type=node_type,
parent_uid=self.uid,
design_flow=self._parse_design_flow(design_flow),
design_flow=design_flow.name.lower(),
)
info = self._osl_server.get_actor_info(
uid=uid, include_log_messages=False, include_integrations_registered_locations=False
Expand Down Expand Up @@ -1878,7 +1886,7 @@ def _find_subtree(tree: dict, uid: str, nodes_tree: List[dict]) -> dict:
if node["uid"] == uid:
nodes_tree.append(node)
if node["kind"] == "system":
__class__._find_subtree(tree=node, uid=uid, nodes_tree=nodes_tree) # type: ignore
TcpSystemProxy._find_subtree(tree=node, uid=uid, nodes_tree=nodes_tree)
return nodes_tree

@staticmethod
Expand Down Expand Up @@ -1920,33 +1928,6 @@ def _get_subtypes(

return algorithm_type, integration_type, mop_node_type, node_type

@staticmethod
def _parse_design_flow(design_flow: Union[DesignFlow, str, None]) -> Union[str, None]:
"""Parse ``design_flow`` argument from ``str`` to ``DesignFlow``.

Parameters
----------
design_flow : Union[DesignFlow, str]
Argument to be converted.

Returns
-------
Union[str, None]
Design flow type as string.

Raises
------
TypeError
Raised when unsupported type of ``design_flow`` argument was passed.
"""
if design_flow is None:
return None
if not isinstance(design_flow, (str, DesignFlow)):
raise TypeError(f"Design flow type: `{type(design_flow)}` is not supported.")
if isinstance(design_flow, str):
design_flow = DesignFlow.from_str(design_flow)
return design_flow.name.lower()


# endregion

Expand Down Expand Up @@ -2042,7 +2023,10 @@ def get_inner_input_slots(
TimeoutError
Raised when the timeout float value expires.
"""
return self._get_slots(type_=SlotType.INNER_INPUT, name=name)
return cast(
Tuple[TcpInnerInputSlotProxy, ...],
self._get_slots(type_=SlotType.INNER_INPUT, name=name),
)

def get_inner_output_slots(
self, name: Optional[str] = None
Expand All @@ -2068,7 +2052,10 @@ def get_inner_output_slots(
TimeoutError
Raised when the timeout float value expires.
"""
return self._get_slots(type_=SlotType.INNER_OUTPUT, name=name)
return cast(
Tuple[TcpInnerOutputSlotProxy, ...],
self._get_slots(type_=SlotType.INNER_OUTPUT, name=name),
)

def get_omdb_files(self) -> Tuple[File]:
"""Get paths to omdb files.
Expand Down Expand Up @@ -2872,14 +2859,14 @@ def _create_connection_script(from_slot: TcpSlotProxy, to_slot: TcpSlotProxy) ->
if isinstance(from_slot.node, TcpRootSystemProxy):
from_actor_script = "from_actor = project.get_root_system()\n"
else:
from_actor_script = __class__._create_find_actor_script( # type: ignore
from_actor_script = TcpSlotProxy._create_find_actor_script(
node=from_slot.node, name="from_actor"
)

if isinstance(to_slot.node, TcpRootSystemProxy):
to_actor_script = "to_actor = project.get_root_system()"
else:
to_actor_script = __class__._create_find_actor_script( # type: ignore
to_actor_script = TcpSlotProxy._create_find_actor_script(
node=to_slot.node, name="to_actor"
)

Expand Down Expand Up @@ -3449,10 +3436,10 @@ def _get_node_class_type(node_dict: dict, type_: NodeType) -> NodeClassType:
"""
# TODO: test
if node_dict["kind"] == "actor":
if type_.subtype in [
AddinType.PYTHON_BASED_INTEGRATION_PLUGIN,
AddinType.INTEGRATION_PLUGIN,
]:
if (
type_.subtype == AddinType.PYTHON_BASED_INTEGRATION_PLUGIN
or type_.subtype == AddinType.INTEGRATION_PLUGIN
):
return NodeClassType.INTEGRATION_NODE
return NodeClassType.NODE

Expand Down
Loading
Loading