text: #!/usr/bin/env python3
  
  # Copyright (c) Facebook, Inc. and its affiliates.
  # This source code is licensed under the MIT license found in the
  # LICENSE file in the root directory of this source tree.
  
  import gzip
  import json
  import os
  from typing import List, Optional
  
  from habitat.config import Config
  from habitat.core.dataset import ALL_SCENES_MASK, Dataset
  from habitat.core.registry import registry
  from habitat_extensions.exploration_task import ExplorationEpisode
  
  CONTENT_SCENES_PATH_FIELD = "content_scenes_path"
  DEFAULT_SCENE_PATH_PREFIX = "data/scene_datasets/"
  
  
  @registry.register_dataset(name="Exploration-v1")
  class ExplorationDatasetV1(Dataset):
      r"""Class inherited from Dataset that loads Point Navigation dataset.
      """
  
      episodes: List[ExplorationEpisode]
      content_scenes_path: str = "{data_path}/content/{scene}.json.gz"
  
      @staticmethod
      def check_config_paths_exist(config: Config) -> bool:
          return os.path.exists(
              config.DATA_PATH.format(split=config.SPLIT)
          ) and os.path.exists(config.SCENES_DIR)
  
      @classmethod
      def get_scenes_to_load(cls, config: Config) -> List[str]:
          r"""Return list of scene ids for which dataset has separate files with
          episodes.
          """
          assert cls.check_config_paths_exist(config)
          dataset_dir = os.path.dirname(config.DATA_PATH.format(split=config.SPLIT))
  
          cfg = config.clone()
          cfg.defrost()
          cfg.CONTENT_SCENES = []
          dataset = cls(cfg)
          has_individual_scene_files = os.path.exists(
              dataset.content_scenes_path.split("{scene}")[0].format(
                  data_path=dataset_dir
              )
          )
          if has_individual_scene_files:
              return cls._get_scenes_from_folder(
                  content_scenes_path=dataset.content_scenes_path,
                  dataset_dir=dataset_dir,
              )
          else:
              # Load the full dataset, things are not split into separate files
              cfg.CONTENT_SCENES = [ALL_SCENES_MASK]
              dataset = cls(cfg)
              return list(map(cls.scene_from_scene_path, dataset.scene_ids))
  
      @staticmethod
      def _get_scenes_from_folder(content_scenes_path, dataset_dir):
          scenes = []
          content_dir = content_scenes_path.split("{scene}")[0]
          scene_dataset_ext = content_scenes_path.split("{scene}")[1]
          content_dir = content_dir.format(data_path=dataset_dir)
          if not os.path.exists(content_dir):
              return scenes
  
          for filename in os.listdir(content_dir):
              if filename.endswith(scene_dataset_ext):
                  scene = filename[: -len(scene_dataset_ext)]
                  scenes.append(scene)
          scenes.sort()
          return scenes
  
      def __init__(self, config: Optional[Config] = None) -> None:
          self.episodes = []
  
          if config is None:
              return
  
          datasetfile_path = config.DATA_PATH.format(split=config.SPLIT)
          with gzip.open(datasetfile_path, "rt") as f:
              self.from_json(f.read(), scenes_dir=config.SCENES_DIR)
  
          # Read separate file for each scene
          dataset_dir = os.path.dirname(datasetfile_path)
          has_individual_scene_files = os.path.exists(
              self.content_scenes_path.split("{scene}")[0].format(data_path=dataset_dir)
          )
          if has_individual_scene_files:
              scenes = config.CONTENT_SCENES
              if ALL_SCENES_MASK in scenes:
                  scenes = self._get_scenes_from_folder(
                      content_scenes_path=self.content_scenes_path,
                      dataset_dir=dataset_dir,
                  )
  
              for scene in scenes:
                  scene_filename = self.content_scenes_path.format(
                      data_path=dataset_dir, scene=scene
                  )
                  with gzip.open(scene_filename, "rt") as f:
                      self.from_json(f.read(), scenes_dir=config.SCENES_DIR)
  
          else:
              self.episodes = list(
                  filter(self.build_content_scenes_filter(config), self.episodes)
              )
  
      def from_json(self, json_str: str, scenes_dir: Optional[str] = None) -> None:
          deserialized = json.loads(json_str)
          if CONTENT_SCENES_PATH_FIELD in deserialized:
              self.content_scenes_path = deserialized[CONTENT_SCENES_PATH_FIELD]
  
          for episode in deserialized["episodes"]:
              episode = ExplorationEpisode(**episode)
  
              if scenes_dir is not None:
                  if episode.scene_id.startswith(DEFAULT_SCENE_PATH_PREFIX):
                      episode.scene_id = episode.scene_id[
                          len(DEFAULT_SCENE_PATH_PREFIX) :
                      ]
  
                  episode.scene_id = os.path.join(scenes_dir, episode.scene_id)
  
              self.episodes.append(episode)
  