Warning
This document is for an old release of Galaxy. You can alternatively view this page in the latest release if it exists or view the top of the latest release's documentation.
Source code for galaxy.tools.remote_tool_eval
import json
import os
import shutil
import tempfile
import traceback
from typing import (
Callable,
NamedTuple,
)
from galaxy.datatypes.registry import Registry
from galaxy.files import ConfiguredFileSources
from galaxy.job_execution.compute_environment import SharedComputeEnvironment
from galaxy.job_execution.setup import JobIO
from galaxy.metadata.set_metadata import (
get_metadata_params,
get_object_store,
validate_and_load_datatypes_config,
)
from galaxy.model import store
from galaxy.model.store import SessionlessContext
from galaxy.objectstore import BaseObjectStore
from galaxy.structured_app import MinimalToolApp
from galaxy.tools import (
create_tool_from_representation,
evaluation,
)
from galaxy.tools.data import (
from_dict,
ToolDataTableManager,
)
from galaxy.util.bunch import Bunch
from galaxy.util.dbkeys import GenomeBuilds
[docs]class ToolAppConfig(NamedTuple):
name: str
tool_data_path: str
galaxy_data_manager_data_path: str
nginx_upload_path: str
len_file_path: str
builds_file_path: str
root: str
is_admin_user: Callable
admin_users: list = []
[docs]class ToolApp(MinimalToolApp):
"""Dummy App that allows loading tools"""
name = "tool_app"
[docs] def __init__(
self,
sa_session: SessionlessContext,
tool_app_config: ToolAppConfig,
datatypes_registry: Registry,
object_store: BaseObjectStore,
tool_data_table_manager: ToolDataTableManager,
file_sources: ConfiguredFileSources,
):
self.model = Bunch(context=sa_session)
self.config = tool_app_config
self.datatypes_registry = datatypes_registry
self.object_store = object_store
self.genome_builds = GenomeBuilds(self)
self.tool_data_tables = tool_data_table_manager
self.file_sources = file_sources
self.biotools_metadata_source = None
self.security = None # type: ignore[assignment]
[docs]def main(TMPDIR, WORKING_DIRECTORY, IMPORT_STORE_DIRECTORY):
metadata_params = get_metadata_params(WORKING_DIRECTORY)
datatypes_config = metadata_params["datatypes_config"]
if not os.path.exists(datatypes_config):
datatypes_config = os.path.join(WORKING_DIRECTORY, "configs", datatypes_config)
datatypes_registry = validate_and_load_datatypes_config(datatypes_config)
object_store = get_object_store(WORKING_DIRECTORY)
import_store = store.imported_store_for_metadata(IMPORT_STORE_DIRECTORY)
# TODO: clean up random places from which we read files in the working directory
job_io = JobIO.from_json(os.path.join(IMPORT_STORE_DIRECTORY, "job_io.json"), sa_session=import_store.sa_session)
tool_app_config = ToolAppConfig(
name="tool_app",
tool_data_path=job_io.tool_data_path,
galaxy_data_manager_data_path=job_io.galaxy_data_manager_data_path,
nginx_upload_path=TMPDIR,
len_file_path=job_io.len_file_path,
builds_file_path=job_io.builds_file_path,
root=TMPDIR,
is_admin_user=lambda _: job_io.user_context.is_admin,
)
with open(os.path.join(IMPORT_STORE_DIRECTORY, "tool_data_tables.json")) as data_tables_json:
tdtm = from_dict(json.load(data_tables_json))
app = ToolApp(
sa_session=import_store.sa_session,
tool_app_config=tool_app_config,
datatypes_registry=datatypes_registry,
object_store=object_store,
tool_data_table_manager=tdtm,
file_sources=job_io.file_sources,
)
# TODO: could try to serialize just a minimal tool variant instead of the whole thing ?
tool = create_tool_from_representation(
app=app,
raw_tool_source=job_io.tool_source,
tool_dir=job_io.tool_dir,
tool_source_class=job_io.tool_source_class,
)
tool_evaluator = evaluation.RemoteToolEvaluator(
app=app, tool=tool, job=job_io.job, local_working_directory=WORKING_DIRECTORY
)
tool_evaluator.set_compute_environment(compute_environment=SharedComputeEnvironment(job_io=job_io, job=job_io.job))
with open(os.path.join(WORKING_DIRECTORY, "tool_script.sh"), "a") as out:
command_line, version_command_line, extra_filenames, environment_variables = tool_evaluator.build()
out.write(f'{version_command_line or ""}{command_line}')
if __name__ == "__main__":
TMPDIR = tempfile.mkdtemp()
WORKING_DIRECTORY = os.getcwd()
WORKING_PARENT = os.path.join(WORKING_DIRECTORY, os.path.pardir)
if not os.path.isdir("working") and os.path.isdir(os.path.join(WORKING_PARENT, "working")):
# We're probably in pulsar
WORKING_DIRECTORY = WORKING_PARENT
METADATA_DIRECTORY = os.path.join(WORKING_DIRECTORY, "metadata")
IMPORT_STORE_DIRECTORY = os.path.join(METADATA_DIRECTORY, "outputs_new")
EXPORT_STORE_DIRECTORY = os.path.join(METADATA_DIRECTORY, "outputs_populated")
try:
main(TMPDIR, WORKING_DIRECTORY, IMPORT_STORE_DIRECTORY)
except Exception:
os.makedirs(EXPORT_STORE_DIRECTORY, exist_ok=True)
with open(os.path.join(EXPORT_STORE_DIRECTORY, "traceback.txt"), "w") as out:
out.write(traceback.format_exc())
raise
finally:
shutil.rmtree(TMPDIR, ignore_errors=True)