appease mypy

Author: pmeier

mypy.ini

@@ -125,3 +125,7 @@ ignore_missing_imports = True
 [mypy-h5py.*]
 
 ignore_missing_imports = True
+
+[mypy-rarfile.*]
+
+ignore_missing_imports = True

torchvision/prototype/datasets/_builtin/hmdb51.py

@@ -58,7 +58,7 @@ def resources(self, config: DatasetConfig) -> List[OnlineResource]:
 
     def _is_split_number(self, data: Tuple[str, Any], *, split_number: str) -> bool:
         path = pathlib.Path(data[0])
-        return self._SPLIT_FILE_PATTERN.match(path.name)["split_number"] == split_number  # type: ignore[union-attr]
+        return self._SPLIT_FILE_PATTERN.match(path.name)["split_number"] == split_number  # type: ignore[index]
 
     _SPLIT_ID_TO_NAME = {
         "1": "train",
@@ -111,7 +111,6 @@ def _generate_categories(self, root: pathlib.Path) -> List[str]:
 
         dp = resources[0].load(root)
         categories = {
-            self._SPLIT_FILE_PATTERN.match(pathlib.Path(path).name)["category"]  # type: ignore[union-attr]
-            for path, _ in dp
+            self._SPLIT_FILE_PATTERN.match(pathlib.Path(path).name)["category"] for path, _ in dp  # type: ignore[index]
         }
         return sorted(categories)

torchvision/prototype/datasets/_builtin/ucf101.py

@@ -88,7 +88,7 @@ def _generate_categories(self, root: pathlib.Path) -> Tuple[str, ...]:
         resources = self.resources(config)
 
         dp = resources[0].load(root)
-        dp = Filter(dp, path_comparator("name", "classInd.txt"))
+        dp: IterDataPipe[Tuple[str, BinaryIO]] = Filter(dp, path_comparator("name", "classInd.txt"))
         dp = CSVParser(dp, dialect="ucf101")
         _, categories = zip(*dp)
         return cast(Tuple[str, ...], categories)

torchvision/prototype/datasets/utils/_video.py

@@ -1,5 +1,5 @@
 import random
-from typing import Any, Dict, Iterator, BinaryIO, Optional, Tuple
+from typing import Any, Dict, Iterator, Optional, Tuple
 
 import av
 import numpy as np
@@ -15,7 +15,7 @@ def __init__(self, datapipe: IterDataPipe, *, inline: bool = True) -> None:
         self.datapipe = datapipe
         self._inline = inline
 
-    def _decode(self, buffer: BinaryIO, meta: Dict[str, Any]) -> Iterator[Dict[str, Any]]:
+    def _decode(self, buffer: ReadOnlyTensorBuffer, meta: Dict[str, Any]) -> Iterator[Dict[str, Any]]:
         raise NotImplementedError
 
     def _find_encoded_video(self, id: Tuple[Any, ...], obj: Any) -> Optional[Tuple[Any, ...]]:
@@ -65,13 +65,12 @@ def __iter__(self) -> Iterator[Any]:
                 raise ValueError("more than one encoded video")
             id, video = ids_and_videos[0]
 
-            buffer = ReadOnlyTensorBuffer(video)
-            for data in self._decode(buffer, video.meta.copy()):
+            for data in self._decode(ReadOnlyTensorBuffer(video), video.meta.copy()):
                 yield self._integrate_data(sample, id, data)
 
 
 class KeyframeDecoder(_VideoDecoder):
-    def _decode(self, buffer: BinaryIO, meta: Dict[str, Any]) -> Iterator[Dict[str, Any]]:
+    def _decode(self, buffer: ReadOnlyTensorBuffer, meta: Dict[str, Any]) -> Iterator[Dict[str, Any]]:
         with av.open(buffer, metadata_errors="ignore") as container:
             stream = container.streams.video[0]
             stream.codec_context.skip_frame = "NONKEY"
@@ -92,7 +91,7 @@ def __init__(self, datapipe: IterDataPipe, *, num_samples: int = 1, inline: bool
         super().__init__(datapipe, inline=inline)
         self.num_sampler = num_samples
 
-    def _decode(self, buffer: BinaryIO, meta: Dict[str, Any]) -> Iterator[Dict[str, Any]]:
+    def _decode(self, buffer: ReadOnlyTensorBuffer, meta: Dict[str, Any]) -> Iterator[Dict[str, Any]]:
         with av.open(buffer, metadata_errors="ignore") as container:
             stream = container.streams.video[0]
             # duration is given in time_base units as int
@@ -147,7 +146,7 @@ def _unfold(self, tensor: torch.Tensor, dilation: int = 1) -> torch.Tensor:
             new_size = (0, self.num_frames_per_clip)
         return torch.as_strided(tensor, new_size, new_stride)
 
-    def _decode(self, buffer: BinaryIO, meta: Dict[str, Any]) -> Iterator[Dict[str, Any]]:
+    def _decode(self, buffer: ReadOnlyTensorBuffer, meta: Dict[str, Any]) -> Iterator[Dict[str, Any]]:
         with av.open(buffer, metadata_errors="ignore") as container:
             stream = container.streams.video[0]
             time_base = stream.time_base