feat: able to load the specific version of the dataset

Signed-off-by: ktro2828 <kotaro.uetake@tier4.jp>

Author: ktro2828

docs/tutorials/cli/t4sanity.md

@@ -11,6 +11,7 @@ $ t4sanity -h
 ╰──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯
 ╭─ Options ────────────────────────────────────────────────────────────────────────────────────────────────────────────╮
 │ --version             -v         Show the application version and exit.                                              │
+│ --revision            -rv   TEXT Specify if you want to load the specific version. [default: None]                   │
 │ --include-warning     -iw        Indicates whether to report any warnings.                                           │
 │ --install-completion             Install completion for the current shell.                                           │
 │ --show-completion                Show completion for the current shell, to copy it or customize the installation.    │

docs/tutorials/initialize.md

@@ -71,6 +71,13 @@ Done loading in 0.061 seconds.
 ======
 ```
 
+Note that if you doesn't specify `revision` parameter in construction, it searches the latest version of the dataset.
+By specifying `revision=<VERSION>`, you can load the specific version of the dataset.
+
+```python
+>>> t4 = Tier4("data/tier4/", revision="2", verbose=True)
+```
+
 ## Access to Schema Fields
 
 ---

t4_devkit/cli/sanity.py

@@ -18,13 +18,18 @@
 )
 
 
-def _run_sanity_check(db_parent: str, *, include_warning: bool = False) -> list[DBException]:
+def _run_sanity_check(
+    db_parent: str,
+    *,
+    revision: str | None = None,
+    include_warning: bool = False,
+) -> list[DBException]:
     exceptions: list[DBException] = []
 
     db_dirs: list[Path] = Path(db_parent).glob("*")
 
     for db_root in tqdm(db_dirs, desc=">>>Sanity checking..."):
-        result = sanity_check(db_root, include_warning=include_warning)
+        result = sanity_check(db_root, revision=revision, include_warning=include_warning)
         if result:
             exceptions.append(result)
     return exceptions
@@ -41,11 +46,14 @@ def main(
         is_eager=True,
     ),
     db_parent: str = typer.Argument(..., help="Path to parent directory of the databases."),
+    revision: str | None = typer.Option(
+        None, "-rv", "--revision", help="Specify if you want to check the specific version."
+    ),
     include_warning: bool = typer.Option(
         False, "-iw", "--include-warning", help="Indicates whether to report any warnings."
     ),
 ) -> None:
-    exceptions = _run_sanity_check(db_parent, include_warning=include_warning)
+    exceptions = _run_sanity_check(db_parent, revision=revision, include_warning=include_warning)
 
     headers = ["DatasetID", "Version", "Message"]
     table = [[e.dataset_id, e.version, e.message] for e in exceptions]

t4_devkit/cli/visualize.py

@@ -20,6 +20,12 @@
 @cli.command("scene", help="Visualize a specific scene.")
 def scene(
     data_root: Annotated[str, typer.Argument(help="Root directory path to the dataset.")],
+    revision: Annotated[
+        str | None,
+        typer.Option(
+            ..., "-rv", "--revision", help="Specify if you want to load the specific version."
+        ),
+    ] = None,
     future: Annotated[
         float,
         typer.Option(
@@ -41,14 +47,20 @@ def scene(
 ) -> None:
     _create_dir(output)
 
-    t4 = Tier4(data_root, verbose=False)
+    t4 = Tier4(data_root, revision=revision, verbose=False)
     t4.render_scene(future_seconds=future, save_dir=output)
 
 
 @cli.command("instance", help="Visualize a particular instance in the corresponding scene.")
 def instance(
     data_root: Annotated[str, typer.Argument(help="Root directory path to the dataset.")],
     instance: Annotated[list[str], typer.Argument(help="Instance token(s).")],
+    revision: Annotated[
+        str | None,
+        typer.Option(
+            ..., "-rv", "--revision", help="Specify if you want to load the specific version."
+        ),
+    ] = None,
     future: Annotated[
         float,
         typer.Option(
@@ -70,13 +82,19 @@ def instance(
 ) -> None:
     _create_dir(output)
 
-    t4 = Tier4(data_root, verbose=False)
+    t4 = Tier4(data_root, revision=revision, verbose=False)
     t4.render_instance(instance_token=instance, future_seconds=future, save_dir=output)
 
 
 @cli.command("pointcloud", help="Visualize pointcloud in the corresponding scene.")
 def pointcloud(
     data_root: Annotated[str, typer.Argument(help="Root directory path to the dataset.")],
+    revision: Annotated[
+        str | None,
+        typer.Option(
+            ..., "-rv", "--revision", help="Specify if you want to load the specific version."
+        ),
+    ] = None,
     ignore_distortion: Annotated[
         bool,
         typer.Option(
@@ -98,7 +116,7 @@ def pointcloud(
 ) -> None:
     _create_dir(output)
 
-    t4 = Tier4(data_root, verbose=False)
+    t4 = Tier4(data_root, revision=revision, verbose=False)
     t4.render_pointcloud(ignore_distortion=ignore_distortion, save_dir=output)
 
 

t4_devkit/common/sanity.py

@@ -19,11 +19,18 @@ class DBException:
     message: str
 
 
-def sanity_check(db_root: str | Path, *, include_warning: bool = False) -> DBException | None:
+def sanity_check(
+    db_root: str | Path,
+    *,
+    revision: str | None = None,
+    include_warning: bool = False,
+) -> DBException | None:
     """Perform sanity check and report exception or warning encountered while loading the dataset.
 
     Args:
         db_root (str | Path): Path to root directory of the dataset.
+        revision (str | None, optional): Specific version of the dataset.
+            If None, search the latest one.
         include_warning (bool, optional): Indicates whether to report warnings.
 
     Returns:
@@ -37,7 +44,7 @@ def sanity_check(db_root: str | Path, *, include_warning: bool = False) -> DBExc
             warnings.filterwarnings("ignore")
 
         try:
-            _ = Tier4(data_root=db_root, verbose=False)
+            _ = Tier4(data_root=db_root, revision=revision, verbose=False)
             exception = None
         except Exception as e:
             metadata = load_metadata(db_root)

t4_devkit/tier4.py

@@ -46,45 +46,72 @@
 
 @define
 class DBMetadata:
+    """A dataclass to represent dataset metadata.
+
+    Attributes:
+        data_root (str): Root directory path.
+        dataset_id (str): Unique dataset ID.
+        version (str | None): Dataset version.
+    """
+
     data_root: str
     dataset_id: str
     version: str | None
 
 
-def load_metadata(db_root: str) -> DBMetadata:
+def load_metadata(db_root: str, revision: str | None = None) -> DBMetadata:
     """Load metadata of T4 dataset including root directory path, dataset ID, and version.
 
     Args:
         db_root (str): Path to root directory of database.
+        revision (str | None, optional): Specify version of the dataset.
+            If None, search the latest one.
 
     Returns:
         Metadata of T4 dataset.
     """
     db_root_path = Path(db_root)
+    dataset_id = db_root_path.name
 
     version_pattern = re.compile(r".*/\d+$")
     versions = [d.name for d in db_root_path.iterdir() if version_pattern.match(d.as_posix())]
 
-    if versions:
-        version = sorted(versions)[-1]
-        data_root = db_root_path.joinpath(version).as_posix()
+    if revision is None:
+        if versions:
+            version = sorted(versions)[-1]
+            data_root = db_root_path.joinpath(version).as_posix()
+        else:
+            version = None
+            data_root = db_root_path.as_posix()
     else:
-        version = None
-        data_root = db_root_path.as_posix()
+        if revision not in versions:
+            raise ValueError(f"The version: {revision} is not included in {dataset_id}")
+        version = revision
+        data_root = db_root_path.joinpath(version).as_posix()
+
+    if version is None:
+        warnings.warn(f"{dataset_id} does't contain any versions.", DeprecationWarning)
 
-    return DBMetadata(data_root=data_root, dataset_id=db_root_path.name, version=version)
+    return DBMetadata(data_root=data_root, dataset_id=dataset_id, version=version)
 
 
 class Tier4:
     """Database class for T4 dataset to help query and retrieve information from the database."""
 
     schema_dir: str = "annotation"
 
-    def __init__(self, data_root: str, verbose: bool = True) -> None:
+    def __init__(
+        self,
+        data_root: str,
+        revision: str | None = None,
+        verbose: bool = True,
+    ) -> None:
         """Load database and creates reverse indexes and shortcuts.
 
         Args:
             data_root (str): Path to the root directory of dataset.
+            revision (str | None, optional): You can specify any specific version if you want.
+                If None, search the latest one.
             verbose (bool, optional): Whether to display status during load.
 
         Examples:
@@ -115,7 +142,7 @@ def __init__(self, data_root: str, verbose: bool = True) -> None:
             ======
 
         """
-        self._metadata = load_metadata(data_root)
+        self._metadata = load_metadata(data_root, revision)
 
         if not osp.exists(self.data_root):
             raise FileNotFoundError(f"Database directory is not found: {self.data_root}")