Skip to content

Commit

Permalink
feat: type hint
Browse files Browse the repository at this point in the history
  • Loading branch information
mkundu1 committed Jan 9, 2025
1 parent 5e540d1 commit 9cca9e7
Showing 1 changed file with 28 additions and 23 deletions.
51 changes: 28 additions & 23 deletions src/ansys/fluent/core/services/datamodel_se.py
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,7 @@

Path = list[tuple[str, str]]
PyMenuT = TypeVar("PyMenuT", bound="PyMenu")
_TValue = None | bool | int | float | str | Sequence["_TValue"] | dict[str, "_TValue"]
ValueT = None | bool | int | float | str | Sequence["ValueT"] | dict[str, "ValueT"]
logger: logging.Logger = logging.getLogger("pyfluent.datamodel")

member_specs_oneof_fields = [
Expand Down Expand Up @@ -304,7 +304,7 @@ def unsubscribe_events(
return self._stub.unsubscribeEvents(request, metadata=self._metadata)


def _convert_value_to_variant(val: _TValue, var: Variant) -> None:
def _convert_value_to_variant(val: ValueT, var: Variant) -> None:
"""Convert a Python data type to Fluent's variant type."""
if isinstance(val, bool):
var.bool_state = val
Expand All @@ -325,7 +325,7 @@ def _convert_value_to_variant(val: _TValue, var: Variant) -> None:
_convert_value_to_variant(v, var.variant_map_state.item[k])


def _convert_variant_to_value(var: Variant) -> _TValue:
def _convert_variant_to_value(var: Variant) -> ValueT:
"""Convert Fluent's variant type to a Python data type."""
if var.HasField("bool_state"):
return var.bool_state
Expand Down Expand Up @@ -498,15 +498,15 @@ def __init__(
self.cache = DataModelCache() if pyfluent.DATAMODEL_USE_STATE_CACHE else None
self.version = version

def get_attribute_value(self, rules: str, path: str, attribute: str) -> _TValue:
def get_attribute_value(self, rules: str, path: str, attribute: str) -> ValueT:
"""Get attribute value."""
request = DataModelProtoModule.GetAttributeValueRequest(
rules=rules, path=path, attribute=attribute
)
response = self._impl.get_attribute_value(request)
return _convert_variant_to_value(response.result)

def get_state(self, rules: str, path: str) -> _TValue:
def get_state(self, rules: str, path: str) -> ValueT:
"""Get state."""
request = DataModelProtoModule.GetStateRequest(rules=rules, path=path)
response = self._impl.get_state(request)
Expand Down Expand Up @@ -571,7 +571,7 @@ def delete_all_child_objects(self, rules: str, path: str, obj_type: str) -> None
version=self.version,
)

def set_state(self, rules: str, path: str, state: _TValue) -> None:
def set_state(self, rules: str, path: str, state: ValueT) -> None:
"""Set state."""
request = DataModelProtoModule.SetStateRequest(
rules=rules, path=path, wait=True
Expand Down Expand Up @@ -604,7 +604,7 @@ def update_dict(
self,
rules: str,
path: str,
dict_state: dict[str, _TValue],
dict_state: dict[str, ValueT],
recursive=False,
) -> None:
"""Update the dict."""
Expand Down Expand Up @@ -636,8 +636,8 @@ def delete_object(self, rules: str, path: str) -> None:
)

def execute_command(
self, rules: str, path: str, command: str, args: dict[str, _TValue]
) -> _TValue:
self, rules: str, path: str, command: str, args: dict[str, ValueT]
) -> ValueT:
"""Execute the command."""
request = DataModelProtoModule.ExecuteCommandRequest(
rules=rules, path=path, command=command, wait=True
Expand All @@ -654,8 +654,8 @@ def execute_command(
return _convert_variant_to_value(response.result)

def execute_query(
self, rules: str, path: str, query: str, args: dict[str, _TValue]
) -> _TValue:
self, rules: str, path: str, query: str, args: dict[str, ValueT]
) -> ValueT:
"""Execute the query."""
request = DataModelProtoModule.ExecuteQueryRequest(
rules=rules, path=path, query=query
Expand Down Expand Up @@ -767,7 +767,7 @@ def add_on_deleted(
return subscription

def add_on_changed(
self, rules: str, path: str, obj, cb: Callable[[Any], None]
self, rules: str, path: str, obj, cb: Callable[[ValueT], None]
) -> EventSubscription:
"""Add on changed."""
request_dict = {
Expand Down Expand Up @@ -818,7 +818,12 @@ def add_on_affected_at_type_path(
return subscription

def add_on_command_executed_old(
self, rules: str, path: str, command: str, obj, cb: Callable[[str, Any], None]
self,
rules: str,
path: str,
command: str,
obj,
cb: Callable[[str, ValueT], None],
) -> EventSubscription:
"""Add on command executed."""
request_dict = {
Expand All @@ -837,7 +842,7 @@ def add_on_command_executed_old(
return subscription

def add_on_command_executed(
self, rules: str, path: str, obj, cb: Callable[[str, Any], None]
self, rules: str, path: str, obj, cb: Callable[[str, ValueT], None]
) -> EventSubscription:
"""Add on command executed."""
request_dict = {
Expand All @@ -855,7 +860,7 @@ def add_on_command_executed(
return subscription

def add_on_attribute_changed(
self, rules: str, path: str, attribute: str, obj, cb: Callable[[Any], None]
self, rules: str, path: str, attribute: str, obj, cb: Callable[[ValueT], None]
) -> EventSubscription:
"""Add on attribute changed."""
request_dict = {
Expand All @@ -880,7 +885,7 @@ def add_on_command_attribute_changed(
command: str,
attribute: str,
obj,
cb: Callable[[Any], None],
cb: Callable[[ValueT], None],
) -> EventSubscription:
"""Add on command attribute changed."""
request_dict = {
Expand Down Expand Up @@ -1352,7 +1357,7 @@ def add_on_changed(self, cb: Callable[[PyMenuT], None]) -> EventSubscription:
EventSubscription instance which can be used to unregister the callback
"""

def cb_service(value: Any):
def cb_service(value: ValueT):
cb(self)

return self.service.add_on_changed(
Expand Down Expand Up @@ -1406,15 +1411,15 @@ def cb_service():
)

def add_on_command_executed_old(
self, command: str, cb: Callable[[PyMenuT, str, Any], None]
self, command: str, cb: Callable[[PyMenuT, str, ValueT], None]
) -> EventSubscription:
"""Register a callback for when a command is executed.
Parameters
----------
command : str
Command name
cb : Callable[[PyMenuT, str, Any], None]
cb : Callable[[PyMenuT, str, ValueT], None]
Callback function
Returns
Expand All @@ -1423,21 +1428,21 @@ def add_on_command_executed_old(
EventSubscription instance which can be used to unregister the callback
"""

def cb_service(command: str, args: Any):
def cb_service(command: str, args: ValueT):
cb(self, command, args)

return self.service.add_on_command_executed_old(
self.rules, convert_path_to_se_path(self.path), command, self, cb_service
)

def add_on_command_executed(
self, cb: Callable[[PyMenuT, str, Any], None]
self, cb: Callable[[PyMenuT, str, ValueT], None]
) -> EventSubscription:
"""Register a callback for when a command is executed.
Parameters
----------
cb : Callable[[PyMenuT, str, Any], None]
cb : Callable[[PyMenuT, str, ValueT], None]
Callback function
Returns
Expand All @@ -1446,7 +1451,7 @@ def add_on_command_executed(
EventSubscription instance which can be used to unregister the callback
"""

def cb_service(command: str, args: Any):
def cb_service(command: str, args: ValueT):
cb(self, command, args)

return self.service.add_on_command_executed(
Expand Down

0 comments on commit 9cca9e7

Please sign in to comment.