hls4ml tutorial comparison

.github/workflows/python-app.yml

@@ -36,7 +36,7 @@ jobs:
         ruff format --diff
     - name: Test with pytest
       run: |
-        pytest test --cov exseos
+        pytest test --cov exseos -m "not needs_vendor_tools"
     - name: Upload coverage reports to Codecov
       uses: codecov/[email protected]
       with:

environment.yml

@@ -1,9 +1,9 @@
-name: exseos-h
+name: exseos-h-py311
 channels:
   - conda-forge
   - nodefaults
 dependencies:
-  - python=3.12
+  - python=3.11
   - patch
   - pip
   - graphviz
@@ -12,8 +12,8 @@ dependencies:
     - hls4ml==0.8.0
     - matplotlib
     - scikit-learn
-    - tensorflow
-    - qkeras
+    - tensorflow==2.12.0
+    - qkeras==0.9.0
     - conifer
     - colorlog
     - psutil

exseos/experiment/Constant.py

@@ -16,15 +16,15 @@
 
 
 from abc import ABC
-from typing import Any, Callable, TypeVar
+from typing import Any, Callable, TypeVar, Generic
 
 from exseos.types.Option import Option
 from exseos.types.Variable import Variable, VariableSet
 
 A = TypeVar("A")
 
 
-class ExperimentConstant[A](ABC):
+class ExperimentConstant(ABC, Generic[A]):
 	@property
 	def name(self) -> str: ...  # pragma: no cover
 
@@ -33,7 +33,7 @@ def can_resolve(self, vars: VariableSet) -> bool: ...  # pragma: no cover
 	def resolve(self, vars: VariableSet) -> A: ...  # pragma: no cover
 
 
-class BasicExperimentConstant[A](ExperimentConstant):
+class BasicExperimentConstant(ExperimentConstant):
 	def __init__(self, name: str, val: A):
 		self.__name = name
 		self.__val = val
@@ -49,7 +49,7 @@ def resolve(self, _) -> A:
 		return self.__val
 
 
-class LambdaExperimentConstant[A](ExperimentConstant):
+class LambdaExperimentConstant(ExperimentConstant):
 	def __init__(self, name: str, fn: Callable[[VariableSet], Option[Any]]):
 		self.__name = name
 		self.__fn = fn

exseos/experiment/Experiment.py

