Unverified Commit 8fa76ddc authored by peterd-NV's avatar peterd-NV Committed by GitHub

Updates Mimic test cases to pytest format (#550)

# Description

<!--
Thank you for your interest in sending a pull request. Please make sure
to check the contribution guidelines.

Link:
https://isaac-sim.github.io/IsaacLab/main/source/refs/contributing.html
-->

Updates the following tests to pytest format:

- test_pink_ik.py
- test_selection_strategy.py
- test_generate_dataset.py

Updates test_generate_dataset.py to check that the expected number of
annotations are successfully generated in the HDF5 during the annotation
phase. If not, then test returns failure. This ensures that physics is
behaving correctly and that the correct annotated demos are being
generated. Previously, if a physics issue is introduced, the test would
timeout instead of failing.

The annotate_demos.py script was updated to return the number of
successfully annotated demos in order to support the above test.

## Type of change

<!-- As you go through the list, delete the ones that are not
applicable. -->

- New feature (non-breaking change which adds functionality)

## Checklist

- [x] I have run the [`pre-commit` checks](https://pre-commit.com/) with
`./isaaclab.sh --format`
- [ ] I have made corresponding changes to the documentation
- [x] My changes generate no new warnings
- [x] I have added tests that prove my fix is effective or that my
feature works
- [ ] I have updated the changelog and the corresponding version in the
extension's `config/extension.toml` file
- [ ] I have added my name to the `CONTRIBUTORS.md` or my name already
exists there

<!--
As you go through the checklist above, you can mark something as done by
putting an x character in it

For example,
- [x] I have done this task
- [ ] I have not done this task
-->
parent a958ac56
...@@ -159,7 +159,7 @@ def main(): ...@@ -159,7 +159,7 @@ def main():
if episode_count == 0: if episode_count == 0:
print("No episodes found in the dataset.") print("No episodes found in the dataset.")
exit() return 0
# get output directory path and file name (without extension) from cli arguments # get output directory path and file name (without extension) from cli arguments
output_dir = os.path.dirname(args_cli.output_file) output_dir = os.path.dirname(args_cli.output_file)
...@@ -236,6 +236,7 @@ def main(): ...@@ -236,6 +236,7 @@ def main():
# simulate environment -- run everything in inference mode # simulate environment -- run everything in inference mode
exported_episode_count = 0 exported_episode_count = 0
processed_episode_count = 0 processed_episode_count = 0
successful_task_count = 0 # Counter for successful task completions
with contextlib.suppress(KeyboardInterrupt) and torch.inference_mode(): with contextlib.suppress(KeyboardInterrupt) and torch.inference_mode():
while simulation_app.is_running() and not simulation_app.is_exiting(): while simulation_app.is_running() and not simulation_app.is_exiting():
# Iterate over the episodes in the loaded dataset file # Iterate over the episodes in the loaded dataset file
...@@ -259,6 +260,7 @@ def main(): ...@@ -259,6 +260,7 @@ def main():
) )
env.recorder_manager.export_episodes() env.recorder_manager.export_episodes()
exported_episode_count += 1 exported_episode_count += 1
successful_task_count += 1 # Increment successful task counter
print("\tExported the annotated episode.") print("\tExported the annotated episode.")
else: else:
print("\tSkipped exporting the episode due to incomplete subtask annotations.") print("\tSkipped exporting the episode due to incomplete subtask annotations.")
...@@ -268,11 +270,16 @@ def main(): ...@@ -268,11 +270,16 @@ def main():
f"\nExported {exported_episode_count} (out of {processed_episode_count}) annotated" f"\nExported {exported_episode_count} (out of {processed_episode_count}) annotated"
f" episode{'s' if exported_episode_count > 1 else ''}." f" episode{'s' if exported_episode_count > 1 else ''}."
) )
print(
f"Successful task completions: {successful_task_count}"
) # This line is used by the dataset generation test case to check if the expected number of demos were annotated
print("Exiting the app.") print("Exiting the app.")
# Close environment after annotation is complete # Close environment after annotation is complete
env.close() env.close()
return successful_task_count
def replay_episode( def replay_episode(
env: ManagerBasedRLMimicEnv, env: ManagerBasedRLMimicEnv,
...@@ -440,6 +447,8 @@ def annotate_episode_in_manual_mode( ...@@ -440,6 +447,8 @@ def annotate_episode_in_manual_mode(
if __name__ == "__main__": if __name__ == "__main__":
# run the main function # run the main function
main() successful_task_count = main()
# close sim app # close sim app
simulation_app.close() simulation_app.close()
# exit with the number of successful task completions as return code
exit(successful_task_count)
[package] [package]
# Note: Semantic Versioning is used: https://semver.org/ # Note: Semantic Versioning is used: https://semver.org/
version = "0.42.24" version = "0.42.25"
# Description # Description
title = "Isaac Lab framework for Robot Learning" title = "Isaac Lab framework for Robot Learning"
......
Changelog Changelog
--------- ---------
0.42.25 (2025-07-17)
~~~~~~~~~~~~~~~~~~~~
Changed
^^^^^^^
* Updated test_pink_ik.py test case to pytest format.
0.42.24 (2025-06-25) 0.42.24 (2025-06-25)
~~~~~~~~~~~~~~~~~~~~ ~~~~~~~~~~~~~~~~~~~~
......
[package] [package]
# Semantic Versioning is used: https://semver.org/ # Semantic Versioning is used: https://semver.org/
version = "1.0.10" version = "1.0.11"
# Description # Description
category = "isaaclab" category = "isaaclab"
......
Changelog Changelog
--------- ---------
1.0.11 (2025-07-17)
~~~~~~~~~~~~~~~~~~
Changed
^^^^^^^
* Updated test_selection_strategy.py and test_generate_dataset.py test cases to pytest format.
* Updated annotate_demos.py script to return the number of successful task completions as the exit code to support check in test_generate_dataset.py test case.
1.0.10 (2025-07-08) 1.0.10 (2025-07-08)
~~~~~~~~~~~~~~~~~~ ~~~~~~~~~~~~~~~~~~
......
...@@ -13,110 +13,129 @@ simulation_app = AppLauncher(headless=True).app ...@@ -13,110 +13,129 @@ simulation_app = AppLauncher(headless=True).app
import os import os
import subprocess import subprocess
import tempfile import tempfile
import unittest
import pytest
from isaaclab.utils.assets import ISAACLAB_NUCLEUS_DIR, retrieve_file_path from isaaclab.utils.assets import ISAACLAB_NUCLEUS_DIR, retrieve_file_path
DATASETS_DOWNLOAD_DIR = tempfile.mkdtemp(suffix="_Isaac-Stack-Cube-Franka-IK-Rel-Mimic-v0") DATASETS_DOWNLOAD_DIR = tempfile.mkdtemp(suffix="_Isaac-Stack-Cube-Franka-IK-Rel-Mimic-v0")
NUCLEUS_DATASET_PATH = os.path.join(ISAACLAB_NUCLEUS_DIR, "Tests", "Mimic", "dataset.hdf5") NUCLEUS_DATASET_PATH = os.path.join(ISAACLAB_NUCLEUS_DIR, "Tests", "Mimic", "dataset.hdf5")
EXPECTED_SUCCESSFUL_ANNOTATIONS = 10
class TestGenerateDataset(unittest.TestCase):
"""Test the dataset generation behavior of the Isaac Lab Mimic workflow.""" @pytest.fixture
def setup_test_environment():
def setUp(self): """Set up the environment for testing."""
"""Set up the environment for testing.""" # Create the datasets directory if it does not exist
# Create the datasets directory if it does not exist if not os.path.exists(DATASETS_DOWNLOAD_DIR):
if not os.path.exists(DATASETS_DOWNLOAD_DIR): print("Creating directory : ", DATASETS_DOWNLOAD_DIR)
print("Creating directory : ", DATASETS_DOWNLOAD_DIR) os.makedirs(DATASETS_DOWNLOAD_DIR)
os.makedirs(DATASETS_DOWNLOAD_DIR)
# Try to download the dataset from Nucleus # Try to download the dataset from Nucleus
try: try:
retrieve_file_path(NUCLEUS_DATASET_PATH, DATASETS_DOWNLOAD_DIR) retrieve_file_path(NUCLEUS_DATASET_PATH, DATASETS_DOWNLOAD_DIR)
except Exception as e: except Exception as e:
print(e) print(e)
print("Could not download dataset from Nucleus") print("Could not download dataset from Nucleus")
self.fail( pytest.fail(
"The dataset required for this test is currently unavailable. Dataset path: " + NUCLEUS_DATASET_PATH "The dataset required for this test is currently unavailable. Dataset path: " + NUCLEUS_DATASET_PATH
) )
# Set the environment variable PYTHONUNBUFFERED to 1 to get all text outputs in result.stdout # Set the environment variable PYTHONUNBUFFERED to 1 to get all text outputs in result.stdout
self.pythonunbuffered_env_var_ = os.environ.get("PYTHONUNBUFFERED") pythonunbuffered_env_var_ = os.environ.get("PYTHONUNBUFFERED")
os.environ["PYTHONUNBUFFERED"] = "1" os.environ["PYTHONUNBUFFERED"] = "1"
# Automatically detect the workflow root (backtrack from current file location) # Automatically detect the workflow root (backtrack from current file location)
current_dir = os.path.dirname(os.path.abspath(__file__)) current_dir = os.path.dirname(os.path.abspath(__file__))
workflow_root = os.path.abspath(os.path.join(current_dir, "../../..")) workflow_root = os.path.abspath(os.path.join(current_dir, "../../.."))
# Run the command to generate core configs # Run the command to generate core configs
config_command = [ config_command = [
workflow_root + "/isaaclab.sh", workflow_root + "/isaaclab.sh",
"-p", "-p",
os.path.join(workflow_root, "scripts/imitation_learning/isaaclab_mimic/annotate_demos.py"), os.path.join(workflow_root, "scripts/imitation_learning/isaaclab_mimic/annotate_demos.py"),
"--task", "--task",
"Isaac-Stack-Cube-Franka-IK-Rel-Mimic-v0", "Isaac-Stack-Cube-Franka-IK-Rel-Mimic-v0",
"--input_file", "--input_file",
DATASETS_DOWNLOAD_DIR + "/dataset.hdf5", DATASETS_DOWNLOAD_DIR + "/dataset.hdf5",
"--output_file", "--output_file",
DATASETS_DOWNLOAD_DIR + "/annotated_dataset.hdf5", DATASETS_DOWNLOAD_DIR + "/annotated_dataset.hdf5",
"--auto", "--auto",
"--headless", "--headless",
] ]
print(config_command) print(config_command)
# Execute the command and capture the result # Execute the command and capture the result
result = subprocess.run(config_command, capture_output=True, text=True) result = subprocess.run(config_command, capture_output=True, text=True)
# Print the result for debugging purposes print(f"Annotate demos result: {result.returncode}\n\n\n\n\n\n\n\n\n\n\n\n")
print("Config generation result:")
print(result.stdout) # Print standard output from the command # Print the result for debugging purposes
print(result.stderr) # Print standard error from the command print("Config generation result:")
print(result.stdout) # Print standard output from the command
# Check if the config generation was successful print(result.stderr) # Print standard error from the command
self.assertEqual(result.returncode, 0, msg=result.stderr)
# Check if the config generation was successful
def tearDown(self): assert result.returncode == 0, result.stderr
"""Clean up after tests."""
if self.pythonunbuffered_env_var_: # Check that at least one task was completed successfully by parsing stdout
os.environ["PYTHONUNBUFFERED"] = self.pythonunbuffered_env_var_ # Look for the line that reports successful task completions
else: success_line = None
del os.environ["PYTHONUNBUFFERED"] for line in result.stdout.split("\n"):
if "Successful task completions:" in line:
def test_generate_dataset(self): success_line = line
"""Test the dataset generation script.""" break
# Automatically detect the workflow root (backtrack from current file location)
current_dir = os.path.dirname(os.path.abspath(__file__)) assert success_line is not None, "Could not find 'Successful task completions:' in output"
workflow_root = os.path.abspath(os.path.join(current_dir, "../../.."))
# Extract the number from the line
# Define the command to run the dataset generation script try:
command = [ successful_count = int(success_line.split(":")[-1].strip())
workflow_root + "/isaaclab.sh", assert (
"-p", successful_count == EXPECTED_SUCCESSFUL_ANNOTATIONS
os.path.join(workflow_root, "scripts/imitation_learning/isaaclab_mimic/generate_dataset.py"), ), f"Expected 10 successful annotations but got {successful_count}"
"--input_file", except (ValueError, IndexError) as e:
DATASETS_DOWNLOAD_DIR + "/annotated_dataset.hdf5", pytest.fail(f"Could not parse successful task count from line: '{success_line}'. Error: {e}")
"--output_file",
DATASETS_DOWNLOAD_DIR + "/generated_dataset.hdf5", # Yield the workflow root for use in tests
"--generation_num_trials", yield workflow_root
"1",
"--headless", # Cleanup: restore the original environment variable
] if pythonunbuffered_env_var_:
os.environ["PYTHONUNBUFFERED"] = pythonunbuffered_env_var_
# Call the script and capture output else:
result = subprocess.run(command, capture_output=True, text=True) del os.environ["PYTHONUNBUFFERED"]
# Print the result for debugging purposes
print("Dataset generation result:") def test_generate_dataset(setup_test_environment):
print(result.stdout) # Print standard output from the command """Test the dataset generation script."""
print(result.stderr) # Print standard error from the command workflow_root = setup_test_environment
# Check if the script executed successfully # Define the command to run the dataset generation script
self.assertEqual(result.returncode, 0, msg=result.stderr) command = [
workflow_root + "/isaaclab.sh",
# Check for specific output "-p",
expected_output = "successes/attempts. Exiting" os.path.join(workflow_root, "scripts/imitation_learning/isaaclab_mimic/generate_dataset.py"),
self.assertIn(expected_output, result.stdout) "--input_file",
DATASETS_DOWNLOAD_DIR + "/annotated_dataset.hdf5",
"--output_file",
if __name__ == "__main__": DATASETS_DOWNLOAD_DIR + "/generated_dataset.hdf5",
unittest.main() "--generation_num_trials",
"1",
"--headless",
]
# Call the script and capture output
result = subprocess.run(command, capture_output=True, text=True)
# Print the result for debugging purposes
print("Dataset generation result:")
print(result.stdout) # Print standard output from the command
print(result.stderr) # Print standard error from the command
# Check if the script executed successfully
assert result.returncode == 0, result.stderr
# Check for specific output
expected_output = "successes/attempts. Exiting"
assert expected_output in result.stdout
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment