feat: add strict option

ktro2828 · ktro2828 · commit c8a6a3fefc57 · 2025-11-12T15:26:17.000+09:00
Signed-off-by: ktro2828 &lt;kotaro.uetake@tier4.jp&gt;
diff --git a/docs/cli/t4sanity.md b/docs/cli/t4sanity.md
@@ -10,10 +10,11 @@ $ t4sanity -h
 ╰───────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯
 ╭─ Options ─────────────────────────────────────────────────────────────────────────────────────────────────────────────╮
 │ --version             -v             Show the application version and exit.                                           │
-│ --output              -o       TEXT  Path to output JSON file.                                                        │
-│ --revision            -rv      TEXT  Specify if you want to check the specific version.                               │
-│ --exclude             -e       TEXT  Exclude specific rules or rule groups.                                           │
+│ --output              -o       TEXT  Path to output JSON file. [default: None]                                        │
+│ --revision            -rv      TEXT  Specify if you want to check the specific version. [default: None]               │
+│ --exclude             -e       TEXT  Exclude specific rules or rule groups. [default: None]                           │
 │ --include-warning     -iw            Indicates whether to report any warnings.                                        │
+│ --strict              -s             Indicates whether warnings are treated as failures.                              │
 │ --install-completion                 Install completion for the current shell.                                        │
 │ --show-completion                    Show completion for the current shell, to copy it or customize the installation. │
 │ --help                -h             Show this message and exit.                                                      │
