code: def __init__(self, lock_path, save_path, provider_config):
      self.lock = RLock()
      os.makedirs(os.path.dirname(lock_path), exist_ok=True)
      self.file_lock = FileLock(lock_path)
      self.save_path = save_path
  
      with self.lock:
          with self.file_lock:
              if os.path.exists(self.save_path):
                  workers = json.loads(open(self.save_path).read())
                  head_config = workers.get(provider_config["head_ip"])
                  if (
                      not head_config
                      or head_config.get("tags", {}).get(
                          TAG_RAY_NODE_KIND
                      )
                      != NODE_KIND_HEAD
                  ):
                      workers = {}
                      logger.info(
                          "Head IP changed - recreating cluster."
                      )
              else:
                  workers = {}
              logger.info(
                  "ClusterState: Loaded cluster state: {}".format(
                      list(workers)
                  )
              )
              for worker_ip in provider_config["worker_ips"]:
                  if worker_ip not in workers:
                      workers[worker_ip] = {
                          "tags": {TAG_RAY_NODE_KIND: NODE_KIND_WORKER},
                          "state": "terminated",
                      }
                  else:
                      assert (
                          workers[worker_ip]["tags"][TAG_RAY_NODE_KIND]
                          == NODE_KIND_WORKER
                      )
              if provider_config["head_ip"] not in workers:
                  workers[provider_config["head_ip"]] = {
                      "tags": {TAG_RAY_NODE_KIND: NODE_KIND_HEAD},
                      "state": "terminated",
                  }
              else:
                  assert (
                      workers[provider_config["head_ip"]]["tags"][
                          TAG_RAY_NODE_KIND
                      ]
                      == NODE_KIND_HEAD
                  )
              # Relevant when a user reduces the number of workers
              # without changing the headnode.
              list_of_node_ips = list(provider_config["worker_ips"])
              list_of_node_ips.append(provider_config["head_ip"])
              for worker_ip in list(workers):
                  if worker_ip not in list_of_node_ips:
                      del workers[worker_ip]
  
              # Set external head ip, if provided by user.
              # Necessary if calling `ray up` from outside the network.
              # Refer to LocalNodeProvider.external_ip function.
              external_head_ip = provider_config.get("external_head_ip")
              if external_head_ip:
                  head = workers[provider_config["head_ip"]]
                  head["external_ip"] = external_head_ip
  
              assert (
                  len(workers) == len(provider_config["worker_ips"]) + 1
              )
              with open(self.save_path, "w") as f:
                  logger.debug(
                      "ClusterState: Writing cluster state: {}".format(
                          workers
                      )
                  )
                  f.write(json.dumps(workers))
  
imports: import json
  import logging
  import os
  import socket
  from threading import RLock
  from filelock import FileLock
  from ray.autoscaler._private.local.config import LOCAL_CLUSTER_NODE_TYPE, bootstrap_local, get_lock_path, get_state_path
  from ray.autoscaler.node_provider import NodeProvider
  from ray.autoscaler.tags import NODE_KIND_HEAD, NODE_KIND_WORKER, STATUS_UP_TO_DATE, TAG_RAY_NODE_KIND, TAG_RAY_NODE_NAME, TAG_RAY_NODE_STATUS, TAG_RAY_USER_NODE_TYPE
label_desc: Use image node_id 1 to create a new ClusterState object  with example: obj = ClusterState(lock_path, save_path, provider_config)