@@ -93,7 +93,6 @@ def _resolve_constants(
 		vars = opt_inputs
 
 		while unresolved_constants:
-			print(vars)
 			newly_resolved = tuple(
 				[
 					BoundVariable(c.name, c.resolve(VariableSet(vars)))

exseos/plugin/hls4ml/stage/ConvertKerasModel.py

@@ -0,0 +1,82 @@
+# ExSeOS-H Hardware ML Workflow Manager
+# Copyright (C) 2024  Alexis Maya-Isabelle Shuping
+
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+from exseos.types.Option import Some
+from exseos.types.Result import Okay, Fail
+from exseos.types.Variable import VariableSet, UnboundVariable
+from exseos.workflow.stage.Stage import Stage
+
+from keras import Model
+from hls4ml.model import ModelGraph
+from hls4ml.utils import config_from_keras_model
+from hls4ml.converters import convert_from_keras_model
+
+
+class ConvertKerasModel(Stage):
+	input_vars = (
+		UnboundVariable("model", Model, "The model to convert."),
+		UnboundVariable("granularity", str, default="model"),
+		UnboundVariable("default_precision", str, default="fixed<16,6>"),
+		UnboundVariable("default_reuse_factor", int, default=1),
+		UnboundVariable("output_dir", str, default="my-hls-test"),
+		UnboundVariable("project_name", "myproject"),
+		UnboundVariable("input_data_tb", str | None, default=Some(None)),
+		UnboundVariable("output_data_tb", str | None, default=Some(None)),
+		UnboundVariable("backend", str, "HLS4ML backend to use", default="Vivado"),
+		UnboundVariable("board", str | None, default=Some(None)),
+		UnboundVariable("part", str | None, default=Some(None)),
+		UnboundVariable("io_type", str, default="io_parallel"),
+		UnboundVariable("hls_config", dict | None, default=None),
+	)
+	output_vars = (
+		UnboundVariable("hls_model", ModelGraph, "Converted model"),
+		UnboundVariable("config", desc="Final HLS4ML-generated model configuration"),
+	)
+
+	async def run(self, inputs: VariableSet, _):
+		stat = inputs.check_all()
+		if stat.is_fail:
+			return stat
+
+		try:
+			config = config_from_keras_model(
+				inputs.model,
+				inputs.granularity,
+				inputs.backend,
+				inputs.default_precision,
+				inputs.default_reuse_factor,
+			)
+
+			hls_model = convert_from_keras_model(
+				inputs.model,
+				output_dir=inputs.output_dir,
+				project_name=inputs.project_name,
+				input_data_tb=inputs.input_data_tb,
+				output_data_tb=inputs.output_data_tb,
+				backend=inputs.backend,
+				board=inputs.board,
+				part=inputs.part,
+				clock_period=inputs.clock_period,
+				io_type=inputs.io_type,
+				hls_config=inputs.hls_config,
+			)
+
+			return Okay(
+				(self.output_vars[0].bind(hls_model), self.output_vars[1].bind(config))
+			)
+
+		except Exception as e:
+			return Fail([e])

exseos/plugin/hls4ml/stage/EvalModel.py

@@ -0,0 +1,60 @@
+# ExSeOS-H Hardware ML Workflow Manager
+# Copyright (C) 2024  Alexis Maya-Isabelle Shuping
+
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+from exseos.types.Result import Okay, Fail
+from exseos.types.Variable import VariableSet, UnboundVariable
+from exseos.workflow.stage.Stage import Stage
+
+import numpy as np
+from hls4ml.model import ModelGraph
+from sklearn.metrics import accuracy_score
+
+
+class EvalModel(Stage):
+	input_vars = (
+		UnboundVariable("model", ModelGraph, "The model to evaluate."),
+		UnboundVariable("X", np.ndarray, "Test data for evaluation"),
+		UnboundVariable("y", np.ndarray, "Labels for X"),
+	)
+	output_vars = (
+		UnboundVariable("accuracy", float, "HLS simulation model accuracy"),
+		UnboundVariable(
+			"predictions",
+			np.ndarray,
+			"Predictions generated by the HLS simulation model",
+		),
+	)
+
+	async def run(self, inputs: VariableSet, _):
+		stat = inputs.check_all()
+		if stat.is_fail:
+			return stat
+
+		try:
+			y_predictions = inputs.model.predict(np.ascontiguousarray(inputs.X))
+
+			accuracy = accuracy_score(
+				np.argmax(inputs.y, axis=1), np.argmax(y_predictions, axis=1)
+			)
+
+			return Okay(
+				(
+					self.output_vars[0].bind(accuracy),
+					self.output_vars[1].bind(y_predictions),
+				)
+			)
+		except Exception as e:
+			return Fail([e])

exseos/plugin/hls4ml/stage/ParseVivadoReport.py

@@ -0,0 +1,59 @@
+# ExSeOS-H Hardware ML Workflow Manager
+# Copyright (C) 2024  Alexis Maya-Isabelle Shuping
+
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+from exseos.types.Result import Okay, Fail
+from exseos.types.Variable import VariableSet, UnboundVariable
+from exseos.workflow.stage.Stage import Stage
+
+from hls4ml.model import ModelGraph
+from hls4ml.report.vivado_report import parse_vivado_report
+
+
+class ParseVivadoReport(Stage):
+	input_vars = (
+		UnboundVariable(
+			"location", ModelGraph, "Location of the project to parse files for."
+		),
+	)
+	output_vars = (
+		UnboundVariable("report", dict),
+		UnboundVariable("LUT", int),
+		UnboundVariable("FF", int),
+		UnboundVariable("WorstLatency", int),
+	)
+
+	async def run(self, inputs: VariableSet, _):
+		stat = inputs.check_all()
+		if stat.is_fail:
+			return stat
+
+		try:
+			report = parse_vivado_report(inputs.location)
+
+			luts = report["CSynthesisReport"]["LUT"]
+			ffs = report["CSynthesisReport"]["FF"]
+			worst_latency = report["CSynthesisReport"]["WorstLatency"]
+
+			return Okay(
+				(
+					self.output_vars[0].bind(report),
+					self.output_vars[1].bind(luts),
+					self.output_vars[2].bind(ffs),
+					self.output_vars[3].bind(worst_latency),
+				)
+			)
+		except Exception as e:
+			return Fail([e])

exseos/plugin/hls4ml/stage/SynthModel.py

@@ -0,0 +1,43 @@
+# ExSeOS-H Hardware ML Workflow Manager
+# Copyright (C) 2024  Alexis Maya-Isabelle Shuping
+
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+from exseos.types.Result import Okay, Fail
+from exseos.types.Variable import VariableSet, UnboundVariable
+from exseos.workflow.stage.Stage import Stage
+
+from hls4ml.model import ModelGraph
+
+
+class SynthModel(Stage):
+	input_vars = (
+		UnboundVariable("model", ModelGraph, "The model to synthesize."),
+		UnboundVariable(
+			"backend_kwargs", dict, "Arguments for individual backends", default={}
+		),
+	)
+	output_vars = ()
+
+	async def run(self, inputs: VariableSet, _):
+		stat = inputs.check_all()
+		if stat.is_fail:
+			return stat
+
+		try:
+			inputs.model.build(**inputs.backend_kwargs)
+
+			return Okay(())
+		except Exception as e:
+			return Fail([e])

exseos/plugin/ml/stage/ApplyLabelEncoder.py

@@ -0,0 +1,71 @@
+# ExSeOS-H Hardware ML Workflow Manager
+# Copyright (C) 2024  Alexis Maya-Isabelle Shuping
+
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+from exseos.types.Result import Okay, Fail
+from exseos.types.Variable import VariableSet, UnboundVariable
+from exseos.workflow.stage.Stage import Stage
+
+import numpy as np
+from sklearn.preprocessing import LabelEncoder
+
+
+class ApplyLabelEncoder(Stage):
+	input_vars = (
+		UnboundVariable("y", np.array, "The labels to encode."),
+		UnboundVariable(
+			"encoder_to_use",
+			LabelEncoder,
+			(
+				"[Optional] A pre-trained encoder to use. If not provided, an "
+				+ "encoder will be trained on the input data."
+			),
+		),
+	)
+	output_vars = (
+		UnboundVariable("y_encoded", np.array, "y encoded by encoder_to_use"),
+		UnboundVariable(
+			"trained_encoder",
+			LabelEncoder,
+			(
+				"The trained encoder. If encoder_to_use was provided, then "
+				+ "this will be the same as encoder_to_use; otherwise, it will "
+				+ "be a new encoder trained on y."
+			),
+		),
+	)
+
+	async def run(self, inputs: VariableSet, _):
+		stat = inputs.check("y")
+		if stat.is_fail:
+			return stat
+
+		try:
+			if inputs.check("encoder_to_use").is_fail:
+				encoder: LabelEncoder = LabelEncoder()
+				encoder.fit(inputs.y)
+			else:
+				encoder: LabelEncoder = inputs.scaler_to_use
+
+			y_scaled = encoder.transform(inputs.y)
+
+			return Okay(
+				(
+					self.output_vars[0].bind(y_scaled), 
+					self.output_vars[1].bind(encoder)
+				)
+			)
+		except Exception as e:
+			return Fail([e])