1+ import hls4ml
2+ import pytest
3+
4+ '''
5+ Tests for model flows.
6+ Construct some dummy optimizer passes and flows that do nothing.
7+ Passes record their label to the model.
8+ Tests check that the order of applied passes matches the expectations
9+ '''
10+
11+ class DummyPass (hls4ml .model .optimizer .OptimizerPass ):
12+ def __init__ (self , label ):
13+ self .label = label
14+ def match (self , node ):
15+ return True
16+ def transform (self , model , node ):
17+ if getattr (model , 'test_flow_passes' , None ) is None :
18+ model .test_flow_passes = []
19+ model .test_flow_passes .append (self .label )
20+ return False
21+
22+ class DummyPassA (DummyPass ):
23+ def __init__ (self ):
24+ super ().__init__ ('A' )
25+ class DummyPassB (DummyPass ):
26+ def __init__ (self ):
27+ super ().__init__ ('B' )
28+ class DummyPassC (DummyPass ):
29+ def __init__ (self ):
30+ super ().__init__ ('C' )
31+
32+ hls4ml .model .optimizer .register_pass ('A' , DummyPassA )
33+ hls4ml .model .optimizer .register_pass ('B' , DummyPassB )
34+ hls4ml .model .optimizer .register_pass ('C' , DummyPassC )
35+
36+ DummyFlowA = hls4ml .model .flow .register_flow ('A' , ['A' ])
37+ DummyFlowB = hls4ml .model .flow .register_flow ('B' , ['B' ])
38+ DummyFlowC = hls4ml .model .flow .register_flow ('C' , ['C' ])
39+ DummyFlowAB = hls4ml .model .flow .register_flow ('AB' , ['A' , 'B' ])
40+ DummyFlowBReqA = hls4ml .model .flow .register_flow ('BReqA' , ['B' ], requires = [DummyFlowA ])
41+ DummyFlowCReqBReqA = hls4ml .model .flow .register_flow ('CReqBReqA' , ['C' ], requires = [DummyFlowBReqA ])
42+
43+ def dummy_flow_model ():
44+ layers = [{'class_name' : 'Input' , 'name' : 'layer0_input' , 'input_shape' : [1 ]}]
45+ config = {'HLSConfig' :{'Model' :{'Precision' :'ap_fixed<32,16>' ,'ReuseFactor' : 1 },
46+ 'Flows' : []}}
47+ model = hls4ml .model .ModelGraph (config , None , layers )
48+ return model
49+
50+ class FlowTester :
51+ index = 0
52+ def __init__ (self , flows_to_apply , expected_pass_order , reapply ):
53+ self .flows_to_apply = flows_to_apply
54+ self .expected_pass_order = expected_pass_order
55+ self .reapply = reapply
56+ self .index = FlowTester .index
57+ FlowTester .index += 1
58+ def run (self ):
59+ model = dummy_flow_model ()
60+ model .test_flow_passes = []
61+ for flow in self .flows_to_apply :
62+ model .apply_flow (flow , self .reapply )
63+ self .observed_pass_order = model .test_flow_passes
64+ return self .observed_pass_order == self .expected_pass_order
65+
66+ flow_tests = [FlowTester (['A' , 'B' , 'C' ], ['A' , 'B' , 'C' ], 'single' ), # independent flows in order
67+ FlowTester (['A' , 'A' ], ['A' , 'A' ], 'single' ), # same flow twice, single application
68+ FlowTester (['A' , 'A' ], ['A' , 'A' ], 'all' ), # same flow twice with reapply
69+ FlowTester (['A' , 'A' ], ['A' ], 'none' ), # same flow twice with none
70+ FlowTester (['BReqA' ], ['A' , 'B' ], 'single' ), # one flow with a dependency
71+ FlowTester (['CReqBReqA' ], ['A' , 'B' , 'C' ], 'single' ), # one flow with dependency chain
72+ FlowTester (['CReqBReqA' , 'A' ], ['A' , 'B' , 'C' , 'A' ], 'single' ), # one flow with dependency chain, repeat dependency
73+ FlowTester (['CReqBReqA' , 'A' ], ['A' , 'B' , 'C' , 'A' ], 'all' ), # one flow with dependency chain, repeat dependency
74+ FlowTester (['CReqBReqA' , 'A' ], ['A' , 'B' , 'C' ], 'none' ), # one flow with dependency chain, repeat depencency
75+ FlowTester (['A' , 'CReqBReqA' ], ['A' , 'B' , 'C' ], 'single' ), # one flow with dependency chain, repeat depencency
76+ FlowTester (['A' , 'CReqBReqA' ], ['A' , 'A' , 'B' , 'C' ], 'all' ), # one flow with dependency chain, repeat depencency
77+ FlowTester (['A' , 'CReqBReqA' ], ['A' , 'B' , 'C' ], 'none' ), # one flow with dependency chain, repeat depencency
78+ FlowTester (['A' , 'BReqA' ], ['A' , 'B' ], 'single' ), # second flow dependency already run
79+ FlowTester (['A' , 'BReqA' ], ['A' , 'A' , 'B' ], 'all' ), # second flow dependency reapply
80+ FlowTester (['A' , 'BReqA' ], ['A' , 'B' ], 'none' ), # second flow dependency no reapply
81+ FlowTester (['A' , 'A' , 'BReqA' ], ['A' , 'A' , 'A' , 'B' ], 'all' ), # second flow dependency reapply
82+ FlowTester (['A' , 'A' , 'BReqA' ], ['A' , 'B' ], 'none' ), # second flow dependency no reapply
83+ FlowTester (['A' , 'A' , 'BReqA' ], ['A' , 'A' , 'B' ], 'single' ), # second flow dependency skip requirements
84+ FlowTester (['A' , 'BReqA' , 'CReqBReqA' ], ['A' , 'B' , 'C' ], 'single' ), # two flows depending on earlier flows
85+ FlowTester (['A' , 'BReqA' , 'CReqBReqA' ], ['A' , 'B' , 'C' ], 'none' ), # two flows depending on earlier flows
86+ FlowTester (['A' , 'BReqA' , 'CReqBReqA' ], ['A' , 'A' , 'B' , 'A' , 'B' , 'C' ], 'all' ), # three flows depending on earlier flows
87+ FlowTester (['CReqBReqA' , 'BReqA' , 'A' ], ['A' , 'B' , 'C' , 'B' , 'A' ], 'single' ), # three flows depending on earlier flows
88+ FlowTester (['CReqBReqA' , 'BReqA' , 'A' ], ['A' , 'B' , 'C' ], 'none' ), # three flows depending on earlier flows
89+ FlowTester (['CReqBReqA' , 'BReqA' , 'A' ], ['A' , 'B' , 'C' , 'A' , 'B' , 'A' ], 'all' ), # three flows depending on earlier flows
90+ FlowTester (['A' , 'CReqBReqA' , 'BReqA' , 'A' ], ['A' , 'B' , 'C' , 'B' , 'A' ], 'single' ), # three flows depending on earlier flows
91+ FlowTester (['A' , 'CReqBReqA' , 'BReqA' , 'A' ], ['A' , 'B' , 'C' ], 'none' ), # three flows depending on earlier flows
92+ FlowTester (['A' , 'CReqBReqA' , 'BReqA' , 'A' ], ['A' , 'A' , 'B' , 'C' , 'A' , 'B' , 'A' ], 'all' ), # three flows depending on earlier flows
93+ ]
94+
95+ @pytest .mark .parametrize ('tester' , flow_tests )
96+ def test_flows (tester ):
97+ success = tester .run ()
98+ i = tester .index
99+ expected = tester .expected_pass_order
100+ observed = tester .observed_pass_order
101+ assert success , f'Tester { i } { expected } { observed }
0 commit comments