Newer
Older
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
"""@package docstring
B-ASIC Save/Load SFG Module.
Given a structure try to serialize it and save it to a file.
Given a serialized file try to deserialize it and load it to the library.
"""
from b_asic import SFG, GraphComponent
from datetime import datetime
from inspect import signature
from os import path
def sfg_to_python(sfg: SFG, counter: int = 0, suffix: str = None) -> str:
result = (
"\n\"\"\"\nB-ASIC automatically generated SFG file.\n" +
"Name: " + f"{sfg.name}" + "\n" +
"Last saved: " + f"{datetime.now()}" + ".\n" +
"\"\"\""
)
result += "\nfrom b_asic import SFG, Signal, Input, Output"
for op in {type(op) for op in sfg.operations}:
result += f", {op.__name__}"
def kwarg_unpacker(comp: GraphComponent, params=None) -> str:
if params is None:
params_filtered = {attr: getattr(op, attr) for attr in signature(op.__init__).parameters if attr != "latency" and hasattr(op, attr)}
params = {attr: getattr(op, attr) if not isinstance(getattr(op, attr), str) else f'"{getattr(op, attr)}"' for attr in params_filtered}
return ", ".join([f"{param[0]}={param[1]}" for param in params.items()])
result += "\n# Inputs:\n"
for op in sfg._input_operations:
result += f"{op.graph_id} = Input({kwarg_unpacker(op)})\n"
result += "\n# Outputs:\n"
for op in sfg._output_operations:
result += f"{op.graph_id} = Output({kwarg_unpacker(op)})\n"
result += "\n# Operations:\n"
for op in sfg.split():
if isinstance(op, SFG):
counter += 1
result = sfg_to_python(op, counter) + result
continue
result += f"{op.graph_id} = {op.__class__.__name__}({kwarg_unpacker(op)})\n"
result += "\n# Signals:\n"
connections = list() # Keep track of already existing connections to avoid adding duplicates
for op in sfg.split():
for out in op.outputs:
for signal in out.signals:
dest_op = signal.destination.operation
connection = f"\nSignal(source={op.graph_id}.output({op.outputs.index(signal.source)}), destination={dest_op.graph_id}.input({dest_op.inputs.index(signal.destination)}))"
if connection in connections:
continue
result += connection
connections.append(connection)
inputs = "[" + ", ".join(op.graph_id for op in sfg.input_operations) + "]"
outputs = "[" + ", ".join(op.graph_id for op in sfg.output_operations) + "]"
sfg_name = sfg.name if sfg.name else "sfg" + str(counter) if counter > 0 else 'sfg'
result += f"\n{sfg_name} = SFG(inputs={inputs}, outputs={outputs}, name='{sfg_name}')\n"
result += "\n# SFG Properties:\n" + "prop = {'name':" + f"{sfg_name}" + "}"
if suffix is not None:
result += "\n" + suffix + "\n"
return result
def python_to_sfg(path: str) -> SFG:
with open(path) as f:
code = compile(f.read(), path, 'exec')
exec(code, globals(), locals())
return locals()["prop"]["name"], locals()["positions"] if "positions" in locals() else None