code: def connect_app(app_name_or_id: str):
      """Connect your local terminal to a running lightning app.
  
      After connecting, the lightning CLI will respond to commands exposed by the app.
  
      Example:
  
      \b
      # connect to an app named pizza-cooker-123
      lightning connect pizza-cooker-123
      \b
      # this will now show the commands exposed by pizza-cooker-123
      lightning --help
      \b
      # while connected, you can run the cook-pizza command exposed
      # by pizza-cooker-123.BTW, this should arguably generate an exception :-)
      lightning cook-pizza --flavor pineapple
      \b
      # once done, disconnect and go back to the standard lightning CLI commands
      lightning disconnect
  
      """
      from lightning.app.utilities.commands.base import (
          _download_command,
      )
  
      _clean_lightning_connection()
  
      if not os.path.exists(_LIGHTNING_CONNECTION_FOLDER):
          os.makedirs(_LIGHTNING_CONNECTION_FOLDER)
  
      connected_file = os.path.join(
          _LIGHTNING_CONNECTION_FOLDER, "connect.txt"
      )
  
      matched_connection_path = _scan_lightning_connections(
          app_name_or_id
      )
  
      if os.path.exists(connected_file):
          with open(connected_file) as f:
              result = f.readlines()[0].replace("\n", "")
  
          if result == app_name_or_id:
              if app_name_or_id == "localhost":
                  click.echo(
                      "You are connected to the local Lightning App."
                  )
              else:
                  click.echo(
                      f"You are already connected to the cloud Lightning App: {app_name_or_id}."
                  )
          else:
              disconnect_app()
              connect_app(app_name_or_id)
  
      elif app_name_or_id.startswith("localhost"):
          with Progress() as progress_bar:
              connecting = progress_bar.add_task(
                  "[magenta]Setting things up for you...", total=1.0
              )
  
              if app_name_or_id != "localhost":
                  raise Exception(
                      "You need to pass localhost to connect to the local Lightning App."
                  )
  
              retriever = _LightningAppOpenAPIRetriever(None)
  
              if retriever.api_commands is None:
                  raise Exception(
                      f"Connection wasn't successful. Is your app {app_name_or_id} running?"
                  )
  
              increment = 1 / (1 + len(retriever.api_commands))
  
              progress_bar.update(connecting, advance=increment)
  
              commands_folder = os.path.join(
                  _LIGHTNING_CONNECTION_FOLDER, "commands"
              )
              if not os.path.exists(commands_folder):
                  os.makedirs(commands_folder)
  
              _write_commands_metadata(retriever.api_commands)
  
              with open(
                  os.path.join(commands_folder, "openapi.json"), "w"
              ) as f:
                  json.dump(retriever.openapi, f)
  
              _install_missing_requirements(retriever)
  
              for (
                  command_name,
                  metadata,
              ) in retriever.api_commands.items():
                  if "cls_path" in metadata:
                      target_file = os.path.join(
                          commands_folder,
                          f"{command_name.replace(' ', '_')}.py",
                      )
                      _download_command(
                          command_name,
                          metadata["cls_path"],
                          metadata["cls_name"],
                          None,
                          target_file=target_file,
                      )
                  else:
                      with open(
                          os.path.join(
                              commands_folder, f"{command_name}.txt"
                          ),
                          "w",
                      ) as f:
                          f.write(command_name)
  
                  progress_bar.update(connecting, advance=increment)
  
          with open(connected_file, "w") as f:
              f.write(app_name_or_id + "\n")
  
          click.echo(
              "The lightning CLI now responds to app commands. Use 'lightning --help' to see them."
          )
          click.echo(" ")
  
          Popen(
              f"LIGHTNING_CONNECT_PPID={_PPID} {sys.executable} -m lightning --help",
              shell=True,
              stdout=sys.stdout,
              stderr=sys.stderr,
          ).wait()
  
      elif matched_connection_path:
          matched_connected_file = os.path.join(
              matched_connection_path, "connect.txt"
          )
          matched_commands = os.path.join(
              matched_connection_path, "commands"
          )
          if os.path.isdir(matched_commands):
              commands = os.path.join(
                  _LIGHTNING_CONNECTION_FOLDER, "commands"
              )
              shutil.copytree(matched_commands, commands)
              shutil.copy(matched_connected_file, connected_file)
  
          click.echo(
              "The lightning CLI now responds to app commands. Use 'lightning --help' to see them."
          )
          click.echo(" ")
  
          Popen(
              f"LIGHTNING_CONNECT_PPID={_PPID} {sys.executable} -m lightning --help",
              shell=True,
              stdout=sys.stdout,
              stderr=sys.stderr,
          ).wait()
  
      else:
          with Progress() as progress_bar:
              connecting = progress_bar.add_task(
                  "[magenta]Setting things up for you...", total=1.0
              )
  
              retriever = _LightningAppOpenAPIRetriever(app_name_or_id)
  
              if not retriever.api_commands:
                  client = LightningClient(retry=False)
                  project = _get_project(client)
                  apps = client.lightningapp_instance_service_list_lightningapp_instances(
                      project_id=project.project_id
                  )
                  click.echo(
                      "We didn't find a matching App. Here are the available Apps that you can "
                      f"connect to {[_get_app_display_name(app) for app in apps.lightningapps]}."
                  )
                  return
  
              increment = 1 / (1 + len(retriever.api_commands))
  
              progress_bar.update(connecting, advance=increment)
  
              _install_missing_requirements(retriever)
  
              commands_folder = os.path.join(
                  _LIGHTNING_CONNECTION_FOLDER, "commands"
              )
              if not os.path.exists(commands_folder):
                  os.makedirs(commands_folder)
  
              _write_commands_metadata(retriever.api_commands)
  
              for (
                  command_name,
                  metadata,
              ) in retriever.api_commands.items():
                  if "cls_path" in metadata:
                      target_file = os.path.join(
                          commands_folder, f"{command_name}.py"
                      )
                      _download_command(
                          command_name,
                          metadata["cls_path"],
                          metadata["cls_name"],
                          retriever.app_id,
                          target_file=target_file,
                      )
                  else:
                      with open(
                          os.path.join(
                              commands_folder, f"{command_name}.txt"
                          ),
                          "w",
                      ) as f:
                          f.write(command_name)
  
                  progress_bar.update(connecting, advance=increment)
  
          with open(connected_file, "w") as f:
              f.write(retriever.app_name + "\n")
              f.write(retriever.app_id + "\n")
  
          click.echo(
              "The lightning CLI now responds to app commands. Use 'lightning --help' to see them."
          )
          click.echo(" ")
  
          Popen(
              f"LIGHTNING_CONNECT_PPID={_PPID} {sys.executable} -m lightning --help",
              shell=True,
              stdout=sys.stdout,
              stderr=sys.stderr,
          ).wait()
  
imports: import json
  import os
  import shutil
  import sys
  from subprocess import Popen
  from typing import List, Optional, Tuple
  import click
  import psutil
  from lightning_utilities.core.imports import package_available
  from rich.progress import Progress
  from lightning.app.utilities.cli_helpers import _get_app_display_name, _LightningAppOpenAPIRetriever
  from lightning.app.utilities.cloud import _get_project
  from lightning.app.utilities.enum import OpenAPITags
  from lightning.app.utilities.log import get_logfile
  from lightning.app.utilities.network import LightningClient
label_desc: Use image node_id 1 for calling a global function with example usage: connect_app(app_name_or_id) without return types