Skip to content

Commit

Permalink
try typing HeavyGraph
Browse files Browse the repository at this point in the history
  • Loading branch information
dromer committed Nov 10, 2024
1 parent 6eed5d6 commit 63b2b76
Show file tree
Hide file tree
Showing 19 changed files with 171 additions and 58 deletions.
38 changes: 20 additions & 18 deletions hvcc/interpreters/pd2hv/HeavyGraph.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,8 +16,9 @@

import json
import os
from typing import Optional, List, Dict
from typing import Optional, List

from .types import HvGraph
from .PdObject import PdObject
from .HeavyObject import HeavyObject

Expand All @@ -34,44 +35,45 @@ def __init__(

# read the heavy graph
with open(hv_path, "r") as f:
self.hv_json = json.load(f)
self.hv_json = HvGraph(**json.load(f))

# parse the heavy data structure to determine the outlet connection type
outlets = [o for o in self.hv_json["objects"].values() if o["type"] == "outlet"]
sorted(outlets, key=lambda o: o["args"]["index"])
self.__outlet_connection_types = [o["args"]["type"] for o in outlets]
outlets = [o for o in self.hv_json.objects.values() if o.type == "outlet"]
sorted(outlets, key=lambda o: o.args.index)
self.__outlet_connection_types = [o.args.type for o in outlets]

# resolve the arguments
for i, a in enumerate(self.hv_json["args"]):
for i, a in enumerate(self.hv_json.args):
if i < len(self.obj_args):
arg_value = self.obj_args[i]
elif a["required"]:
self.add_error(f"Required argument \"{a['name']}\" not found.")
elif a.required:
self.add_error(f"Required argument \"{a.name}\" not found.")
continue
else:
arg_value = a["default"]
arg_value = a.default

try:
arg_value = HeavyObject.force_arg_type(arg_value, a["value_type"])
arg_value = HeavyObject.force_arg_type(arg_value, a.value_type)
except Exception as e:
self.add_error(
f"Heavy {self.obj_type} cannot convert argument \"{a['name']}\""
f" with value \"{arg_value}\" to type {a['value_type']}: {e}")
f"Heavy {self.obj_type} cannot convert argument \"{a.name}\""
f" with value \"{arg_value}\" to type {a.value_type}: {e}")

# resolve all arguments for each object in the graph
for o in self.hv_json["objects"].values():
for k, v in o["args"].items():
for o in self.hv_json.objects.values():
for k, v in o.args:
# TODO(mhroth): make resolution more robust
if v == "$" + a["name"]:
o["args"][k] = arg_value
# if v == f"${a.name}":
# o.args[k] = arg_value
pass

# reset all arguments, as they have all been resolved
# any required arguments would break hv2ir as they will no longer
# be supplied (because they are resolved)
self.hv_json["args"] = []
self.hv_json.args = []

def get_outlet_connection_type(self, outlet_index: int) -> str:
return self.__outlet_connection_types[outlet_index]

def to_hv(self) -> Dict:
def to_hv(self) -> HvGraph:
return self.hv_json
9 changes: 6 additions & 3 deletions hvcc/interpreters/pd2hv/HeavyObject.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,12 +17,13 @@
import decimal
import json
import importlib_resources
from typing import Optional, List, Dict, Any, Union, cast
from typing import Optional, List, Any, Union, cast

from hvcc.core.hv2ir.types import HeavyIRType, HeavyLangType, IRNode, LangNode, IRArg, LangArg
from .Connection import Connection
from .NotificationEnum import NotificationEnum
from .PdObject import PdObject
from .types import HvObject


class HeavyObject(PdObject):
Expand Down Expand Up @@ -204,8 +205,8 @@ def add_connection(self, c: Connection) -> None:
else:
raise Exception("Adding a connection to the wrong object!")

