Add local tool runner (f91cb167) · Commits · Galaxy / Tool Sources / Fusion / IPS Fastran GUI

src/ips_fastran_gui/app/models/ips_fastran.py

+7 −1

Original line number	Diff line number	Diff line
		@@ -8,6 +8,7 @@ from typing import Callable, List, Tuple
		from nova.galaxy import Connection, Dataset, Parameters, Tool
		from nova.galaxy.interfaces import BasicTool

		from ips_fastran_gui.app.models.local import LocalTool
		from ips_fastran_gui.app.models.main_model import MainModel, RunLocationOption
		from ips_fastran_gui.app.models.superfacility import SuperfacilityTool

		@@ -28,7 +29,7 @@ class IPSFastranTool(BasicTool):
		def prepare_tool(self) -> Tuple[Tool, Parameters]:
		match self.model.resource_params.run_location:
		case RunLocationOption.local:
		return None, None
		return self.prepare_local()
		case RunLocationOption.sf_perlmutter:
		return self.prepare_superfacility()
		case RunLocationOption.galaxy_perlmutter:
		@@ -65,6 +66,11 @@ class IPSFastranTool(BasicTool):

		return self.tool, tool_params

		def prepare_local(self) -> Tuple[Tool, Parameters]:
		self.tool = LocalTool(self.model)

		return self.tool, Parameters()

		def prepare_superfacility(self) -> Tuple[Tool, Parameters]:
		self.tool = SuperfacilityTool(self.model)

src/ips_fastran_gui/app/models/local.py

0 → 100644

+133 −0

Original line number	Diff line number	Diff line
		"""Tool class for running via a local ips.py build."""

		import os
		import subprocess
		from datetime import datetime as dt
		from shutil import copy
		from time import time
		from typing import Any, Dict, List, Optional

		from nova.galaxy.job import JobStatus, WorkState
		from nova.galaxy.outputs import DatasetCollection, Outputs

		from ips_fastran_gui.app.models.main_model import MainModel

		# Pull new access token 60 seconds before expiration
		REFRESH_BUFFER = 60

		# Check task status every 10 seconds
		STATUS_INTERVAL = 10


		class LocalTool:
		"""Tool class for running via a local ips.py build."""

		def __init__(self, model: MainModel) -> None:
		self.model = model

		self.state = JobStatus()
		self.process: Optional[subprocess.Popen] = None
		self.working_directory = ""

		self.stdout = ""
		self.stderr = ""
		self.last_status_check = 0
		self.last_stdout_check = 0
		self.last_stderr_check = 0

		def cancel(self) -> None:
		if not self.process:
		return

		self.state.state = WorkState.CANCELING

		self.process.terminate()
		self.process = None

		self.state.state = WorkState.CANCELED

		def copy_input_files(self, input_files: List[Dict[str, Any]]) -> None:
		for file in input_files:
		if "children" in file:
		self.copy_input_files(file["children"])
		else:
		new_path = os.path.join(self.working_directory, file["relative_path"])
		os.makedirs(os.path.dirname(new_path), exist_ok=True)
		copy(file["path"], new_path)

		def get_full_status(self) -> JobStatus:
		current_time = int(time())
		if current_time < self.last_status_check + STATUS_INTERVAL:
		return self.state
		self.last_status_check = current_time

		if not self.process:
		self.state.state = WorkState.NOT_STARTED
		return self.state

		exit_code = self.process.poll()
		if exit_code is None:
		self.state.state = WorkState.RUNNING
		elif exit_code == 0:
		self.state.state = WorkState.FINISHED
		self.process = None
		else:
		self.state.state = WorkState.ERROR

		return self.state

		def get_results(self) -> Outputs:
		outputs = Outputs()
		dataset_collection = DatasetCollection("outputs")

		outputs.add_output(dataset_collection)

		return outputs

		def get_stderr(self, args: Any, *kwargs: Any) -> str:
		current_time = int(time())
		if current_time < self.last_stderr_check + STATUS_INTERVAL:
		return ""

		if not self.process or self.state.state != WorkState.RUNNING:
		return self.stderr
		self.last_stderr_check = current_time

		_, new_content = self.process.communicate()
		return_value = new_content.removeprefix(self.stderr)

		self.stderr = new_content

		return return_value

		def get_stdout(self, args: Any, *kwargs: Any) -> str:
		current_time = int(time())
		if current_time < self.last_stdout_check + STATUS_INTERVAL:
		return ""

		if not self.process or self.state.state != WorkState.RUNNING:
		return self.stdout
		self.last_stdout_check = current_time

		new_content, _ = self.process.communicate()
		return_value = new_content.removeprefix(self.stdout)

		self.stdout = new_content

		return return_value

		def run(self, args: Any, *kwargs: Any) -> None:
		self.working_directory = os.path.join(
		os.path.dirname(self.model.resource_params.executable), f"run-{dt.now().isoformat()}"
		)
		os.makedirs(self.working_directory, exist_ok=True)

		self.copy_input_files(self.model.config.input_files)

		self.process = subprocess.Popen(
		[self.model.resource_params.python_path, self.model.resource_params.executable],
		cwd=self.working_directory,
		stdout=subprocess.PIPE,
		stderr=subprocess.PIPE,
		text=True,
		)

src/ips_fastran_gui/app/models/main_model.py

+5 −0

Original line number	Diff line number	Diff line
		@@ -2,6 +2,7 @@

		import json
		import os
		import sys
		import zipfile
		from enum import Enum
		from io import BytesIO
		@@ -47,6 +48,10 @@ class ResourceParameters(BaseModel):

		# Local Run Options
		executable: str = Field(default="", title="IPS Fastran Executable (use full path)")
		python_path: str = Field(
		default=sys.executable,
		title="Python Executable (use full path)",
		)

		# Perlmutter Options
		partition: str = Field(default="debug", title="Partition")

src/ips_fastran_gui/app/views/tabs/resources_tab.py

+2 −1

Original line number	Diff line number	Diff line
		@@ -15,7 +15,8 @@ class ResourcesTab:
		with VBoxLayout():
		InputField(v_model="resource_params.run_location", type="select")

		with VBoxLayout(v_if="resource_params.run_location == 'Local Machine'"):
		with GridLayout(v_if="resource_params.run_location == 'Local Machine'", columns=2, gap="0.5em"):
		InputField(v_model="resource_params.python_path")
		InputField(v_model="resource_params.executable")

		with GridLayout(