Generic launch_task_for_single_config_asset script (#428)

Author: james-isbister

obi_one/__init__.py

@@ -6,6 +6,7 @@
 from obi_one.core.path import NamedPath
 from obi_one.core.run_tasks import (
     run_task_for_single_config,
+    run_task_for_single_config_asset,
     run_task_for_single_configs,
     run_tasks_for_generated_scan,
 )
@@ -165,6 +166,7 @@
     "nbS1VPMInputs",
     "rCA1CA3Inputs",
     "run_task_for_single_config",
+    "run_task_for_single_config_asset",
     "run_task_for_single_configs",
     "run_tasks_for_generated_scan",
     "write_circuit_node_set_file",

obi_one/core/launch_task_for_single_config_asset.py

@@ -0,0 +1,62 @@
+import argparse
+import logging
+import os
+
+import entitysdk
+from entitysdk import Client, ProjectContext
+
+from obi_one.core.run_tasks import run_task_for_single_config_asset
+
+L = logging.getLogger(__name__)
+
+
+def main() -> None:
+    """Script to launch a task for a single configuration asset.
+
+    Example usage.
+
+    python launch_task_for_single_config_asset.py
+        --entity_type Simulation
+        --entity_id 1569a81e-b578-4c39-a3a9-f9a05f123db9
+        --config_asset_id c9edaedf-e5c0-4643-979c-47375f3160e0
+        --lab_id 123e4567-e89b-12d3-a456-426614174000
+        --project_id 987e6543-e21b-12d3-a456-426614174999
+
+    Environment Variables Required:
+        OBI_AUTHENTICATION_TOKEN: Your authentication token for the platform.
+    """
+    parser = argparse.ArgumentParser(
+        description="Script to launch a task for a single configuration asset."
+    )
+
+    parser.add_argument("--entity_type", required=True, help="EntitySDK Entity type as string")
+    parser.add_argument("--entity_id", required=True, help="Entity ID as string")
+    parser.add_argument("--config_asset_id", required=True, help="Configuration Asset ID as string")
+    parser.add_argument("--lab_id", required=True, help="Virtual Lab ID as string")
+    parser.add_argument("--project_id", required=True, help="Project ID as string.")
+
+    args = parser.parse_args()
+
+    entity_type_str = args.entity_type
+    entity_id = args.entity_id
+    config_asset_id = args.config_asset_id
+    lab_id = args.lab_id
+    project_id = args.project_id
+
+    entity_type = getattr(entitysdk.models, entity_type_str)
+
+    token = os.getenv("OBI_AUTHENTICATION_TOKEN")
+
+    project_context = ProjectContext(virtual_lab_id=lab_id, project_id=project_id)
+    db_client = Client(token_manager=token, project_context=project_context)
+
+    run_task_for_single_config_asset(
+        entity_type=entity_type,
+        entity_id=entity_id,
+        config_asset_id=config_asset_id,
+        db_client=db_client,
+    )
+
+
+if __name__ == "__main__":
+    main()

obi_one/core/run_tasks.py

@@ -1,6 +1,7 @@
 import entitysdk
 
 from obi_one.core.scan_generation import ScanGenerationTask
+from obi_one.core.serialization import deserialize_obi_object_from_json_data
 from obi_one.core.single import SingleConfigMixin
 from obi_one.scientific.unions.config_task_map import get_configs_task_type
 
@@ -35,3 +36,32 @@ def run_tasks_for_generated_scan(
     run_task_for_single_configs(
         scan_generation.single_configs, db_client=db_client, entity_cache=entity_cache
     )
+
+
+def run_task_for_single_config_asset(
+    entity_type: type[entitysdk.models.entity.Entity],
+    entity_id: str,
+    config_asset_id: str,
+    *,
+    db_client: entitysdk.client.Client = None,
+    entity_cache: bool = False,
+) -> None:
+    """Run the appropriate task for a single configuration stored as an asset.
+
+    Example usage:
+    obi.run_task_for_single_config_asset(entity_type=entitysdk.models.simulation.Simulation,
+                                     entity_id="1569a81e-b578-4c39-a3a9-f9a05f123db9",
+                                     config_asset_id="c9edaedf-e5c0-4643-979c-47375f3160e0",
+                                     db_client=db_client)
+
+    """
+    content = db_client.download_content(
+        entity_id=entity_id, entity_type=entity_type, asset_id=config_asset_id
+    ).decode(encoding="utf-8")
+
+    single_config = deserialize_obi_object_from_json_data(content)
+    run_task_for_single_config(
+        single_config,
+        db_client=db_client,
+        entity_cache=entity_cache,
+    )