Add HPC pipeline configuration extension

source/extensions/hpc_pipeline_config/cli.py

@@ -0,0 +1,43 @@
+"""
+* SPDX-FileCopyrightText: Copyright (c) 2024 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+* SPDX-License-Identifier: Apache-2.0
+*
+* Licensed under the Apache License, Version 2.0 (the "License");
+* you may not use this file except in compliance with the License.
+* You may obtain a copy of the License at
+*
+* https://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing, software
+* distributed under the License is distributed on an "AS IS" BASIS,
+* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+* See the License for the specific language governing permissions and
+* limitations under the License.
+"""
+
+import argparse
+import carb
+from .commands import HPCPipelineConfigCommands
+
+def main():
+    parser = argparse.ArgumentParser(description="HPC Pipeline Configuration CLI")
+    parser.add_argument("--config", type=str, help="Path to the configuration file")
+    parser.add_argument("--undo", action="store_true", help="Undo the last configuration command")
+    parser.add_argument("--redo", action="store_true", help="Redo the last undone configuration command")
+    args = parser.parse_args()
+
+    hpc_commands = HPCPipelineConfigCommands()
+
+    if args.config:
+        with open(args.config, "r") as config_file:
+            config_options = eval(config_file.read())
+            hpc_commands.configure_hpc_pipeline(config_options)
+
+    if args.undo:
+        hpc_commands.undo()
+
+    if args.redo:
+        hpc_commands.redo()
+
+if __name__ == "__main__":
+    main()

source/extensions/hpc_pipeline_config/commands.py

@@ -0,0 +1,73 @@
+"""
+* SPDX-FileCopyrightText: Copyright (c) 2024 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+* SPDX-License-Identifier: Apache-2.0
+*
+* Licensed under the Apache License, Version 2.0 (the "License");
+* you may not use this file except in compliance with the License.
+* You may obtain a copy of the License at
+*
+* https://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing, software
+* distributed under the License is distributed on an "AS IS" BASIS,
+* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+* See the License for the specific language governing permissions and
+* limitations under the License.
+"""
+
+import carb
+import omni.ext
+
+class HPCPipelineConfigCommands:
+    """Class to handle HPC pipeline configuration commands with undo support"""
+
+    def __init__(self):
+        self._hpc_config = {}
+        self._undo_stack = []
+        self._redo_stack = []
+
+    def configure_hpc_pipeline(self, config_options):
+        """
+        Configure the HPC pipeline with the provided options.
+
+        Args:
+            config_options (dict): A dictionary containing configuration options for the HPC pipeline.
+        """
+        self._hpc_config = config_options
+
+        for command, value in config_options.items():
+            self._execute_command(command, value)
+
+    def _execute_command(self, command, value):
+        """
+        Execute a configuration command and add it to the undo stack.
+
+        Args:
+            command (str): The configuration command to execute.
+            value (any): The value associated with the command.
+        """
+        # Execute the command (this is a placeholder, replace with actual implementation)
+        print(f"Executing command: {command} with value: {value}")
+
+        # Add the command to the undo stack
+        self._undo_stack.append((command, value))
+
+    def undo(self):
+        """
+        Undo the last configuration command.
+        """
+        if self._undo_stack:
+            command, value = self._undo_stack.pop()
+            # Implement undo logic (this is a placeholder, replace with actual implementation)
+            print(f"Undoing command: {command} with value: {value}")
+            self._redo_stack.append((command, value))
+
+    def redo(self):
+        """
+        Redo the last undone configuration command.
+        """
+        if self._redo_stack:
+            command, value = self._redo_stack.pop()
+            # Implement redo logic (this is a placeholder, replace with actual implementation)
+            print(f"Redoing command: {command} with value: {value}")
+            self._execute_command(command, value)

source/extensions/hpc_pipeline_config/extension.py

@@ -0,0 +1,80 @@
+"""
+* SPDX-FileCopyrightText: Copyright (c) 2024 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+* SPDX-License-Identifier: Apache-2.0
+*
+* Licensed under the Apache License, Version 2.0 (the "License");
+* you may not use this file except in compliance with the License.
+* You may obtain a copy of the License at
+*
+* https://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing, software
+* distributed under the License is distributed on an "AS IS" BASIS,
+* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+* See the License for the specific language governing permissions and
+* limitations under the License.
+"""
+
+import carb
+import omni.ext
+
+class HPCPipelineConfigExtension(omni.ext.IExt):
+    """Extension for configuring HPC pipeline options"""
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self._hpc_config = {}
+        self._undo_stack = []
+        self._redo_stack = []
+
+    def on_startup(self, ext_id):
+        carb.log_info("[hpc_pipeline_config] startup")
+
+    def on_shutdown(self):
+        carb.log_info("[hpc_pipeline_config] shutdown")
+
+    def configure_hpc_pipeline(self, config_options):
+        """
+        Configure the HPC pipeline with the provided options.
+
+        Args:
+            config_options (dict): A dictionary containing configuration options for the HPC pipeline.
+        """
+        self._hpc_config = config_options
+
+        for command, value in config_options.items():
+            self._execute_command(command, value)
+
+    def _execute_command(self, command, value):
+        """
+        Execute a configuration command and add it to the undo stack.
+
+        Args:
+            command (str): The configuration command to execute.
+            value (any): The value associated with the command.
+        """
+        # Execute the command (this is a placeholder, replace with actual implementation)
+        print(f"Executing command: {command} with value: {value}")
+
+        # Add the command to the undo stack
+        self._undo_stack.append((command, value))
+
+    def undo(self):
+        """
+        Undo the last configuration command.
+        """
+        if self._undo_stack:
+            command, value = self._undo_stack.pop()
+            # Implement undo logic (this is a placeholder, replace with actual implementation)
+            print(f"Undoing command: {command} with value: {value}")
+            self._redo_stack.append((command, value))
+
+    def redo(self):
+        """
+        Redo the last undone configuration command.
+        """
+        if self._redo_stack:
+            command, value = self._redo_stack.pop()
+            # Implement redo logic (this is a placeholder, replace with actual implementation)
+            print(f"Redoing command: {command} with value: {value}")
+            self._execute_command(command, value)

source/extensions/hpc_pipeline_config/tests/test_configuration.py

@@ -0,0 +1,79 @@
+"""
+* SPDX-FileCopyrightText: Copyright (c) 2024 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+* SPDX-License-Identifier: Apache-2.0
+*
+* Licensed under the Apache License, Version 2.0 (the "License");
+* you may not use this file except in compliance with the License.
+* You may obtain a copy of the License at
+*
+* https://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing, software
+* distributed under the License is distributed on an "AS IS" BASIS,
+* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+* See the License for the specific language governing permissions and
+* limitations under the License.
+"""
+
+import unittest
+from source.extensions.hpc_pipeline_config.commands import HPCPipelineConfigCommands
+
+class TestHPCPipelineConfig(unittest.TestCase):
+    def setUp(self):
+        self.hpc_commands = HPCPipelineConfigCommands()
+
+    def test_configure_hpc_pipeline(self):
+        config_options = {
+            "option1": "value1",
+            "option2": "value2"
+        }
+        self.hpc_commands.configure_hpc_pipeline(config_options)
+        self.assertEqual(self.hpc_commands._hpc_config, config_options)
+
+    def test_undo_redo(self):
+        config_options = {
+            "option1": "value1",
+            "option2": "value2"
+        }
+        self.hpc_commands.configure_hpc_pipeline(config_options)
+        self.hpc_commands.undo()
+        self.assertEqual(len(self.hpc_commands._undo_stack), 1)
+        self.assertEqual(len(self.hpc_commands._redo_stack), 1)
+        self.hpc_commands.redo()
+        self.assertEqual(len(self.hpc_commands._undo_stack), 1)
+        self.assertEqual(len(self.hpc_commands._redo_stack), 0)
+
+    def test_multiple_undo_redo(self):
+        config_options = {
+            "option1": "value1",
+            "option2": "value2"
+        }
+        self.hpc_commands.configure_hpc_pipeline(config_options)
+        self.hpc_commands.undo()
+        self.hpc_commands.undo()
+        self.assertEqual(len(self.hpc_commands._undo_stack), 0)
+        self.assertEqual(len(self.hpc_commands._redo_stack), 2)
+        self.hpc_commands.redo()
+        self.hpc_commands.redo()
+        self.assertEqual(len(self.hpc_commands._undo_stack), 2)
+        self.assertEqual(len(self.hpc_commands._redo_stack), 0)
+
+    def test_invalid_configuration(self):
+        config_options = {
+            "invalid_option": "value"
+        }
+        with self.assertRaises(ValueError):
+            self.hpc_commands.configure_hpc_pipeline(config_options)
+
+    def test_partial_configuration(self):
+        config_options = {
+            "option1": "value1"
+        }
+        self.hpc_commands.configure_hpc_pipeline(config_options)
+        self.assertEqual(self.hpc_commands._hpc_config, config_options)
+        self.hpc_commands.undo()
+        self.assertEqual(len(self.hpc_commands._undo_stack), 0)
+        self.assertEqual(len(self.hpc_commands._redo_stack), 1)
+
+if __name__ == "__main__":
+    unittest.main()

source/extensions/hpc_pipeline_config/tests/test_performance.py

@@ -0,0 +1,57 @@
+"""
+* SPDX-FileCopyrightText: Copyright (c) 2024 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+* SPDX-License-Identifier: Apache-2.0
+*
+* Licensed under the Apache License, Version 2.0 (the "License");
+* you may not use this file except in compliance with the License.
+* You may obtain a copy of the License at
+*
+* https://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing, software
+* distributed under the License is distributed on an "AS IS" BASIS,
+* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+* See the License for the specific language governing permissions and
+* limitations under the License.
+"""
+
+import unittest
+import time
+from source.extensions.hpc_pipeline_config.commands import HPCPipelineConfigCommands
+
+class TestHPCPipelinePerformance(unittest.TestCase):
+    def setUp(self):
+        self.hpc_commands = HPCPipelineConfigCommands()
+
+    def test_performance_configure_hpc_pipeline(self):
+        config_options = {
+            "option1": "value1",
+            "option2": "value2"
+        }
+        start_time = time.time()
+        self.hpc_commands.configure_hpc_pipeline(config_options)
+        end_time = time.time()
+        duration = end_time - start_time
+        self.assertLess(duration, 1, "Configuration took too long")
+
+    def test_performance_undo_redo(self):
+        config_options = {
+            "option1": "value1",
+            "option2": "value2"
+        }
+        self.hpc_commands.configure_hpc_pipeline(config_options)
+
+        start_time = time.time()
+        self.hpc_commands.undo()
+        end_time = time.time()
+        duration = end_time - start_time
+        self.assertLess(duration, 1, "Undo took too long")
+
+        start_time = time.time()
+        self.hpc_commands.redo()
+        end_time = time.time()
+        duration = end_time - start_time
+        self.assertLess(duration, 1, "Redo took too long")
+
+if __name__ == "__main__":
+    unittest.main()