code: def _get_actual_scheduler_key(key: str, scheduler: Scheduler) -> Key:
      key = escape.url_unescape(key, plus=False)
  
      if key in scheduler.tasks:
          return key  # Basic str key
  
      # Tuple, bytes, or int key
      # First try safely reverting str(Key)
      def lists_to_tuples(o: object) -> Any:
          if isinstance(o, list):
              return tuple(lists_to_tuples(i) for i in o)
          else:
              return o
  
      key2 = key.replace("(", "[").replace(")", "]").replace("'", '"')
      try:
          key2 = lists_to_tuples(json.loads(key2))
          if key2 in scheduler.tasks:
              return key2
      except json.JSONDecodeError:
          pass
  
      # Edge case of keys with string elements containing [ ] ( ) ' " or bytes
      for key3 in scheduler.tasks:
          if str(key3) == key:
              return key3
  
      raise KeyError(key)
  
imports: from __future__ import annotations
  import json
  import logging
  import os
  import os.path
  from datetime import datetime
  from typing import TYPE_CHECKING, Any
  from tlz import first, merge
  from tornado import escape
  from tornado.websocket import WebSocketHandler
  from dask.typing import Key
  from dask.utils import format_bytes, format_time
  from distributed.diagnostics.websocket import WebsocketPlugin
  from distributed.http.utils import RequestHandler, redirect
  from distributed.metrics import time
  from distributed.utils import log_errors
label_desc: Use image node_id 1 for calling a global function with example usage: _get_actual_scheduler_key(key, scheduler) and returns: key, tuple, o, key2, key3