def to_hv(self) -> Dict:
return {
def to_hv(self) -> HvObject:
hv_obj = {
"type": self.obj_type,
"args": self.obj_dict,
"properties": {
Expand All @@ -214,3 +215,5 @@ def to_hv(self) -> Dict:
},
"annotations": self.__annotations
}

return HvObject(**hv_obj)
9 changes: 6 additions & 3 deletions hvcc/interpreters/pd2hv/HvSwitchcase.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,9 +14,10 @@
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.

from typing import Optional, List, Dict
from typing import Optional, List

from .PdObject import PdObject
from .types import HvObject


class HvSwitchcase(PdObject):
Expand All @@ -40,8 +41,8 @@ def __init__(
def get_outlet_connection_type(self, outlet_index: int = 0) -> str:
return "-->"

def to_hv(self) -> Dict:
return {
def to_hv(self) -> HvObject:
hv_obj = {
"type": "__switchcase",
"args": {
"cases": self.obj_args
Expand All @@ -51,3 +52,5 @@ def to_hv(self) -> Dict:
"y": self.pos_y
}
}

return HvObject(**hv_obj)
9 changes: 6 additions & 3 deletions hvcc/interpreters/pd2hv/PdAudioIoObject.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,10 +14,11 @@
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.

from typing import Optional, List, Dict
from typing import Optional, List

from .NotificationEnum import NotificationEnum
from .PdObject import PdObject
from .types import HvObject


class PdAudioIoObject(PdObject):
Expand All @@ -39,8 +40,8 @@ def validate_configuration(self) -> None:
f"{self.obj_type} does not support control connections (inlet {i}). They should be removed.",
NotificationEnum.ERROR_UNSUPPORTED_CONNECTION)

def to_hv(self) -> Dict:
return {
def to_hv(self) -> HvObject:
hv_obj = {
"type": self.obj_type.strip("~"),
"args": {
"channels": [1, 2] if len(self.obj_args) == 0 else [int(a) for a in self.obj_args]
Expand All @@ -50,3 +51,5 @@ def to_hv(self) -> Dict:
"y": self.pos_y
}
}

return HvObject(**hv_obj)
9 changes: 6 additions & 3 deletions hvcc/interpreters/pd2hv/PdBinopObject.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,11 +14,12 @@
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.

from typing import Optional, List, Dict
from typing import Optional, List

from .Connection import Connection
from .HeavyObject import HeavyObject
from .PdObject import PdObject
from .types import HvObject


class PdBinopObject(PdObject):
Expand Down Expand Up @@ -145,8 +146,8 @@ def convert_ctrl_to_sig_connections_at_inlet(self, connection_list: List, inlet_
from_obj.remove_connection(old_conn)
self.remove_connection(old_conn)

def to_hv(self) -> Dict:
return {
def to_hv(self) -> HvObject:
hv_obj = {
"type": self.__PD_HEAVY_DICT[self.obj_type],
"args": {
"k": self.__k
Expand All @@ -156,3 +157,5 @@ def to_hv(self) -> Dict:
"y": self.pos_y
}
}

return HvObject(**hv_obj)
7 changes: 5 additions & 2 deletions hvcc/interpreters/pd2hv/PdGraph.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
from .Connection import Connection
from .NotificationEnum import NotificationEnum
from .PdObject import PdObject
from .types import HvGraph


class PdGraph(PdObject):
Expand Down Expand Up @@ -218,12 +219,12 @@ def get_depth(self) -> int:
# NOTE(dromer): we should never get here
raise Exception("parent_graph argument is None")

def to_hv(self, export_args: bool = False) -> Dict:
def to_hv(self, export_args: bool = False) -> HvGraph:
# NOTE(mhroth): hv_args are not returned. Because all arguments have
# been resolved, no arguments are otherwise passed. hv2ir would break
# on required arguments that are not passed to the graph
assert all(a is not None for a in self.hv_args), "Graph is missing a @hv_arg."
return {
hvgraph = {
"type": "graph",
"imports": [],
"args": self.hv_args if export_args else [],
Expand All @@ -235,5 +236,7 @@ def to_hv(self, export_args: bool = False) -> Dict:
}
}

return HvGraph(**hvgraph)

def __repr__(self) -> str:
return self.subpatch_name or os.path.basename(self.__pd_path)
9 changes: 6 additions & 3 deletions hvcc/interpreters/pd2hv/PdLetObject.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,9 +14,10 @@
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.

from typing import Optional, List, Dict
from typing import Optional, List

from .PdObject import PdObject
from .types import HvObject


class PdLetObject(PdObject):
Expand All @@ -37,8 +38,8 @@ def get_outlet_connection_type(self, outlet_index: int) -> Optional[str]:
else:
return super().get_outlet_connection_type(outlet_index)

def to_hv(self) -> Dict:
return {
def to_hv(self) -> HvObject:
hv_obj = {
"type": self.obj_type.strip("~"),
"args": {
"name": "", # Pd does not give an inlet name
Expand All @@ -50,3 +51,5 @@ def to_hv(self) -> Dict:
"y": self.pos_y
}
}

return HvObject(**hv_obj)
7 changes: 5 additions & 2 deletions hvcc/interpreters/pd2hv/PdMessageObject.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@

from .NotificationEnum import NotificationEnum
from .PdObject import PdObject
from .types import HvObject


class PdMessageObject(PdObject):
Expand Down Expand Up @@ -73,12 +74,14 @@ def __init__(
"message": l_split[1:]
})

def to_hv(self) -> Dict:
return {
def to_hv(self) -> HvObject:
hv_obj = {
"type": "message",
"args": self.obj_dict,
"properties": {
"x": self.pos_x,
"y": self.pos_y
}
}

return HvObject(**hv_obj)
5 changes: 3 additions & 2 deletions hvcc/interpreters/pd2hv/PdObject.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,8 +18,9 @@
import random
import string

from typing import Optional, List, Dict, TYPE_CHECKING
from typing import Optional, List, Dict, TYPE_CHECKING, Union

from .types import HvGraph, HvObject
from .Connection import Connection
from .NotificationEnum import NotificationEnum

Expand Down Expand Up @@ -175,7 +176,7 @@ def get_supported_objects(cls) -> set:
"""
raise NotImplementedError()

def to_hv(self) -> Dict:
def to_hv(self) -> Union[HvGraph, HvObject]:
""" Returns the HeavyLang JSON representation of this object.
"""
raise NotImplementedError()
Expand Down
9 changes: 6 additions & 3 deletions hvcc/interpreters/pd2hv/PdPackObject.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,10 +14,11 @@
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.

from typing import Optional, List, Dict
from typing import Optional, List

from .NotificationEnum import NotificationEnum
from .PdObject import PdObject
from .types import HvObject


class PdPackObject(PdObject):
Expand Down Expand Up @@ -47,8 +48,8 @@ def __init__(
f"\"{x}\" argument to [pack] object not supported.",
NotificationEnum.ERROR_PACK_FLOAT_ARGUMENTS)

def to_hv(self) -> Dict:
return {
def to_hv(self) -> HvObject:
hv_obj = {
"type": "__pack",
"args": {
"values": self.values
Expand All @@ -58,3 +59,5 @@ def to_hv(self) -> Dict:
"y": self.pos_y
}
}

return HvObject(**hv_obj)
7 changes: 5 additions & 2 deletions hvcc/interpreters/pd2hv/PdReceiveObject.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@

from .PdObject import PdObject
from .PdRaw import parse_pd_raw_args, PdRawException
from .types import HvGraph


class PdReceiveObject(PdObject):
Expand Down Expand Up @@ -103,7 +104,7 @@ def validate_configuration(self) -> None:
if len(self._inlet_connections.get("0", [])) > 0:
self.add_error("[receive~] inlet connections are not supported.")

def to_hv(self) -> Dict:
def to_hv(self) -> HvGraph:
# note: control rate send objects should not modify their name argument
names = {
"r": "",
Expand All @@ -119,7 +120,7 @@ def to_hv(self) -> Dict:
((self.__priority is None) or (self.__receiver_name == "__hv_init" and self.__priority == 0)):
self.__priority = (self.parent_graph.get_depth() * 1000) - self.__instance

return {
hv_graph = {
"type": "receive",
"args": {
"name": f"{names[self.obj_type]}{self.__receiver_name}",
Expand All @@ -135,3 +136,5 @@ def to_hv(self) -> Dict:
"scope": "public"
}
}

return HvGraph(**hv_graph)
5 changes: 3 additions & 2 deletions hvcc/interpreters/pd2hv/PdRouteObject.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@

from .NotificationEnum import NotificationEnum
from .PdObject import PdObject
from .types import HvGraph


class PdRouteObject(PdObject):
Expand Down Expand Up @@ -59,7 +60,7 @@ def validate_configuration(self) -> None:
if len(self._inlet_connections.get("1", [])) > 0:
self.add_warning("The right inlet of route is not supported. It will not do anything.")

def to_hv(self) -> Dict:
def to_hv(self) -> HvGraph:
"""Creates a graph dynamically based on the number of arguments.
An unconnected right inlet is added.
Expand Down Expand Up @@ -165,4 +166,4 @@ def to_hv(self) -> Dict:
"type": "-->"
})

return route_graph
return HvGraph(**route_graph)
Loading

0 comments on commit 63b2b76

Please sign in to comment.