Configure_Flow.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "0d3813a6",
   "metadata": {},
   "source": [
    "# Tomopy Reconstruction Demo for ALS at ALCF\n",
    "\n",
    "This notebook demonstrates how to perform tomography reconstructions using Tomopy for ALS data on Polaris at ALCF using [Globus Flows](https://www.globus.org/globus-flows-service).  In this example, Globus flows will launch the application on Polaris and then transfer results from the Eagle filesystem.\n",
    "\n",
    "This notebook can be run from anywhere, it only requires a local installation of Globus software (described below) and access to a Globus Compute Endpoint setup by the user on Polaris that has access to tomopy (also described below).\n",
    "\n",
    "This demo uses Globus Flows and Globus Compute.  Globus Flows is a reliable and secure platform for orchestrating and performing research data management and analysis tasks. A flow is often needed to manage data coming from instruments, e.g., image files can be moved from local storage attached to a microscope to a high-performance storage system where they may be accessed by all members of the research project.  Globus Compute is a remote executor for tasks expressed as python functions that are sent to remote machines following a fire-and-forget model.\n",
    "\n",
    "In this notebook we will first describe necessary setup tasks for the local environment and on Polaris; second, we will describe how to create and test a Globus Compute function that can remotely launch a tomopy task on Polaris compute nodes; and third, we will describe how to incorporate this function with a Globus Flow that coordinates the execution of the tomopy task with a data transfer step.\n",
    "\n",
    "More examples of creating and running Globus Flows can be found on Globus' [demo instance](https://jupyter.demo.globus.org/hub/).\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "3dd8e562",
   "metadata": {},
   "source": [
    "## Tomopy on Polaris\n",
    "\n",
    "Tomopy has been installed in a conda environment on Polaris at this path which is accessible to members of the IRIBeta allocation: `/eagle/IRIBeta/als/env/tomopy`."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b93d4cc1",
   "metadata": {},
   "source": [
    "## Local Setup"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "72d36389",
   "metadata": {},
   "source": [
    "This notebook can be run from anywhere.  The only requirement is a local environment, such as a conda environment, that has python 3.11 installed along with the Globus packages `globus_compute_sdk` and `globus_cli`.  If you have a local installation of conda you can set up an environment that can run this notebook with these steps:\n",
    "\n",
    "```bash\n",
    "conda create -n globus_env python==3.11\n",
    "conda activate globus_env\n",
    "pip install globus_compute_sdk globus_cli python-dotenv\n",
    "```\n",
    "\n",
    "Note that the tomopy environment on Polaris contains python 3.11. It is therefore necessary for this environment on your local machine to have a python version close to this version."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "60401bc9",
   "metadata": {},
   "source": [
    "## Create a Globus Compute Endpoint on Polaris"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ed5f27c7",
   "metadata": {},
   "source": [
    "The first step for a user to execute applications on Polaris through the Globus service is to create a Globus compute endpoint on Polaris.  This requires the user to do a one-time setup task to configure the endpoint.\n",
    "\n",
    "In a shell seperate from this notebook, log into Polaris.  Copy the file included with this notebook called `template_config.yaml` to the Polaris filesystem (doesn't matter where).  Inside `template_config.yaml` you should see options setting your project name (`IRIBeta`), the queue you will use (`debug`), and commands that activate a pre-made conda environment on Polaris that can run tomopy.\n",
    "\n",
    "In your shell on Polaris, execute the following commands:\n",
    "\n",
    "```bash\n",
    "module use /soft/modulefiles\n",
    "module load conda\n",
    "conda activate /eagle/IRIBeta/als/env/tomopy\n",
    "globus-compute-endpoint configure --endpoint-config template_config.yaml als_endpoint\n",
    "globus-compute-endpoint start als_endpoint\n",
    "globus-compute-endpoint list\n",
    "```\n",
    "This will create an endpoint and display its status.  Its status should be listed as `running`.  There will also be displayed a unique Endpoint ID in the form of a UUID.  Copy the file `.env.example` to `.env`. Then edit the file aast the endpoint id in Copy that ID and paste it below as the value for `GLOBUS_FLOW_ENDPOINT_ID`. `.env` is in `.gitignore` so won't be accidently pushed to github."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "2f4551b2",
   "metadata": {},
   "outputs": [],
   "source": [
    "# reload accompanying python code\n",
    "%load_ext autoreload\n",
    "%autoreload 2"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "cbac1fca",
   "metadata": {},
   "source": [
    "\n",
    "Your endpoint is now active as a daemon process running on the Polaris login node.  It is communicating with the Globus service and waiting for work.  If you ever want to stop the process you can run:\n",
    "```bash\n",
    "globus-compute-endpoint stop als_endpoint\n",
    "```\n",
    "Your process may need to be periodically restarted, for example after Polaris comes back from a maintance period.\n",
    "\n",
    "If you ever need to make changes to your endpoint configuration, you can find the settings file in `~/.globus_compute/als_endpoint/config.yaml`.  Edit this file and then restart the endpoint with `globus-compute-endpoint restart als_endpoint` to make the changes active.\n",
    "\n",
    "This endpoint will submit work to the `debug` queue since this demo is for learning purposes.  In production, ALS will be able to submit work to the [demand queue](https://docs.alcf.anl.gov/polaris/running-jobs/#queues) which will give immediate access to Polaris compute nodes."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "38640cb7",
   "metadata": {},
   "source": [
    "## Create a Function"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f22d0caf",
   "metadata": {},
   "source": [
    "We first need to create a python function that wraps around the application call.  We will call it `reconstruction_wrapper` that takes as an input the directory on the eagle file system where the input data are located, `rundir`, and the name of the input `parametersfile`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "0a08c0e5",
   "metadata": {},
   "outputs": [],
   "source": [
    "def reconstruction_wrapper(rundir, parametersfile=\"inputOneSliceOfEach.txt\"):\n",
    "    import os\n",
    "    import time\n",
    "    import subprocess\n",
    "\n",
    "    start = time.time()\n",
    "\n",
    "    # Move to directory where data are located\n",
    "    os.chdir(rundir)\n",
    "\n",
    "    # Run reconstruction.py\n",
    "    command = f\"python /eagle/IRIBeta/als/example/reconstruction.py {parametersfile}\"\n",
    "    res = subprocess.run(command.split(\" \"), stdout=subprocess.PIPE, stderr=subprocess.PIPE)\n",
    "\n",
    "    end = time.time()\n",
    "    \n",
    "    return f\"Reconstructed data specified in {parametersfile} in {end-start} seconds;\\n {res}\""
   ]
  },
  {
   "cell_type": "markdown",
   "id": "caae3a4e",
   "metadata": {},
   "source": [
    "## Authenticate Client and Test Function"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "764af861",
   "metadata": {},
   "source": [
    "We will now instantiate a Globus Compute client to test the function.  Globus will prompt the user for their credentials if running for the first time.  The user should have a Globus account through their ALCF account and should validate with their ALCF credentials."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "c27ef6c0-b404-441e-9b76-102bf35b2974\n"
     ]
    }
   ],
   "source": [
    "import os\n",
    "\n",
    "import dotenv\n",
    "from globus_compute_sdk import Client, Executor\n",
    "\n",
    "dotenv.load_dotenv()\n",
    "\n",
    "gc = Client()\n",
    "# polaris_endpoint_id = os.getenv(\"GLOBUS_FLOW_ENDPOINT_ID\")\n",
    "polaris_endpoint_id = \"b33b3f5f-eb14-4bbd-89ed-9333a6ece6b6\"\n",
    "gce = Executor(endpoint_id=polaris_endpoint_id)\n",
    "\n",
    "print(os.getenv(\"GLOBUS_CLIENT_ID\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "67262fdc",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Warning - dependency versions are mismatched between local SDK and remote worker on endpoint <unknown>: local SDK uses Python 3.11.9/Dill 0.3.6 but worker used 3.11.7/0.3.6\n",
      "(worker SDK version: 2.12.0; worker OS: Linux-5.14.21-150500.55.49-default-x86_64-with-glibc2.31)\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Reconstructed data specified in inputOneSliceOfEach.txt in 17.838907718658447 seconds;\n",
      " CompletedProcess(args=['python', '/eagle/IRIBeta/als/example/reconstruction.py', 'inputOneSliceOfEach.txt'], returncode=0, stdout=b\"Read user input:\\n{'filename': '20220923_154543_ethan_robin_SiCircuit.h5', 'sinoused': [-1, 1, 1]}\\nRead user input:\\n{'filename': '20220923_160531_ethan_robin_climbing-vine_x00y00.h5', 'sinoused': [-1, 1, 1]}\\nRead user input:\\n{'filename': '20220923_160531_ethan_robin_climbing-vine_x00y01.h5', 'sinoused': [-1, 1, 1]}\\nRead user input:\\n{'filename': '20220923_160531_ethan_robin_climbing-vine_x00y02.h5', 'sinoused': [-1, 1, 1]}\\nRead user input:\\n{'filename': '20220923_160531_ethan_robin_climbing-vine_x00y03.h5', 'sinoused': [-1, 1, 1]}\\nRead user input:\\n{'filename': '20220923_160531_ethan_robin_climbing-vine_x00y04.h5', 'sinoused': [-1, 1, 1]}\\nRead user input:\\n{'filename': '20220923_160531_ethan_robin_climbing-vine_x00y05.h5', 'sinoused': [-1, 1, 1]}\\nRead user input:\\n{'filename': '20221221_084752_twist-tie_pointing_up_towards_door.h5', 'sinoused': [-1, 1, 1]}\\nRead user input:\\n{'filename': '20221221_084931_twist-tie_pointing_up_towards_door.h5', 'sinoused': [-1, 1, 1]}\\nRead user input:\\n{'filename': '20221222_085501_looking_from_above_spiralUP_CounterClockwise_endPointAtDoor.h5', 'sinoused': [-1, 1, 1]}\\n\", stderr=b'')\n"
     ]
    }
   ],
   "source": [
    "future = gce.submit(reconstruction_wrapper, \"/eagle/IRIBeta/als/example\")\n",
    "print(future.result())"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f50d9cb5",
   "metadata": {},
   "source": []
  },
  {
   "cell_type": "markdown",
   "id": "ae9b20d4",
   "metadata": {},
   "source": [
    "## Register Function"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6fcefda2",
   "metadata": {},
   "source": [
    "Now that the function has been tested and works, register the function with the Globus service.  This will allow the user to call the function from within a flow."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "742fd4e8",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "9916e300-fb19-4033-be3b-9c6ab12e0432\n"
     ]
    }
   ],
   "source": [
    "reconstruction_func = gc.register_function(reconstruction_wrapper)\n",
    "\n",
    "print(reconstruction_func)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c34c26d8",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'Reconstructed data specified in inputOneSliceOfEach.txt in 17.712284088134766 seconds;\\n CompletedProcess(args=[\\'python\\', \\'/eagle/IRIBeta/als/example/reconstruction.py\\', \\'inputOneSliceOfEach.txt\\'], returncode=0, stdout=b\"Read user input:\\\\n{\\'filename\\': \\'20220923_154543_ethan_robin_SiCircuit.h5\\', \\'sinoused\\': [-1, 1, 1]}\\\\nRead user input:\\\\n{\\'filename\\': \\'20220923_160531_ethan_robin_climbing-vine_x00y00.h5\\', \\'sinoused\\': [-1, 1, 1]}\\\\nRead user input:\\\\n{\\'filename\\': \\'20220923_160531_ethan_robin_climbing-vine_x00y01.h5\\', \\'sinoused\\': [-1, 1, 1]}\\\\nRead user input:\\\\n{\\'filename\\': \\'20220923_160531_ethan_robin_climbing-vine_x00y02.h5\\', \\'sinoused\\': [-1, 1, 1]}\\\\nRead user input:\\\\n{\\'filename\\': \\'20220923_160531_ethan_robin_climbing-vine_x00y03.h5\\', \\'sinoused\\': [-1, 1, 1]}\\\\nRead user input:\\\\n{\\'filename\\': \\'20220923_160531_ethan_robin_climbing-vine_x00y04.h5\\', \\'sinoused\\': [-1, 1, 1]}\\\\nRead user input:\\\\n{\\'filename\\': \\'20220923_160531_ethan_robin_climbing-vine_x00y05.h5\\', \\'sinoused\\': [-1, 1, 1]}\\\\nRead user input:\\\\n{\\'filename\\': \\'20221221_084752_twist-tie_pointing_up_towards_door.h5\\', \\'sinoused\\': [-1, 1, 1]}\\\\nRead user input:\\\\n{\\'filename\\': \\'20221221_084931_twist-tie_pointing_up_towards_door.h5\\', \\'sinoused\\': [-1, 1, 1]}\\\\nRead user input:\\\\n{\\'filename\\': \\'20221222_085501_looking_from_above_spiralUP_CounterClockwise_endPointAtDoor.h5\\', \\'sinoused\\': [-1, 1, 1]}\\\\n\", stderr=b\\'\\')'"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "future = gce.submit_to_registered_function(args=[\"/eagle/IRIBeta/als/example\"], function_id=reconstruction_func)\n",
    "future.result()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "0a51665b",
   "metadata": {},
   "source": [
    "## Incorporate Function into a Flow"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "819502c8",
   "metadata": {},
   "source": []
  },
  {
   "cell_type": "markdown",
   "id": "90530096",
   "metadata": {},
   "source": [
    "Now we will incorporate the Tomopy function into an example flow to run Tomopy on Polaris in coordination with other tasks.\n",
    "\n",
    "This example simply includes two steps:\n",
    "1. Run Tomopy via Globus Compute\n",
    "2. Transfer results from the eagle file system to the home file system.\n",
    "\n",
    "This can easily be extended to include other steps to import data, perform postprocessing, or publish and catalog results."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c405e165",
   "metadata": {},
   "source": [
    "This is the flow definition for this two-step flow."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a972fafa",
   "metadata": {},
   "outputs": [],
   "source": [
    "flow_definition = {\n",
    "    \"Comment\": \"Run Reconstruction and transfer results\",\n",
    "    \"StartAt\": \"Reconstruction\",\n",
    "    \"States\": {\n",
    "        \"Reconstruction\": {\n",
    "            \"Comment\": \"Reconstruction with Tomopy\",\n",
    "            \"Type\": \"Action\",\n",
    "            \"ActionUrl\": \"https://compute.actions.globus.org/fxap\",\n",
    "            \"Parameters\": {\n",
    "                \"endpoint.$\": \"$.input.compute_endpoint_id\",\n",
    "                \"function.$\": \"$.input.compute_function_id\",\n",
    "                \"kwargs.$\": \"$.input.compute_function_kwargs\"\n",
    "            },\n",
    "            \"ResultPath\": \"$.ReconstructionOutput\",\n",
    "            \"WaitTime\": 3600,\n",
    "            \"Next\": \"Transfer_Out\"\n",
    "        },\n",
    "        \"Transfer_Out\": {\n",
    "            \"Comment\": \"Transfer files\",\n",
    "            \"Type\": \"Action\",\n",
    "            \"ActionUrl\": \"https://actions.automate.globus.org/transfer/transfer\",\n",
    "            \"Parameters\": {\n",
    "                \"source_endpoint_id.$\": \"$.input.source.id\",\n",
    "                \"destination_endpoint_id.$\": \"$.input.destination.id\",\n",
    "                \"transfer_items\": [\n",
    "                    {\n",
    "                        \"source_path.$\": \"$.input.source.path\",\n",
    "                        \"destination_path.$\": \"$.input.destination.path\",\n",
    "                        \"recursive.$\": \"$.input.recursive_tx\"\n",
    "                    }\n",
    "                ]\n",
    "            },\n",
    "            \"ResultPath\": \"$.TransferFiles\",\n",
    "            \"WaitTime\": 300,\n",
    "            \"End\": True\n",
    "        },\n",
    "    }\n",
    "}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "267fa4ec",
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "id": "1811d7a0",
   "metadata": {},
   "source": [
    "Next, we need to provide a flows client id to run the flow.  For now we will use the demo client id, but a project should create a client for their work."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "id": "0a3efe39",
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "import time\n",
    "import globus_sdk\n",
    "\n",
    "from utils import get_flows_client, get_specific_flow_client\n",
    "\n",
    "# Tutorial client ID\n",
    "# We recommend replacing this with your own client for any production use-cases\n",
    "# Create your own at developers.globus.org\n",
    "# CLIENT_ID = \"61338d24-54d5-408f-a10d-66c06b59f6d2\""
   ]
  },
  {
   "cell_type": "markdown",
   "id": "fbffcd67",
   "metadata": {},
   "source": [
    "Now get an instance of the flows client.  You will be asked to validate credentials with the Globus service.  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "id": "ee782e64",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "!!!!!!!!!\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Warning - dependency versions are mismatched between local SDK and remote worker on endpoint <unknown>: local SDK uses Python 3.11.9/Dill 0.3.6 but worker used 3.11.7/0.3.6\n",
      "(worker SDK version: 2.12.0; worker OS: Linux-5.14.21-150500.55.49-default-x86_64-with-glibc2.31)\n"
     ]
    },
    {
     "ename": "AuthAPIError",
     "evalue": "('POST', 'https://auth.globus.org/v2/oauth2/token', 'Basic', 401, 'UNAUTHORIZED', 'Basic auth failed', 'ec293f4b-7926-41e3-84be-ea88730c21b4')",
     "output_type": "error",
     "traceback": [
      "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[0;31mAuthAPIError\u001b[0m                              Traceback (most recent call last)",
      "Cell \u001b[0;32mIn[34], line 1\u001b[0m\n\u001b[0;32m----> 1\u001b[0m fc \u001b[38;5;241m=\u001b[39m \u001b[43mget_flows_client\u001b[49m\u001b[43m(\u001b[49m\u001b[43m)\u001b[49m\n",
      "File \u001b[0;32m~/dev/als_at_alcf_tomopy_compute/utils.py:63\u001b[0m, in \u001b[0;36mget_flows_client\u001b[0;34m()\u001b[0m\n\u001b[1;32m     58\u001b[0m \u001b[38;5;66;03m# resource_server = globus_sdk.FlowsClient.resource_server\u001b[39;00m\n\u001b[1;32m     59\u001b[0m all_scopes \u001b[38;5;241m=\u001b[39m [\n\u001b[1;32m     60\u001b[0m     globus_sdk\u001b[38;5;241m.\u001b[39mFlowsClient\u001b[38;5;241m.\u001b[39mscopes\u001b[38;5;241m.\u001b[39mmanage_flows,\n\u001b[1;32m     61\u001b[0m     globus_sdk\u001b[38;5;241m.\u001b[39mFlowsClient\u001b[38;5;241m.\u001b[39mscopes\u001b[38;5;241m.\u001b[39mrun_status,\n\u001b[1;32m     62\u001b[0m ]\n\u001b[0;32m---> 63\u001b[0m authorizer \u001b[38;5;241m=\u001b[39m \u001b[43mglobus_sdk\u001b[49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mClientCredentialsAuthorizer\u001b[49m\u001b[43m(\u001b[49m\n\u001b[1;32m     64\u001b[0m \u001b[43m    \u001b[49m\u001b[43mconfidential_client\u001b[49m\u001b[43m,\u001b[49m\n\u001b[1;32m     65\u001b[0m \u001b[43m    \u001b[49m\u001b[43mall_scopes\u001b[49m\u001b[43m)\u001b[49m\n\u001b[1;32m     66\u001b[0m \u001b[38;5;28;01mreturn\u001b[39;00m globus_sdk\u001b[38;5;241m.\u001b[39mFlowsClient(authorizer\u001b[38;5;241m=\u001b[39mauthorizer)\n",
      "File \u001b[0;32m~/miniconda3/envs/iribeta/lib/python3.11/site-packages/globus_sdk/authorizers/client_credentials.py:76\u001b[0m, in \u001b[0;36mClientCredentialsAuthorizer.__init__\u001b[0;34m(self, confidential_client, scopes, access_token, expires_at, on_refresh)\u001b[0m\n\u001b[1;32m     69\u001b[0m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39mscopes \u001b[38;5;241m=\u001b[39m MutableScope\u001b[38;5;241m.\u001b[39mscopes2str(scopes)\n\u001b[1;32m     71\u001b[0m log\u001b[38;5;241m.\u001b[39minfo(\n\u001b[1;32m     72\u001b[0m     \u001b[38;5;124m\"\u001b[39m\u001b[38;5;124mSetting up ClientCredentialsAuthorizer with confidential_client=\u001b[39m\u001b[38;5;124m\"\u001b[39m\n\u001b[1;32m     73\u001b[0m     \u001b[38;5;124mf\u001b[39m\u001b[38;5;124m\"\u001b[39m\u001b[38;5;124m[instance:\u001b[39m\u001b[38;5;132;01m{\u001b[39;00m\u001b[38;5;28mid\u001b[39m(confidential_client)\u001b[38;5;132;01m}\u001b[39;00m\u001b[38;5;124m] and scopes=\u001b[39m\u001b[38;5;132;01m{\u001b[39;00m\u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39mscopes\u001b[38;5;132;01m}\u001b[39;00m\u001b[38;5;124m\"\u001b[39m\n\u001b[1;32m     74\u001b[0m )\n\u001b[0;32m---> 76\u001b[0m \u001b[38;5;28;43msuper\u001b[39;49m\u001b[43m(\u001b[49m\u001b[43m)\u001b[49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[38;5;21;43m__init__\u001b[39;49m\u001b[43m(\u001b[49m\u001b[43maccess_token\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43mexpires_at\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43mon_refresh\u001b[49m\u001b[43m)\u001b[49m\n",
      "File \u001b[0;32m~/miniconda3/envs/iribeta/lib/python3.11/site-packages/globus_sdk/authorizers/renewing.py:87\u001b[0m, in \u001b[0;36mRenewingAuthorizer.__init__\u001b[0;34m(self, access_token, expires_at, on_refresh)\u001b[0m\n\u001b[1;32m     81\u001b[0m \u001b[38;5;66;03m# if data were unspecified, fetch a new access token\u001b[39;00m\n\u001b[1;32m     82\u001b[0m \u001b[38;5;28;01melse\u001b[39;00m:\n\u001b[1;32m     83\u001b[0m     log\u001b[38;5;241m.\u001b[39minfo(\n\u001b[1;32m     84\u001b[0m         \u001b[38;5;124m\"\u001b[39m\u001b[38;5;124mCreating RenewingAuthorizer without Access \u001b[39m\u001b[38;5;124m\"\u001b[39m\n\u001b[1;32m     85\u001b[0m         \u001b[38;5;124m\"\u001b[39m\u001b[38;5;124mToken. Fetching initial token now.\u001b[39m\u001b[38;5;124m\"\u001b[39m\n\u001b[1;32m     86\u001b[0m     )\n\u001b[0;32m---> 87\u001b[0m     \u001b[38;5;28;43mself\u001b[39;49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43m_get_new_access_token\u001b[49m\u001b[43m(\u001b[49m\u001b[43m)\u001b[49m\n",
      "File \u001b[0;32m~/miniconda3/envs/iribeta/lib/python3.11/site-packages/globus_sdk/authorizers/renewing.py:122\u001b[0m, in \u001b[0;36mRenewingAuthorizer._get_new_access_token\u001b[0;34m(self)\u001b[0m\n\u001b[1;32m    116\u001b[0m \u001b[38;5;250m\u001b[39m\u001b[38;5;124;03m\"\"\"\u001b[39;00m\n\u001b[1;32m    117\u001b[0m \u001b[38;5;124;03mGiven token data from _get_token_response and _extract_token_data,\u001b[39;00m\n\u001b[1;32m    118\u001b[0m \u001b[38;5;124;03mset the access token and expiration time, calculate the new token\u001b[39;00m\n\u001b[1;32m    119\u001b[0m \u001b[38;5;124;03mhash, and call on_refresh\u001b[39;00m\n\u001b[1;32m    120\u001b[0m \u001b[38;5;124;03m\"\"\"\u001b[39;00m\n\u001b[1;32m    121\u001b[0m \u001b[38;5;66;03m# get the first (and only) token\u001b[39;00m\n\u001b[0;32m--> 122\u001b[0m res \u001b[38;5;241m=\u001b[39m \u001b[38;5;28;43mself\u001b[39;49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43m_get_token_response\u001b[49m\u001b[43m(\u001b[49m\u001b[43m)\u001b[49m\n\u001b[1;32m    123\u001b[0m token_data \u001b[38;5;241m=\u001b[39m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_extract_token_data(res)\n\u001b[1;32m    125\u001b[0m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39mexpires_at \u001b[38;5;241m=\u001b[39m token_data[\u001b[38;5;124m\"\u001b[39m\u001b[38;5;124mexpires_at_seconds\u001b[39m\u001b[38;5;124m\"\u001b[39m]\n",
      "File \u001b[0;32m~/miniconda3/envs/iribeta/lib/python3.11/site-packages/globus_sdk/authorizers/client_credentials.py:82\u001b[0m, in \u001b[0;36mClientCredentialsAuthorizer._get_token_response\u001b[0;34m(self)\u001b[0m\n\u001b[1;32m     78\u001b[0m \u001b[38;5;28;01mdef\u001b[39;00m \u001b[38;5;21m_get_token_response\u001b[39m(\u001b[38;5;28mself\u001b[39m) \u001b[38;5;241m-\u001b[39m\u001b[38;5;241m>\u001b[39m OAuthTokenResponse:\n\u001b[1;32m     79\u001b[0m \u001b[38;5;250m    \u001b[39m\u001b[38;5;124;03m\"\"\"\u001b[39;00m\n\u001b[1;32m     80\u001b[0m \u001b[38;5;124;03m    Make a client credentials grant\u001b[39;00m\n\u001b[1;32m     81\u001b[0m \u001b[38;5;124;03m    \"\"\"\u001b[39;00m\n\u001b[0;32m---> 82\u001b[0m     \u001b[38;5;28;01mreturn\u001b[39;00m \u001b[38;5;28;43mself\u001b[39;49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mconfidential_client\u001b[49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43moauth2_client_credentials_tokens\u001b[49m\u001b[43m(\u001b[49m\n\u001b[1;32m     83\u001b[0m \u001b[43m        \u001b[49m\u001b[43mrequested_scopes\u001b[49m\u001b[38;5;241;43m=\u001b[39;49m\u001b[38;5;28;43mself\u001b[39;49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mscopes\u001b[49m\n\u001b[1;32m     84\u001b[0m \u001b[43m    \u001b[49m\u001b[43m)\u001b[49m\n",
      "File \u001b[0;32m~/miniconda3/envs/iribeta/lib/python3.11/site-packages/globus_sdk/services/auth/client/confidential_client.py:134\u001b[0m, in \u001b[0;36mConfidentialAppAuthClient.oauth2_client_credentials_tokens\u001b[0;34m(self, requested_scopes)\u001b[0m\n\u001b[1;32m    132\u001b[0m log\u001b[38;5;241m.\u001b[39minfo(\u001b[38;5;124m\"\u001b[39m\u001b[38;5;124mFetching token(s) using client credentials\u001b[39m\u001b[38;5;124m\"\u001b[39m)\n\u001b[1;32m    133\u001b[0m requested_scopes_string \u001b[38;5;241m=\u001b[39m stringify_requested_scopes(requested_scopes)\n\u001b[0;32m--> 134\u001b[0m \u001b[38;5;28;01mreturn\u001b[39;00m \u001b[38;5;28;43mself\u001b[39;49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43moauth2_token\u001b[49m\u001b[43m(\u001b[49m\n\u001b[1;32m    135\u001b[0m \u001b[43m    \u001b[49m\u001b[43m{\u001b[49m\u001b[38;5;124;43m\"\u001b[39;49m\u001b[38;5;124;43mgrant_type\u001b[39;49m\u001b[38;5;124;43m\"\u001b[39;49m\u001b[43m:\u001b[49m\u001b[43m \u001b[49m\u001b[38;5;124;43m\"\u001b[39;49m\u001b[38;5;124;43mclient_credentials\u001b[39;49m\u001b[38;5;124;43m\"\u001b[39;49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[38;5;124;43m\"\u001b[39;49m\u001b[38;5;124;43mscope\u001b[39;49m\u001b[38;5;124;43m\"\u001b[39;49m\u001b[43m:\u001b[49m\u001b[43m \u001b[49m\u001b[43mrequested_scopes_string\u001b[49m\u001b[43m}\u001b[49m\n\u001b[1;32m    136\u001b[0m \u001b[43m\u001b[49m\u001b[43m)\u001b[49m\n",
      "File \u001b[0;32m~/miniconda3/envs/iribeta/lib/python3.11/site-packages/globus_sdk/services/auth/client/base_login_client.py:396\u001b[0m, in \u001b[0;36mAuthLoginClient.oauth2_token\u001b[0;34m(self, form_data, body_params, response_class)\u001b[0m\n\u001b[1;32m    393\u001b[0m \u001b[38;5;28;01mif\u001b[39;00m body_params:\n\u001b[1;32m    394\u001b[0m     data\u001b[38;5;241m.\u001b[39mupdate(body_params)\n\u001b[1;32m    395\u001b[0m \u001b[38;5;28;01mreturn\u001b[39;00m response_class(\n\u001b[0;32m--> 396\u001b[0m     \u001b[38;5;28;43mself\u001b[39;49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mpost\u001b[49m\u001b[43m(\u001b[49m\n\u001b[1;32m    397\u001b[0m \u001b[43m        \u001b[49m\u001b[38;5;124;43m\"\u001b[39;49m\u001b[38;5;124;43m/v2/oauth2/token\u001b[39;49m\u001b[38;5;124;43m\"\u001b[39;49m\u001b[43m,\u001b[49m\n\u001b[1;32m    398\u001b[0m \u001b[43m        \u001b[49m\u001b[43mdata\u001b[49m\u001b[38;5;241;43m=\u001b[39;49m\u001b[43mdata\u001b[49m\u001b[43m,\u001b[49m\n\u001b[1;32m    399\u001b[0m \u001b[43m        \u001b[49m\u001b[43mencoding\u001b[49m\u001b[38;5;241;43m=\u001b[39;49m\u001b[38;5;124;43m\"\u001b[39;49m\u001b[38;5;124;43mform\u001b[39;49m\u001b[38;5;124;43m\"\u001b[39;49m\u001b[43m,\u001b[49m\n\u001b[1;32m    400\u001b[0m \u001b[43m    \u001b[49m\u001b[43m)\u001b[49m\n\u001b[1;32m    401\u001b[0m )\n",
      "File \u001b[0;32m~/miniconda3/envs/iribeta/lib/python3.11/site-packages/globus_sdk/client.py:156\u001b[0m, in \u001b[0;36mBaseClient.post\u001b[0;34m(self, path, query_params, data, headers, encoding)\u001b[0m\n\u001b[1;32m    150\u001b[0m \u001b[38;5;250m\u001b[39m\u001b[38;5;124;03m\"\"\"\u001b[39;00m\n\u001b[1;32m    151\u001b[0m \u001b[38;5;124;03mMake a POST request to the specified path.\u001b[39;00m\n\u001b[1;32m    152\u001b[0m \n\u001b[1;32m    153\u001b[0m \u001b[38;5;124;03mSee :py:meth:`~.BaseClient.request` for details on the various parameters.\u001b[39;00m\n\u001b[1;32m    154\u001b[0m \u001b[38;5;124;03m\"\"\"\u001b[39;00m\n\u001b[1;32m    155\u001b[0m log\u001b[38;5;241m.\u001b[39mdebug(\u001b[38;5;124mf\u001b[39m\u001b[38;5;124m\"\u001b[39m\u001b[38;5;124mPOST to \u001b[39m\u001b[38;5;132;01m{\u001b[39;00mpath\u001b[38;5;132;01m}\u001b[39;00m\u001b[38;5;124m with query_params \u001b[39m\u001b[38;5;132;01m{\u001b[39;00mquery_params\u001b[38;5;132;01m}\u001b[39;00m\u001b[38;5;124m\"\u001b[39m)\n\u001b[0;32m--> 156\u001b[0m \u001b[38;5;28;01mreturn\u001b[39;00m \u001b[38;5;28;43mself\u001b[39;49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mrequest\u001b[49m\u001b[43m(\u001b[49m\n\u001b[1;32m    157\u001b[0m \u001b[43m    \u001b[49m\u001b[38;5;124;43m\"\u001b[39;49m\u001b[38;5;124;43mPOST\u001b[39;49m\u001b[38;5;124;43m\"\u001b[39;49m\u001b[43m,\u001b[49m\n\u001b[1;32m    158\u001b[0m \u001b[43m    \u001b[49m\u001b[43mpath\u001b[49m\u001b[43m,\u001b[49m\n\u001b[1;32m    159\u001b[0m \u001b[43m    \u001b[49m\u001b[43mquery_params\u001b[49m\u001b[38;5;241;43m=\u001b[39;49m\u001b[43mquery_params\u001b[49m\u001b[43m,\u001b[49m\n\u001b[1;32m    160\u001b[0m \u001b[43m    \u001b[49m\u001b[43mdata\u001b[49m\u001b[38;5;241;43m=\u001b[39;49m\u001b[43mdata\u001b[49m\u001b[43m,\u001b[49m\n\u001b[1;32m    161\u001b[0m \u001b[43m    \u001b[49m\u001b[43mheaders\u001b[49m\u001b[38;5;241;43m=\u001b[39;49m\u001b[43mheaders\u001b[49m\u001b[43m,\u001b[49m\n\u001b[1;32m    162\u001b[0m \u001b[43m    \u001b[49m\u001b[43mencoding\u001b[49m\u001b[38;5;241;43m=\u001b[39;49m\u001b[43mencoding\u001b[49m\u001b[43m,\u001b[49m\n\u001b[1;32m    163\u001b[0m \u001b[43m\u001b[49m\u001b[43m)\u001b[49m\n",
      "File \u001b[0;32m~/miniconda3/envs/iribeta/lib/python3.11/site-packages/globus_sdk/client.py:296\u001b[0m, in \u001b[0;36mBaseClient.request\u001b[0;34m(self, method, path, query_params, data, headers, encoding, allow_redirects, stream)\u001b[0m\n\u001b[1;32m    293\u001b[0m     \u001b[38;5;28;01mreturn\u001b[39;00m GlobusHTTPResponse(r, \u001b[38;5;28mself\u001b[39m)\n\u001b[1;32m    295\u001b[0m log\u001b[38;5;241m.\u001b[39mdebug(\u001b[38;5;124mf\u001b[39m\u001b[38;5;124m\"\u001b[39m\u001b[38;5;124mrequest completed with (error) response code: \u001b[39m\u001b[38;5;132;01m{\u001b[39;00mr\u001b[38;5;241m.\u001b[39mstatus_code\u001b[38;5;132;01m}\u001b[39;00m\u001b[38;5;124m\"\u001b[39m)\n\u001b[0;32m--> 296\u001b[0m \u001b[38;5;28;01mraise\u001b[39;00m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39merror_class(r)\n",
      "\u001b[0;31mAuthAPIError\u001b[0m: ('POST', 'https://auth.globus.org/v2/oauth2/token', 'Basic', 401, 'UNAUTHORIZED', 'Basic auth failed', 'ec293f4b-7926-41e3-84be-ea88730c21b4')"
     ]
    }
   ],
   "source": [
    "fc = get_flows_client()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "680519c8",
   "metadata": {},
   "source": [
    "Next, create a flow.  You will again be asked to validate credentials with the globus service."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "ba0fb24e",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "{\n",
      "  \"id\": \"91ce59ba-cbf1-45e3-b4f3-c86aeaacda42\",\n",
      "  \"title\": \"Reconstruction flow\",\n",
      "  \"subtitle\": \"\",\n",
      "  \"description\": \"\",\n",
      "  \"definition\": {\n",
      "    \"StartAt\": \"Reconstruction\",\n",
      "    \"States\": {\n",
      "      \"Reconstruction\": {\n",
      "        \"Parameters\": {\n",
      "          \"endpoint.$\": \"$.input.compute_endpoint_id\",\n",
      "          \"function.$\": \"$.input.compute_function_id\",\n",
      "          \"kwargs.$\": \"$.input.compute_function_kwargs\"\n",
      "        },\n",
      "        \"Type\": \"Action\",\n",
      "        \"Comment\": \"Reconstruction with Tomopy\",\n",
      "        \"Next\": \"Transfer_Out\",\n",
      "        \"ActionUrl\": \"https://compute.actions.globus.org/fxap\",\n",
      "        \"ResultPath\": \"$.ReconstructionOutput\",\n",
      "        \"WaitTime\": 3600\n",
      "      },\n",
      "      \"Transfer_Out\": {\n",
      "        \"Parameters\": {\n",
      "          \"source_endpoint_id.$\": \"$.input.source.id\",\n",
      "          \"destination_endpoint_id.$\": \"$.input.destination.id\",\n",
      "          \"transfer_items\": [\n",
      "            {\n",
      "              \"source_path.$\": \"$.input.source.path\",\n",
      "              \"destination_path.$\": \"$.input.destination.path\",\n",
      "              \"recursive.$\": \"$.input.recursive_tx\"\n",
      "            }\n",
      "          ]\n",
      "        },\n",
      "        \"Type\": \"Action\",\n",
      "        \"Comment\": \"Transfer files\",\n",
      "        \"End\": true,\n",
      "        \"ActionUrl\": \"https://actions.automate.globus.org/transfer/transfer\",\n",
      "        \"ResultPath\": \"$.TransferFiles\",\n",
      "        \"WaitTime\": 300\n",
      "      }\n",
      "    },\n",
      "    \"Comment\": \"Run Reconstruction and transfer results\"\n",
      "  },\n",
      "  \"input_schema\": {},\n",
      "  \"flow_url\": \"https://flows.automate.globus.org/flows/91ce59ba-cbf1-45e3-b4f3-c86aeaacda42\",\n",
      "  \"action_url\": \"https://flows.automate.globus.org/flows/91ce59ba-cbf1-45e3-b4f3-c86aeaacda42\",\n",
      "  \"globus_auth_scope\": \"https://auth.globus.org/scopes/91ce59ba-cbf1-45e3-b4f3-c86aeaacda42/flow_91ce59ba_cbf1_45e3_b4f3_c86aeaacda42_user\",\n",
      "  \"globus_auth_scopes_by_RunAs\": null,\n",
      "  \"synchronous\": false,\n",
      "  \"log_supported\": true,\n",
      "  \"types\": [\n",
      "    \"Action\"\n",
      "  ],\n",
      "  \"api_version\": \"1.0\",\n",
      "  \"keywords\": [],\n",
      "  \"principal_urn\": \"urn:globus:auth:identity:91ce59ba-cbf1-45e3-b4f3-c86aeaacda42\",\n",
      "  \"globus_auth_username\": \"91ce59ba-cbf1-45e3-b4f3-c86aeaacda42@clients.auth.globus.org\",\n",
      "  \"created_at\": \"2024-05-03T21:31:14.282267+00:00\",\n",
      "  \"updated_at\": \"2024-05-03T21:31:14.282267+00:00\",\n",
      "  \"user_role\": \"flow_owner\",\n",
      "  \"flow_owner\": \"urn:globus:auth:identity:f722020f-3b13-4d09-8a17-b65ace182899\",\n",
      "  \"created_by\": \"urn:globus:auth:identity:f722020f-3b13-4d09-8a17-b65ace182899\",\n",
      "  \"flow_viewers\": [],\n",
      "  \"visible_to\": [],\n",
      "  \"flow_starters\": [],\n",
      "  \"runnable_by\": [],\n",
      "  \"flow_administrators\": [],\n",
      "  \"administered_by\": [],\n",
      "  \"subscription_id\": null,\n",
      "  \"dry_run\": false\n",
      "}\n",
      "Newly created flow with id:\n",
      "91ce59ba-cbf1-45e3-b4f3-c86aeaacda42\n",
      "and scope:\n",
      "https://auth.globus.org/scopes/91ce59ba-cbf1-45e3-b4f3-c86aeaacda42/flow_91ce59ba_cbf1_45e3_b4f3_c86aeaacda42_user\n"
     ]
    }
   ],
   "source": [
    "flow = fc.create_flow(definition=flow_definition, title=\"Reconstruction flow\", input_schema={})\n",
    "flow_id = flow['id']\n",
    "print(flow)\n",
    "flow_scope = flow['globus_auth_scope']\n",
    "print(f'Newly created flow with id:\\n{flow_id}\\nand scope:\\n{flow_scope}')"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b146f619",
   "metadata": {},
   "source": [
    "## Run the Flow\n",
    "\n",
    "Now we need to create a set of inputs for the flow, that will follow the json structure below.  The key elements that are needed are:\n",
    "1. The Globus compute endpoint id\n",
    "2. The Globus compute function id\n",
    "3. Inputs to the Globus compute function\n",
    "4. The endpoint and path from which to transfer data\n",
    "5. The endpoint and path to which to transfer data"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7ca578c2",
   "metadata": {},
   "source": [
    "The user should choose a destination path and endpoint that they have access to.  As a sample endpoint, below the ALCF /home space is used.  To use this endpoint, the user should set a path in their home space (but remove the leading `/home` so that it will appear like `/csimpson/als_example`)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "6056aeb2",
   "metadata": {},
   "outputs": [],
   "source": [
    "# lcfhome_transfer_endpoint_id = \"9032dd3a-e841-4687-a163-2720da731b5b\"\n",
    "nersc_transfer_endpoint_id = \"6bdc7956-fc0f-4ad2-989c-7aa5ee643a79\"\n",
    "destination_path_on_alcfhome = \"/global/cfs/cdirs/als/data_mover/iribeta/\" # Note that paths for transfers on the home endpoint should remove the leading /home"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5ea4510b",
   "metadata": {},
   "source": [
    "The example problem is setup in a directory on the Eagle filesystem at ALCF, so that endpoint and path is used as the source of the data.  All the endpoints and functions are added to a `flow_input`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "aee0f949",
   "metadata": {},
   "outputs": [],
   "source": [
    "eagle_transfer_endpoint_id = \"05d2c76a-e867-4f67-aa57-76edeb0beda0\"\n",
    "source_path_on_eagle = \"/IRIBeta/als/example\" # Note that paths for transfers on the eagle endpoint should remove the leading /eagle\n",
    "\n",
    "function_inputs = {\"rundir\": \"/eagle/IRIBeta/als/example\"}\n",
    "\n",
    "flow_input = {\n",
    "    \"input\": {\n",
    "      \"source\": {\n",
    "        \"id\": eagle_transfer_endpoint_id,\n",
    "        \"path\": source_path_on_eagle\n",
    "      },\n",
    "      \"destination\": {\n",
    "        \"id\": nersc_transfer_endpoint_id,\n",
    "        \"path\": destination_path_on_alcfhome\n",
    "      },\n",
    "      \"recursive_tx\": True,\n",
    "      \"compute_endpoint_id\": polaris_endpoint_id,\n",
    "      \"compute_function_id\": reconstruction_func,\n",
    "      \"compute_function_kwargs\": function_inputs\n",
    "    }\n",
    "}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "8dc12cc8",
   "metadata": {},
   "outputs": [],
   "source": [
    "# collection_ids should contain all the transfer endpoint ids involved in the flow\n",
    "collection_ids = [flow_input[\"input\"][\"source\"][\"id\"], flow_input[\"input\"][\"destination\"][\"id\"]]\n",
    "from utils import get_specific_flow_client\n",
    "run_client = get_specific_flow_client(\"91ce59ba-cbf1-45e3-b4f3-c86aeaacda42\", collection_ids=collection_ids)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "id": "2635cf91",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Flow action started with id: da9e769b-c686-4823-8646-eeaa7f7db6b8\n",
      "Monitor your flow here: https://app.globus.org/runs/da9e769b-c686-4823-8646-eeaa7f7db6b8\n"
     ]
    },
    {
     "ename": "NameError",
     "evalue": "name 'fc' is not defined",
     "output_type": "error",
     "traceback": [
      "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[0;31mNameError\u001b[0m                                 Traceback (most recent call last)",
      "Cell \u001b[0;32mIn[17], line 11\u001b[0m\n\u001b[1;32m      9\u001b[0m \u001b[38;5;28;01mwhile\u001b[39;00m flow_status \u001b[38;5;129;01min\u001b[39;00m [\u001b[38;5;124m'\u001b[39m\u001b[38;5;124mACTIVE\u001b[39m\u001b[38;5;124m'\u001b[39m, \u001b[38;5;124m'\u001b[39m\u001b[38;5;124mINACTIVE\u001b[39m\u001b[38;5;124m'\u001b[39m]:\n\u001b[1;32m     10\u001b[0m     time\u001b[38;5;241m.\u001b[39msleep(\u001b[38;5;241m10\u001b[39m)\n\u001b[0;32m---> 11\u001b[0m     flow_action \u001b[38;5;241m=\u001b[39m \u001b[43mfc\u001b[49m\u001b[38;5;241m.\u001b[39mget_run(flow_run_id)\n\u001b[1;32m     12\u001b[0m     flow_status \u001b[38;5;241m=\u001b[39m flow_action[\u001b[38;5;124m'\u001b[39m\u001b[38;5;124mstatus\u001b[39m\u001b[38;5;124m'\u001b[39m]\n\u001b[1;32m     13\u001b[0m     \u001b[38;5;28mprint\u001b[39m(\u001b[38;5;124mf\u001b[39m\u001b[38;5;124m'\u001b[39m\u001b[38;5;124mFlow status: \u001b[39m\u001b[38;5;132;01m{\u001b[39;00mflow_status\u001b[38;5;132;01m}\u001b[39;00m\u001b[38;5;124m'\u001b[39m)\n",
      "\u001b[0;31mNameError\u001b[0m: name 'fc' is not defined"
     ]
    }
   ],
   "source": [
    "flow_action = run_client.run_flow(flow_input, label=\"ALS run\", tags=[\"demo\", \"als\", \"tomopy\"])\n",
    "flow_run_id = flow_action['action_id']\n",
    "\n",
    "print(f'Flow action started with id: {flow_run_id}')\n",
    "\n",
    "print(f\"Monitor your flow here: https://app.globus.org/runs/{flow_run_id}\")\n",
    "import time\n",
    "flow_status = flow_action['status']\n",
    "while flow_status in ['ACTIVE', 'INACTIVE']:\n",
    "    time.sleep(10)\n",
    "    flow_action = fc.get_run(flow_run_id)\n",
    "    flow_status = flow_action['status']\n",
    "    print(f'Flow status: {flow_status}')\n",
    "print(f'Final status: {flow_status}')"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e1778c51",
   "metadata": {},
   "source": [
    "## Next Steps\n",
    "\n",
    "Additional steps to the flow can be included if needed (for example an initial transfer step before tomopy is run).  The flow can also be adapted to be executed through the Globus web UI by adding a schema.  More information on flows can be found [here](https://docs.globus.org/guides/tutorials/flow-automation/create-a-flow/).\n",
    "\n",
    "The Globus compute endpoint configuration can be adapted to the user's needs.  Multiple instances of tomopy can be run in parallel on each node by adapting the `config.yaml` file.\n",
    "\n",
    "The content in this notebook can also be adapted to be run in a python script or a bash script.  More information can be found in the documentation for the [globus python api](https://globus-sdk-python.readthedocs.io/en/stable/services/flows.html) or the documentation for the [cli api](https://docs.globus.org/cli/reference/#globus_flows_commands)."
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.11.9"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}