Newer
Older
Angus Lothian
committed
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
"""B-ASIC Save/Load Structure Module.
Contains functions for saving/loading SFGs to/from strings that can be stored
as files.
"""
from b_asic.signal_flow_graph import SFG
from b_asic.graph_component import 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:
"""Given an SFG structure try to serialize it for saving to a file."""
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"
# Keep track of already existing connections to avoid adding duplicates
Angus Lothian
committed
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
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'
sfg_name_var = sfg_name.replace(" ", "_")
result += f"\n{sfg_name_var} = SFG(inputs={inputs}, outputs={outputs}, name='{sfg_name}')\n"
result += "\n# SFG Properties:\n" + \
"prop = {'name':" + f"{sfg_name_var}" + "}"
if suffix is not None:
result += "\n" + suffix + "\n"
return result
def python_to_sfg(path: str) -> SFG:
"""Given a serialized file try to deserialize it and load it to the library."""
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 {}