diff --git a/hvcc/interpreters/pd2hv/HeavyGraph.py b/hvcc/interpreters/pd2hv/HeavyGraph.py
index 4a5b769..63fc98e 100644
--- a/hvcc/interpreters/pd2hv/HeavyGraph.py
+++ b/hvcc/interpreters/pd2hv/HeavyGraph.py
@@ -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
@@ -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
diff --git a/hvcc/interpreters/pd2hv/HeavyObject.py b/hvcc/interpreters/pd2hv/HeavyObject.py
index c9a1371..dbf7fba 100644
--- a/hvcc/interpreters/pd2hv/HeavyObject.py
+++ b/hvcc/interpreters/pd2hv/HeavyObject.py
@@ -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):
@@ -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": {
@@ -214,3 +215,5 @@ def to_hv(self) -> Dict:
},
"annotations": self.__annotations
}
+
+ return HvObject(**hv_obj)
diff --git a/hvcc/interpreters/pd2hv/HvSwitchcase.py b/hvcc/interpreters/pd2hv/HvSwitchcase.py
index 2d7834e..d0a1c90 100644
--- a/hvcc/interpreters/pd2hv/HvSwitchcase.py
+++ b/hvcc/interpreters/pd2hv/HvSwitchcase.py
@@ -14,9 +14,10 @@
# You should have received a copy of the GNU General Public License
# along with this program. If not, see .
-from typing import Optional, List, Dict
+from typing import Optional, List
from .PdObject import PdObject
+from .types import HvObject
class HvSwitchcase(PdObject):
@@ -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
@@ -51,3 +52,5 @@ def to_hv(self) -> Dict:
"y": self.pos_y
}
}
+
+ return HvObject(**hv_obj)
diff --git a/hvcc/interpreters/pd2hv/PdAudioIoObject.py b/hvcc/interpreters/pd2hv/PdAudioIoObject.py
index 6159f74..507732d 100644
--- a/hvcc/interpreters/pd2hv/PdAudioIoObject.py
+++ b/hvcc/interpreters/pd2hv/PdAudioIoObject.py
@@ -14,10 +14,11 @@
# You should have received a copy of the GNU General Public License
# along with this program. If not, see .
-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):
@@ -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]
@@ -50,3 +51,5 @@ def to_hv(self) -> Dict:
"y": self.pos_y
}
}
+
+ return HvObject(**hv_obj)
diff --git a/hvcc/interpreters/pd2hv/PdBinopObject.py b/hvcc/interpreters/pd2hv/PdBinopObject.py
index f7a217c..51016ba 100644
--- a/hvcc/interpreters/pd2hv/PdBinopObject.py
+++ b/hvcc/interpreters/pd2hv/PdBinopObject.py
@@ -14,11 +14,12 @@
# You should have received a copy of the GNU General Public License
# along with this program. If not, see .
-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):
@@ -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
@@ -156,3 +157,5 @@ def to_hv(self) -> Dict:
"y": self.pos_y
}
}
+
+ return HvObject(**hv_obj)
diff --git a/hvcc/interpreters/pd2hv/PdGraph.py b/hvcc/interpreters/pd2hv/PdGraph.py
index 271c9b4..46b8385 100644
--- a/hvcc/interpreters/pd2hv/PdGraph.py
+++ b/hvcc/interpreters/pd2hv/PdGraph.py
@@ -20,6 +20,7 @@
from .Connection import Connection
from .NotificationEnum import NotificationEnum
from .PdObject import PdObject
+from .types import HvGraph
class PdGraph(PdObject):
@@ -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 [],
@@ -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)
diff --git a/hvcc/interpreters/pd2hv/PdLetObject.py b/hvcc/interpreters/pd2hv/PdLetObject.py
index ad80c85..357f4ac 100644
--- a/hvcc/interpreters/pd2hv/PdLetObject.py
+++ b/hvcc/interpreters/pd2hv/PdLetObject.py
@@ -14,9 +14,10 @@
# You should have received a copy of the GNU General Public License
# along with this program. If not, see .
-from typing import Optional, List, Dict
+from typing import Optional, List
from .PdObject import PdObject
+from .types import HvObject
class PdLetObject(PdObject):
@@ -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
@@ -50,3 +51,5 @@ def to_hv(self) -> Dict:
"y": self.pos_y
}
}
+
+ return HvObject(**hv_obj)
diff --git a/hvcc/interpreters/pd2hv/PdMessageObject.py b/hvcc/interpreters/pd2hv/PdMessageObject.py
index e069434..6fe3c13 100644
--- a/hvcc/interpreters/pd2hv/PdMessageObject.py
+++ b/hvcc/interpreters/pd2hv/PdMessageObject.py
@@ -19,6 +19,7 @@
from .NotificationEnum import NotificationEnum
from .PdObject import PdObject
+from .types import HvObject
class PdMessageObject(PdObject):
@@ -73,8 +74,8 @@ 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": {
@@ -82,3 +83,5 @@ def to_hv(self) -> Dict:
"y": self.pos_y
}
}
+
+ return HvObject(**hv_obj)
diff --git a/hvcc/interpreters/pd2hv/PdObject.py b/hvcc/interpreters/pd2hv/PdObject.py
index 0eb1e5a..01c93c1 100644
--- a/hvcc/interpreters/pd2hv/PdObject.py
+++ b/hvcc/interpreters/pd2hv/PdObject.py
@@ -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
@@ -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()
diff --git a/hvcc/interpreters/pd2hv/PdPackObject.py b/hvcc/interpreters/pd2hv/PdPackObject.py
index 09d252d..1c36421 100644
--- a/hvcc/interpreters/pd2hv/PdPackObject.py
+++ b/hvcc/interpreters/pd2hv/PdPackObject.py
@@ -14,10 +14,11 @@
# You should have received a copy of the GNU General Public License
# along with this program. If not, see .
-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):
@@ -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
@@ -58,3 +59,5 @@ def to_hv(self) -> Dict:
"y": self.pos_y
}
}
+
+ return HvObject(**hv_obj)
diff --git a/hvcc/interpreters/pd2hv/PdReceiveObject.py b/hvcc/interpreters/pd2hv/PdReceiveObject.py
index a171fc8..c5bd54f 100644
--- a/hvcc/interpreters/pd2hv/PdReceiveObject.py
+++ b/hvcc/interpreters/pd2hv/PdReceiveObject.py
@@ -18,6 +18,7 @@
from .PdObject import PdObject
from .PdRaw import parse_pd_raw_args, PdRawException
+from .types import HvGraph
class PdReceiveObject(PdObject):
@@ -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": "",
@@ -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}",
@@ -135,3 +136,5 @@ def to_hv(self) -> Dict:
"scope": "public"
}
}
+
+ return HvGraph(**hv_graph)
diff --git a/hvcc/interpreters/pd2hv/PdRouteObject.py b/hvcc/interpreters/pd2hv/PdRouteObject.py
index 4e43c38..f5a782e 100644
--- a/hvcc/interpreters/pd2hv/PdRouteObject.py
+++ b/hvcc/interpreters/pd2hv/PdRouteObject.py
@@ -19,6 +19,7 @@
from .NotificationEnum import NotificationEnum
from .PdObject import PdObject
+from .types import HvGraph
class PdRouteObject(PdObject):
@@ -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.
@@ -165,4 +166,4 @@ def to_hv(self) -> Dict:
"type": "-->"
})
- return route_graph
+ return HvGraph(**route_graph)
diff --git a/hvcc/interpreters/pd2hv/PdSelectObject.py b/hvcc/interpreters/pd2hv/PdSelectObject.py
index 694bd6e..198561b 100644
--- a/hvcc/interpreters/pd2hv/PdSelectObject.py
+++ b/hvcc/interpreters/pd2hv/PdSelectObject.py
@@ -19,6 +19,7 @@
from .NotificationEnum import NotificationEnum
from .PdObject import PdObject
+from .types import HvGraph
class PdSelectObject(PdObject):
@@ -55,7 +56,7 @@ def validate_configuration(self) -> None:
if len(self._inlet_connections.get("1", [])) > 0:
self.add_warning("The right inlet of select 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.
@@ -152,4 +153,4 @@ def to_hv(self) -> Dict:
"type": "-->"
})
- return route_graph
+ return HvGraph(**route_graph)
diff --git a/hvcc/interpreters/pd2hv/PdSendObject.py b/hvcc/interpreters/pd2hv/PdSendObject.py
index 8772d2d..e7a4bcd 100644
--- a/hvcc/interpreters/pd2hv/PdSendObject.py
+++ b/hvcc/interpreters/pd2hv/PdSendObject.py
@@ -19,6 +19,7 @@
from .NotificationEnum import NotificationEnum
from .PdObject import PdObject
from .PdRaw import parse_pd_raw_args, PdRawException
+from .types import HvGraph
class PdSendObject(PdObject):
@@ -105,7 +106,7 @@ def validate_configuration(self) -> None:
"are not supported. A name should be given.",
NotificationEnum.ERROR_MISSING_REQUIRED_ARGUMENT)
- def to_hv(self) -> Dict:
+ def to_hv(self) -> HvGraph:
# note: control rate send/receive objects should not modify their name argument
names = {
"s": "",
@@ -115,7 +116,7 @@ def to_hv(self) -> Dict:
"throw~": "thrwctch_sig_"
}
- return {
+ hv_graph = {
"type": "send",
"args": {
"name": names[self.obj_type] + self.__send_name,
@@ -130,3 +131,5 @@ def to_hv(self) -> Dict:
"scope": "public"
}
}
+
+ return HvGraph(**hv_graph)
diff --git a/hvcc/interpreters/pd2hv/PdTableObject.py b/hvcc/interpreters/pd2hv/PdTableObject.py
index b3046e9..182957b 100644
--- a/hvcc/interpreters/pd2hv/PdTableObject.py
+++ b/hvcc/interpreters/pd2hv/PdTableObject.py
@@ -14,10 +14,11 @@
# You should have received a copy of the GNU General Public License
# along with this program. If not, see .
-from typing import Optional, List, Dict
+from typing import Optional, List
from .NotificationEnum import NotificationEnum
from .PdObject import PdObject
+from .types import HvGraph
class PdTableObject(PdObject):
@@ -53,8 +54,8 @@ def __init__(
except Exception:
pass
- def to_hv(self) -> Dict:
- return {
+ def to_hv(self) -> HvGraph:
+ hv_graph = {
"type": "table",
"args": {
"name": self.__table_name,
@@ -70,3 +71,5 @@ def to_hv(self) -> Dict:
"scope": "public"
}
}
+
+ return HvGraph(**hv_graph)
diff --git a/hvcc/interpreters/pd2hv/PdTriggerObject.py b/hvcc/interpreters/pd2hv/PdTriggerObject.py
index 4e38c3a..0d3b3bf 100644
--- a/hvcc/interpreters/pd2hv/PdTriggerObject.py
+++ b/hvcc/interpreters/pd2hv/PdTriggerObject.py
@@ -14,10 +14,11 @@
# You should have received a copy of the GNU General Public License
# along with this program. If not, see .
-from typing import Optional, List, Dict
+from typing import Optional, List
from .NotificationEnum import NotificationEnum
from .PdObject import PdObject
+from .types import HvObject
class PdTriggerObject(PdObject):
@@ -43,8 +44,8 @@ def __init__(
"Heavy only supports arguments 'a', 'f', 's', and 'b'.",
NotificationEnum.ERROR_TRIGGER_ABFS)
- def to_hv(self) -> Dict:
- return {
+ def to_hv(self) -> HvObject:
+ hv_obj = {
"type": "sequence",
"args": {
"casts": self.obj_args
@@ -55,6 +56,8 @@ def to_hv(self) -> Dict:
}
}
+ return HvObject(**hv_obj)
+
@classmethod
def __is_float(cls, x: int) -> bool:
""" Returns True if the input can be converted to a float. False otherwise.
diff --git a/hvcc/interpreters/pd2hv/PdUnpackObject.py b/hvcc/interpreters/pd2hv/PdUnpackObject.py
index 75ac3e5..cba731a 100644
--- a/hvcc/interpreters/pd2hv/PdUnpackObject.py
+++ b/hvcc/interpreters/pd2hv/PdUnpackObject.py
@@ -17,6 +17,7 @@
from typing import Optional, List, Dict
from .PdObject import PdObject
+from .types import HvGraph
class PdUnpackObject(PdObject):
@@ -35,7 +36,7 @@ def __init__(
if not (set(self.obj_args) <= set(["f", "s"])):
self.add_warning("Heavy only supports arguments 'f' and 's' to unpack.")
- def to_hv(self) -> Dict:
+ def to_hv(self) -> HvGraph:
""" Creates a graph dynamically based on the number of arguments.
[inlet ]
@@ -100,4 +101,4 @@ def to_hv(self) -> Dict:
"type": "-->"
})
- return hv_graph
+ return HvGraph(**hv_graph)
diff --git a/hvcc/interpreters/pd2hv/types/HvGraph.py b/hvcc/interpreters/pd2hv/types/HvGraph.py
new file mode 100644
index 0000000..6aa46ce
--- /dev/null
+++ b/hvcc/interpreters/pd2hv/types/HvGraph.py
@@ -0,0 +1,70 @@
+from typing import Dict, List, Literal, Optional
+
+from pydantic import BaseModel, Field
+
+
+ConnectionType = Literal["~f>", "-->"]
+ExternType = Literal["param", "event"]
+
+
+class From(BaseModel):
+ id: str
+ outlet: int
+
+
+class To(BaseModel):
+ id: str
+ inlet: int
+
+
+class Connection(BaseModel):
+ from_field: From = Field(..., alias="from")
+ to: To
+ type: ConnectionType
+
+
+class HvProperties(BaseModel):
+ x: int
+ y: int
+
+
+class HvArgs(BaseModel):
+ extern: Optional[ExternType] = None
+ index: int
+ name: Optional[str] = None
+ priority: Optional[int] = None
+ size: Optional[int] = None
+ values: Optional[List] = None
+ type: str
+ required: Optional[bool]
+ default: Optional[List]
+ value_type: Optional[str]
+
+
+class HvObject(BaseModel):
+ type: str
+ annotations: Optional[Dict[str, str]] = None
+ args: HvArgs
+ properties: HvProperties
+
+
+class HvGraph(BaseModel):
+ type: str
+ args: List[HvArgs]
+ connections: List[Connection]
+ imports: List[str]
+ objects: Dict[str, HvObject]
+ route_Ymaxs: HvObject
+ properties: HvProperties
+ annotations: Optional[Dict[str, str]]
+
+
+# if __name__ == "__main__":
+# import json
+# import importlib_resources
+
+# heavy_graph_json = importlib_resources.files('hvcc') / 'interpreters/pd2hv/libs/heavy_converted/lorenz~.hv.json'
+# with open(heavy_graph_json, "r") as f:
+# data = json.load(f)
+# heavy_graph = HvGraph(**data)
+# print(heavy_graph.keys())
diff --git a/hvcc/interpreters/pd2hv/types/__init__.py b/hvcc/interpreters/pd2hv/types/__init__.py
new file mode 100644
index 0000000..8de7cfa
--- /dev/null
+++ b/hvcc/interpreters/pd2hv/types/__init__.py
@@ -0,0 +1 @@
+from .HvGraph import HvGraph, HvObject # noqa