Fix Argument list too long error (#84)

Author: Hevagog

src/services/simulation_service/core/connectors/common.py

@@ -11,7 +11,7 @@
 WellName: TypeAlias = str
 GridCell: TypeAlias = tuple[int, int, int]
 Point: TypeAlias = tuple[float, float, float]
-SerializedJson: TypeAlias = str
+JsonPath: TypeAlias = str
 
 
 class PerforationSchema(TypedDict):
@@ -76,5 +76,5 @@ class ConnectorInterface(ABC):
     @staticmethod
     @abstractmethod
     def run(
-        config: SerializedJson, stop: threading.Event | None = None
+        config: JsonPath, stop: threading.Event | None = None
     ) -> tuple[SimulationStatus, SimulationResults]: ...

src/services/simulation_service/core/connectors/open_darts.py

@@ -22,8 +22,8 @@
 from .common import (
     ConnectorInterface,
     GridCell,
+    JsonPath,
     Point,
-    SerializedJson,
     SimulationResults,
     SimulationResultType,
     SimulationStatus,
@@ -88,7 +88,7 @@ def open_darts_input_configuration_injector(func: Callable[..., None]) -> Any:
     The decorator should be used on top of the main function which trigger Open-Darts simulation
     Decorator allow to inject configuration json directly from the command line
 
-    Usage: python3 main.py (or other file name with open-darts simulation) 'configuration.json'
+    Usage: python3 main.py (or other file name with open-darts simulation) <path-to-configuration.json>
 
     example:
 
@@ -101,25 +101,41 @@ def main(configuration_content):
     @wraps(func)
     def wrapper(*args: Any, **kwargs: Any) -> None:
         if len(sys.argv) < 2:
-            logger.info("Usage: python main.py 'configuration.json (TBD)'")
+            logger.info("Usage: python main.py <path-to-configuration.json>")
             sys.exit(1)
             return
         json_config_str = sys.argv[1]
-        try:
-            config = json.loads(json_config_str)
-        except json.JSONDecodeError:
-            logger.error("Invalid JSON input.")
-            sys.exit(1)
-            return
-
-        if not isinstance(config, dict):
-            logger.error("Invalid JSON input.")
-            sys.exit(1)
-            return
-        if not all(isinstance(k, str) for k in config.keys()):
-            logger.error("Invalid JSON input.")
+        if os.path.isfile(json_config_str):
+            try:
+                with open(json_config_str, "r") as f:
+                    config = json.load(f)
+                if isinstance(config, str):
+                    config = json.loads(config)
+            except Exception:
+                logger.error("Failed to read configuration file.")
+                sys.exit(1)
+                return
+        else:
+            # For legacy reason, allowing passing json string directly
+            try:
+                config = json.loads(json_config_str)
+            except json.JSONDecodeError:
+                logger.error("Invalid JSON input.")
+                sys.exit(1)
+                return
+
+            except Exception:
+                pass
+
+        if not isinstance(config, (dict, list)):
+            logger.error(f"Invalid JSON input, got:{type(config).__name__}")
             sys.exit(1)
             return
+        if isinstance(config, dict):
+            if not all(isinstance(k, str) for k in config.keys()):
+                logger.error(f"Invalid JSON input:{config}")
+                sys.exit(1)
+                return
 
         func(config, *args, **kwargs)
 
@@ -135,7 +151,7 @@ class OpenDartsConnector(ConnectorInterface):
 
     @staticmethod
     def run(
-        config: SerializedJson,
+        config_path: JsonPath,
         stop: threading.Event | None = None,
     ) -> tuple[SimulationStatus, SimulationResults]:
         # Choose runner implementation via environment. Default uses subprocess runner.
@@ -150,9 +166,9 @@ def run(
 
         if runner_mode == "thread":
             thread_runner = ThreadRunner(subprocess_runner)
-            return thread_runner.run(config, stop)
+            return thread_runner.run(config_path, stop)
 
-        return subprocess_runner.run(config, stop)
+        return subprocess_runner.run(config_path, stop)
 
     @staticmethod
     def _get_broadcast_results(

src/services/simulation_service/core/connectors/runner.py

@@ -14,7 +14,7 @@
 from logger import get_logger, stream_reader
 
 from .common import (
-    SerializedJson,
+    JsonPath,
     SimulationResults,
     SimulationResultType,
     SimulationStatus,
@@ -26,7 +26,7 @@
 
 class SimulationRunner(Protocol):
     def run(
-        self, config: SerializedJson, stop: threading.Event | None = None
+        self, config: JsonPath, stop: threading.Event | None = None
     ) -> Tuple[SimulationStatus, SimulationResults]: ...
 
 
@@ -47,7 +47,7 @@ def __init__(
         self._timeout_duration = get_timeout_value()
 
     def run(
-        self, config: SerializedJson, stop: threading.Event | None = None
+        self, config: JsonPath, stop: threading.Event | None = None
     ) -> Tuple[SimulationStatus, SimulationResults]:
         managed_factory = self._managed_subprocess_factory
         if managed_factory is None:
@@ -222,7 +222,7 @@ def __init__(self, subprocess_runner: SubprocessRunner | None = None):
         self._subprocess_runner = subprocess_runner or SubprocessRunner()
 
     def run(
-        self, config: SerializedJson, stop: threading.Event | None = None
+        self, config: JsonPath, stop: threading.Event | None = None
     ) -> Tuple[SimulationStatus, SimulationResults]:
         os.environ.setdefault("OPEN_DARTS_THREAD_MODE", "1")
         return self._subprocess_runner.run(config, stop)

src/services/simulation_service/core/infrastructure/worker/src/_simulation_worker_grpc_docker.py

@@ -1,6 +1,8 @@
 import asyncio
 import io
+import json
 import os
+import tempfile
 import uuid
 from zipfile import BadZipFile, ZipFile
 
@@ -27,7 +29,10 @@
 
 def _run_simulator(simulation_job) -> tuple[SimulationStatus, SimulationResults]:
     connector = ConnectorFactory.get_connector(simulation_job.simulator)
-    return connector.run(simulation_job.simulation.input.wells)
+    with tempfile.NamedTemporaryFile(mode="w", delete=False, suffix=".json") as tf:
+        json.dump(simulation_job.simulation.input.wells, tf)
+        tf_path = tf.name
+    return connector.run(tf_path)
 
 
 async def request_simulation_job(stub, worker_id):

src/services/simulation_service/core/infrastructure/worker/src/_simulation_worker_grpc_thread.py

@@ -1,6 +1,8 @@
 import asyncio
 import io
+import json
 import os
+import tempfile
 import threading
 import uuid
 from contextlib import contextmanager
@@ -37,10 +39,14 @@
 
 
 def _run_simulator(
-    simulation_job, stop_flag: threading.Event | None = None
+    simulation_job,
+    stop_flag: threading.Event | None = None,
 ) -> Tuple[SimulationStatus, SimulationResults]:
     connector = ConnectorFactory.get_connector(simulation_job.simulator)
-    return connector.run(simulation_job.simulation.input.wells, stop=stop_flag)
+    with tempfile.NamedTemporaryFile(mode="w", delete=False, suffix=".json") as tf:
+        json.dump(simulation_job.simulation.input.wells, tf)
+        tf_path = tf.name
+    return connector.run(tf_path, stop=stop_flag)
 
 
 async def request_simulation_job(stub, worker_id):
@@ -64,7 +70,7 @@ async def submit_simulation_job(stub, simulation_job, simulation_result, status)
 
 
 async def handle_simulation_job(
-    stub, simulation_job, worker_id, stop_flag: threading.Event | None = None
+    stub, simulation_job, worker_id: str, stop_flag: threading.Event | None = None
 ):
     """
     Handle a single simulation job with proper error handling and logging.