Add list_file() functional API to FSSpecFileLister and IoPathFileList…

…er (#463)

Summary:
Fixes #387

### Changes
- Adds `list_file()` method on `IoPathFileListerIterDataPipe`
- Adds `list_file()` method on `FSSpecFileListerIterDataPipe`
- Add tests for those methods

#### Additional comments
I feel as if the implementation is quite naive. Would appreciate any feedback on it.

Pull Request resolved: #463

Reviewed By: NivekT

Differential Revision: D36777142

Pulled By: ejguan

fbshipit-source-id: 1c4474776f3fcd377ae545bd8bd7bf26d0b2fa88

test/test_fsspec.py

@@ -64,6 +64,15 @@ def test_fsspec_file_lister_iterdatapipe(self):
  {fsspec.implementations.local.make_path_posix(file) for file in self.temp_sub_files},
  )
 
+ # checks for functional API
+ datapipe = IterableWrapper(["file://" + self.temp_sub_dir.name])
+ datapipe = datapipe.list_files_by_fsspec()
+ for path in datapipe:
+ self.assertIn(
+ path.split("://")[1],
+ {fsspec.implementations.local.make_path_posix(file) for file in self.temp_sub_files},
+ )
+
  @skipIfNoFSSpec
  def test_fsspec_file_lister_iterdatapipe_with_list(self):
  datapipe = FSSpecFileLister(root=["file://" + self.temp_sub_dir.name, "file://" + self.temp_sub_dir_2.name])
@@ -82,6 +91,20 @@ def test_fsspec_file_lister_iterdatapipe_with_list(self):
  # check all file paths within sub_folder are listed
  self.assertEqual(file_lister, temp_files)
 
+ # checks for functional API
+ datapipe = IterableWrapper(["file://" + self.temp_sub_dir.name, "file://" + self.temp_sub_dir_2.name])
+ datapipe = datapipe.list_files_by_fsspec()
+ res = list(map(lambda path: path.split("://")[1], datapipe))
+ res.sort()
+ temp_files = list(
+ map(
+ lambda file: fsspec.implementations.local.make_path_posix(file),
+ self.temp_sub_files + self.temp_sub_files_2,
+ )
+ )
+ temp_files.sort()
+ self.assertEqual(res, temp_files)
+
  @skipIfNoFSSpec
  def test_fsspec_file_loader_iterdatapipe(self):
  datapipe1 = FSSpecFileLister(root="file://" + self.temp_sub_dir.name)

test/test_local_io.py

@@ -660,6 +660,11 @@ def test_io_path_file_lister_iterdatapipe(self):
  for path in datapipe:
  self.assertTrue(path in self.temp_sub_files)
 
+ datapipe = IterableWrapper([self.temp_sub_dir.name])
+ datapipe = datapipe.list_files_by_iopath()
+ for path in datapipe:
+ self.assertTrue(path in self.temp_sub_files)
+
  @skipIfNoIoPath
  def test_io_path_file_lister_iterdatapipe_with_list(self):
  datapipe = IoPathFileLister(root=[self.temp_sub_dir.name, self.temp_sub_dir_2.name])
@@ -672,6 +677,12 @@ def test_io_path_file_lister_iterdatapipe_with_list(self):
  # check all file paths within sub_folder are listed
  self.assertEqual(file_lister, all_temp_files)
 
+ datapipe = IterableWrapper([self.temp_sub_dir.name, self.temp_sub_dir_2.name])
+ datapipe = datapipe.list_files_by_iopath()
+ results = list(datapipe)
+ results.sort()
+ self.assertEqual(results, all_temp_files)
+
  @skipIfNoIoPath
  def test_io_path_file_loader_iterdatapipe(self):
  datapipe1 = IoPathFileLister(root=self.temp_sub_dir.name)

torchdata/datapipes/iter/load/fsspec.py

@@ -33,6 +33,7 @@ def _assert_fsspec() -> None:
  )
 
 
+@functional_datapipe("list_files_by_fsspec")
 class FSSpecFileListerIterDataPipe(IterDataPipe[str]):
  r"""
  Lists the contents of the directory at the provided ``root`` pathname or URL,

torchdata/datapipes/iter/load/iopath.py

@@ -39,6 +39,7 @@ def _create_default_pathmanager():
  return pathmgr
 
 
+@functional_datapipe("list_files_by_iopath")
 class IoPathFileListerIterDataPipe(IterDataPipe[str]):
  r"""
  Lists the contents of the directory at the provided ``root`` pathname or URL,