@@ -100,3 +101,14 @@ With `-e; --excludes` option enables us to exclude specific checks by specifying
 # Exclude STR001 and all FMT-relevant rules
 t4sanity <DATA_ROOT> -e STR001 -e FMT
 ```
+
+### Strict Mode
+
+Basically, rules whose **severity is WARNING** will be treated as success.
+
+With `-s; --strict` option enables us to treat warnings as failures:
+
+```shell
+# Run strict mode
+t4sanity <DATA_ROOT> -s
+```
diff --git a/t4_devkit/cli/sanity.py b/t4_devkit/cli/sanity.py
@@ -37,12 +37,16 @@ def main(
     include_warning: bool = typer.Option(
         False, "-iw", "--include-warning", help="Indicates whether to report any warnings."
     ),
+    strict: bool = typer.Option(
+        False, "-s", "--strict", help="Indicates whether warnings are treated as failures."
+    ),
 ) -> None:
     result = sanity_check(
         data_root=data_root,
         revision=revision,
         excludes=excludes,
         include_warning=include_warning,
+        strict=strict,
     )
 
     print_sanity_result(result)
diff --git a/t4_devkit/sanity/checker.py b/t4_devkit/sanity/checker.py
@@ -6,7 +6,7 @@
 
 from returns.maybe import Maybe, Nothing, Some
 
-from .result import make_failure, make_skipped, make_success
+from .result import make_report, make_skipped
 
 if TYPE_CHECKING:
     from .context import SanityContext
@@ -48,15 +48,22 @@ def __call__(self, context: SanityContext) -> Report:
                 return make_skipped(self.id, self.name, self.severity, self.description, skip)
 
         reasons = self.check(context)
-        if reasons:
-            return make_failure(self.id, self.name, self.severity, self.description, reasons)
-        else:
-            return make_success(self.id, self.name, self.severity, self.description)
+        return make_report(
+            self.id, self.name, self.severity, self.description, reasons, strict=context.strict
+        )
 
     def can_skip(self, _: SanityContext) -> Maybe[Reason]:
         """Return a skip reason if the checker should be skipped."""
         return Nothing
 
     @abstractmethod
-    def check(self, context: SanityContext) -> list[Reason]:
+    def check(self, context: SanityContext) -> list[Reason] | None:
+        """Return a list of reasons if the checker fails, or None if it passes.
+
+        Args:
+            context (SanityContext): The sanity context.
+
+        Returns:
+            A list of reasons if the checker fails, or None if it passes.
+        """
         pass
diff --git a/t4_devkit/sanity/context.py b/t4_devkit/sanity/context.py
@@ -15,12 +15,13 @@
 @define
 class SanityContext:
     metadata: Maybe[DBMetadata]
+    strict: bool
 
     @classmethod
-    def from_path(cls, data_root: str, revision: str | None = None) -> Self:
+    def from_path(cls, data_root: str, revision: str | None = None, strict: bool = False) -> Self:
         metadata_result = _load_metadata_safe(data_root, revision=revision)
         metadata = metadata_result.unwrap() if is_successful(metadata_result) else None
-        return cls(Maybe.from_optional(metadata))
+        return cls(Maybe.from_optional(metadata), strict)
 
     @property
     def data_root(self) -> Maybe[Path]:
diff --git a/t4_devkit/sanity/format/base.py b/t4_devkit/sanity/format/base.py
@@ -37,24 +37,24 @@ def can_skip(self, context: SanityContext) -> Maybe[Reason]:
             case _:
                 return Nothing
 
-    def check(self, context: SanityContext) -> list[Reason]:
+    def check(self, context: SanityContext) -> list[Reason] | None:
         filepath = context.to_schema_file(self.schema).unwrap()
 
         if self.schema.is_optional() and not filepath.exists():
-            return []
+            return None
 
         records = load_json_safe(filepath)
         return _build_records(self.schema, records.unwrap())
 
 
-def _build_records(schema: SchemaName, records: list[dict]) -> list[Reason]:
+def _build_records(schema: SchemaName, records: list[dict]) -> list[Reason] | None:
     module = SCHEMAS.get(schema)
     failures = []
     for record in records:
         conversion = _safe_from_dict(module, record)
         if not is_successful(conversion):
             failures.append(Reason(f"[{schema.name}] {record['token']}: {conversion.failure()}"))
-    return failures
+    return failures if failures else None
 
 
 @safe
diff --git a/t4_devkit/sanity/record/base.py b/t4_devkit/sanity/record/base.py
@@ -37,19 +37,19 @@ def can_skip(self, context: SanityContext) -> Maybe[Reason]:
             case _:
                 return Maybe.from_value(Reason("Missing 'annotation' directory path"))
 
-    def check(self, context: SanityContext) -> list[Reason]:
+    def check(self, context: SanityContext) -> list[Reason] | None:
         filepath = context.to_schema_file(self.schema).unwrap()
         records = load_json_safe(filepath).unwrap()
         return self.check_count(records)
 
     @abstractmethod
-    def check_count(self, records: list[dict]) -> list[Reason]:
+    def check_count(self, records: list[dict]) -> list[Reason] | None:
         """Check the count of records.
 
         Args:
             records (list[dict]): The list of records to check.
 
         Returns:
-            A list of reasons for any issues found, otherwise an empty list.
+            A list of reasons for any issues found, otherwise None.
         """
         pass
diff --git a/t4_devkit/sanity/record/rec001.py b/t4_devkit/sanity/record/rec001.py
@@ -19,10 +19,10 @@ class REC001(RecordCountChecker):
     description = "'Scene' record is a single."
     schema = SchemaName.SCENE
 
-    def check_count(self, records: list[dict]) -> list[Reason]:
+    def check_count(self, records: list[dict]) -> list[Reason] | None:
         num_scene = len(records)
         return (
-            []
+            None
             if num_scene == 1
             else [Reason(f"'Scene' must contain exactly one element: {num_scene}")]
         )
diff --git a/t4_devkit/sanity/record/rec002.py b/t4_devkit/sanity/record/rec002.py
@@ -19,6 +19,6 @@ class REC002(RecordCountChecker):
     description = "'Sample' record is not empty."
     schema = SchemaName.SAMPLE
 
-    def check_count(self, records: list[dict]) -> list[Reason]:
+    def check_count(self, records: list[dict]) -> list[Reason] | None:
         num_sample = len(records)
-        return [Reason("'Sample' record must not be empty")] if num_sample == 0 else []
+        return [Reason("'Sample' record must not be empty")] if num_sample == 0 else None
diff --git a/t4_devkit/sanity/record/rec003.py b/t4_devkit/sanity/record/rec003.py
@@ -19,6 +19,6 @@ class REC003(RecordCountChecker):
     description = "'SampleData' record is not empty."
     schema = SchemaName.SAMPLE_DATA
 
-    def check_count(self, records: list[dict]) -> list[Reason]:
+    def check_count(self, records: list[dict]) -> list[Reason] | None:
         num_sample_data = len(records)
-        return [Reason("'SampleData' record must not be empty")] if num_sample_data == 0 else []
+        return [Reason("'SampleData' record must not be empty")] if num_sample_data == 0 else None
diff --git a/t4_devkit/sanity/record/rec004.py b/t4_devkit/sanity/record/rec004.py
@@ -19,6 +19,6 @@ class REC004(RecordCountChecker):
     description = "'EgoPose' record is not empty."
     schema = SchemaName.EGO_POSE
 
-    def check_count(self, records: list[dict]) -> list[Reason]:
+    def check_count(self, records: list[dict]) -> list[Reason] | None:
         num_ego_pose = len(records)
-        return [Reason("'EgoPose' record must not be empty")] if num_ego_pose == 0 else []
+        return [Reason("'EgoPose' record must not be empty")] if num_ego_pose == 0 else None
diff --git a/t4_devkit/sanity/record/rec005.py b/t4_devkit/sanity/record/rec005.py
@@ -19,10 +19,10 @@ class REC005(RecordCountChecker):
     description = "'CalibratedSensor' record is not empty."
     schema = SchemaName.CALIBRATED_SENSOR
 
-    def check_count(self, records: list[dict]) -> list[Reason]:
+    def check_count(self, records: list[dict]) -> list[Reason] | None:
         num_calibrated_sensor = len(records)
         return (
             [Reason("'CalibratedSensor' record must not be empty")]
             if num_calibrated_sensor == 0
-            else []
+            else None
         )
diff --git a/t4_devkit/sanity/record/rec006.py b/t4_devkit/sanity/record/rec006.py
@@ -19,6 +19,6 @@ class REC006(RecordCountChecker):
     description = "'Instance' record is not empty."
     schema = SchemaName.INSTANCE
 
-    def check_count(self, records: list[dict]) -> list[Reason]:
+    def check_count(self, records: list[dict]) -> list[Reason] | None:
         num_instance = len(records)
-        return [Reason("'Instance' record must not be empty")] if num_instance == 0 else []
+        return [Reason("'Instance' record must not be empty")] if num_instance == 0 else None
diff --git a/t4_devkit/sanity/reference/base.py b/t4_devkit/sanity/reference/base.py
@@ -44,7 +44,7 @@ def can_skip(self, context: SanityContext) -> Maybe[Reason]:
             case _:
                 return Maybe.from_value(Reason("Missing 'annotation' directory path"))
 
-    def check(self, context: SanityContext) -> list[Reason]:
+    def check(self, context: SanityContext) -> list[Reason] | None:
         source_file = context.to_schema_file(self.source).unwrap()
         target_file = context.to_schema_file(self.target).unwrap()
         source_records = load_json_safe(source_file).unwrap()
@@ -56,7 +56,7 @@ def check(self, context: SanityContext) -> list[Reason]:
             for record in source_records
             if record[self.reference] not in target_tokens
             and self.is_additional_condition_ok(record)
-        ]
+        ] or None
 
     def is_additional_condition_ok(self, record: dict[str, Any]) -> bool:
         """Return True if the additional condition is met.
diff --git a/t4_devkit/sanity/reference/ref013.py b/t4_devkit/sanity/reference/ref013.py
@@ -26,12 +26,12 @@ class REF013(FileReferenceChecker):
     description = "'SampleData.filename' exists."
     schema = SchemaName.SAMPLE_DATA
 
-    def check(self, context: SanityContext) -> list[Reason]:
+    def check(self, context: SanityContext) -> list[Reason] | None:
         filepath = context.to_schema_file(self.schema).unwrap()
         records = load_json_safe(filepath).unwrap()
         data_root = context.data_root.unwrap()
         return [
             Reason(f"File not found: {record['filename']}")
             for record in records
             if not data_root.joinpath(record["filename"]).exists()
-        ]
+        ] or None
diff --git a/t4_devkit/sanity/reference/ref014.py b/t4_devkit/sanity/reference/ref014.py
@@ -26,7 +26,7 @@ class REF014(FileReferenceChecker):
     description = "'SampleData.filename' exists."
     schema = SchemaName.SAMPLE_DATA
 
-    def check(self, context: SanityContext) -> list[Reason]:
+    def check(self, context: SanityContext) -> list[Reason] | None:
         filepath = context.to_schema_file(self.schema).unwrap()
         records = load_json_safe(filepath).unwrap()
         data_root = context.data_root.unwrap()
@@ -35,4 +35,4 @@ def check(self, context: SanityContext) -> list[Reason]:
             for record in records
             if record.get("info_filename") is not None
             and not data_root.joinpath(record["info_filename"]).exists()
-        ]
+        ] or None
diff --git a/t4_devkit/sanity/reference/ref015.py b/t4_devkit/sanity/reference/ref015.py
@@ -26,12 +26,12 @@ class REF015(FileReferenceChecker):
     description = "'LidarSeg.filename' exists."
     schema = SchemaName.LIDARSEG
 
-    def check(self, context: SanityContext) -> list[Reason]:
+    def check(self, context: SanityContext) -> list[Reason] | None:
         filepath = context.to_schema_file(self.schema).unwrap()
         records = load_json_safe(filepath).unwrap()
         data_root = context.data_root.unwrap()
         return [
             Reason(f"File not found: {record['filename']}")
             for record in records
             if not data_root.joinpath(record["filename"]).exists()
-        ]
+        ] or None
diff --git a/t4_devkit/sanity/result.py b/t4_devkit/sanity/result.py
@@ -46,9 +46,9 @@ class Report:
     reasons: list[Reason] | None = field(default=None)
 
     def __attrs_post_init__(self) -> None:
-        if self.is_success():
-            assert self.reasons is None, "Success report cannot have reasons"
-        else:
+        if self.is_success() and self.severity.is_error():
+            assert self.reasons is None, "Success report for error rule cannot have reasons"
+        elif self.is_failure():
             assert self.reasons is not None, "Non-success report must have reasons"
 
     def is_success(self) -> bool:
@@ -64,9 +64,27 @@ def is_skipped(self) -> bool:
         return self.status == Status.SKIPPED
 
 
-def make_success(id: RuleID, name: RuleName, severity: Severity, description: str) -> Report:
-    """Make a success report for the given rule."""
-    return Report(id, name, severity, description, Status.SUCCESS)
+def make_report(
+    id: RuleID,
+    name: RuleName,
+    severity: Severity,
+    description: str,
+    reasons: list[Reason] | None = None,
+    *,
+    strict: bool = False,
+) -> Report:
+    """Make a report for the given rule."""
+    print(reasons)
+    if reasons:
+        if severity.is_warning():
+            return (
+                Report(id, name, severity, description, Status.FAILURE, reasons)
+                if strict
+                else Report(id, name, severity, description, Status.SUCCESS, reasons)
+            )
+        return Report(id, name, severity, description, Status.FAILURE, reasons)
+    else:
+        return Report(id, name, severity, description, Status.SUCCESS)
 
 
 def make_skipped(
@@ -76,13 +94,6 @@ def make_skipped(
     return Report(id, name, severity, description, Status.SKIPPED, [reason])
 
 
-def make_failure(
-    id: RuleID, name: RuleName, severity: Severity, description: str, reasons: list[Reason]
-) -> Report:
-    """Make a failure report for the given rule."""
-    return Report(id, name, severity, description, Status.FAILURE, reasons)
-
-
 @define
 class SanityResult:
     """The result of a Sanity check.
@@ -141,9 +152,11 @@ def print_sanity_result(result: SanityResult) -> None:
 
     # print summary result
     success = sum(1 for rp in result.reports if rp.is_success())
-    failures = sum(1 for rp in result.reports if rp.severity.is_error() and rp.is_failure())
+    failures = sum(1 for rp in result.reports if rp.is_failure())
     skips = sum(1 for rp in result.reports if rp.is_skipped())
-    warnings = sum(1 for rp in result.reports if rp.severity.is_warning() and rp.is_failure())
+
+    # just count the number of warnings
+    warnings = sum(1 for rp in result.reports if rp.severity.is_warning() and rp.reasons)
 
     summary_rows = [
         [
diff --git a/t4_devkit/sanity/run.py b/t4_devkit/sanity/run.py
@@ -16,6 +16,7 @@ def sanity_check(
     *,
     include_warning: bool = False,
     excludes: Sequence[str] | None = None,
+    strict: bool = False,
 ) -> SanityResult:
     """Run sanity checks on the given data root.
 
@@ -24,6 +25,7 @@ def sanity_check(
         revision (str | None, optional): The revision to check. If None, the latest revision is used.
         include_warning (bool, optional): Whether to include warning checks.
         excludes (Sequence[str] | None, optional): A list of rule names or groups to exclude.
+        strict (bool, optional): Indicate whether checkers whose severity is set to "WARNING" should fail.
 
     Returns:
         A SanityResult object.
@@ -34,7 +36,7 @@ def sanity_check(
         else:
             warnings.simplefilter("ignore")
 
-        context = SanityContext.from_path(data_root, revision=revision)
+        context = SanityContext.from_path(data_root, revision=revision, strict=strict)
 
         checkers = CHECKERS.build(excludes=excludes)
         reports = [checker(context) for checker in checkers]
diff --git a/t4_devkit/sanity/structure/str001.py b/t4_devkit/sanity/structure/str001.py
@@ -23,9 +23,9 @@ class STR001(Checker):
     severity = Severity.WARNING
     description = "'version/' directory exists under the dataset root directory."
 
-    def check(self, context: SanityContext) -> list[Reason]:
+    def check(self, context: SanityContext) -> list[Reason] | None:
         match context.version:
             case Some(_):
-                return []
+                return None
             case _:
                 return [Reason("'version' directory doesn't exist")]
diff --git a/t4_devkit/sanity/structure/str002.py b/t4_devkit/sanity/structure/str002.py
@@ -23,11 +23,11 @@ class STR002(Checker):
     severity = Severity.ERROR
     description = "'annotation/' directory exists under the dataset root directory."
 
-    def check(self, context: SanityContext) -> list[Reason]:
+    def check(self, context: SanityContext) -> list[Reason] | None:
         match context.annotation_dir:
             case Some(x):
                 return (
-                    []
+                    None
                     if x.exists()
                     else [Reason(f"Path to 'annotation' not found: {x.as_posix()}")]
                 )
diff --git a/t4_devkit/sanity/structure/str003.py b/t4_devkit/sanity/structure/str003.py
diff --git a/t4_devkit/sanity/structure/str004.py b/t4_devkit/sanity/structure/str004.py
diff --git a/t4_devkit/sanity/structure/str005.py b/t4_devkit/sanity/structure/str005.py
diff --git a/t4_devkit/sanity/structure/str006.py b/t4_devkit/sanity/structure/str006.py
diff --git a/t4_devkit/sanity/structure/str007.py b/t4_devkit/sanity/structure/str007.py
diff --git a/t4_devkit/sanity/structure/str008.py b/t4_devkit/sanity/structure/str008.py
diff --git a/t4_devkit/sanity/structure/str009.py b/t4_devkit/sanity/structure/str009.py
diff --git a/t4_devkit/sanity/tier4/tiv001.py b/t4_devkit/sanity/tier4/tiv001.py
