[parser.py] SimpleAnnotationParserByTaskにメソッドを追加

Makefile

@@ -10,7 +10,7 @@ format:
 	poetry run black annofabapi tests
 
 lint:
-	poetry run mypy annofabapi tests/create_test_project.py
+	poetry run mypy annofabapi tests
 	poetry run flake8 annofabapi tests/create_test_project.py
 	poetry run pylint annofabapi tests/create_test_project.py
 

annofabapi/__version__.py

@@ -1 +1 @@
-__version__ = "0.42.0"
+__version__ = "0.42.1"

annofabapi/parser.py

@@ -51,8 +51,7 @@ def json_file_path(self) -> str:
     @property
     def input_data_id(self) -> str:
         """
-        JSONファイルから決まる、input_data_id.
-        Simple(v2)版用です。
+        JSONファイルから決まるinput_data_id.
         """
         return self.__input_data_id
 
@@ -201,7 +200,7 @@ class SimpleAnnotationDirParser(SimpleAnnotationParser):
     Examples:
         JSONファイルをパースする::
 
-            p = SimpleAnnotationDirParser(Path("task_id/input_data_id.json"))
+            p = SimpleAnnotationDirParser(Path("annotation/task_id/input_data_id.json"))
             annotation = p.parse()
 
     """
@@ -311,28 +310,134 @@ def __init__(self, task_id: str):
     def task_id(self) -> str:
         return self.__task_id
 
+    @property
+    @abc.abstractmethod
+    def json_file_path_list(self) -> List[str]:
+        """
+        パースするJSONファイルパスのリスト
+        """
+
+    @abc.abstractmethod
+    def get_parser(self, json_file_path: str) -> SimpleAnnotationParser:
+        """
+        JSONファイルパスから、Simple Annotation parserを取得する。
+
+        Args:
+            json_file_path: パースするJSONファイルのパス。``json_file_path_list`` に含まれる値を指定すること。
+
+        Returns:
+            Simple Annotation parser
+
+        Raises:
+            ValueError: ``json_file_path`` の値が ``json_file_path_list`` に含まれていないとき
+
+        """
+
     @abc.abstractmethod
     def lazy_parse(self) -> Iterator[SimpleAnnotationParser]:
         pass
 
 
 class SimpleAnnotationZipParserByTask(SimpleAnnotationParserByTask):
-    def __init__(self, zip_file: zipfile.ZipFile, task_id: str, json_path_list: List[str]):
+    """
+    Simple Annotation zipのparserをタスクごとにまとめたもの。
+
+    Args:
+        zip_file: Simple Annotation zipのzipfileオブジェクト
+        task_id: タスクID
+        json_path_list: パースするJSONパスのリスト。
+            Noneの場合は、``zipfile.ZipFile.infolist()`` 関数を呼び出して、JSONパスのリストを生成します。
+
+    Examples:
+        JSONファイルをパースする::
+
+            with zipfile.ZipFile("simple-annotation.zip", "r") as zip_file:
+                p = SimpleAnnotationZipParserByTask(zip_file, "task1")
+
+    """
+
+    def __get_json_file_path_list(self, task_id: str) -> List[str]:
+        """
+        task_idとJSONパスリストの辞書を取得する。
+        """
+
+        def _match_task_id_and_contain_input_data_json(zip_info: zipfile.ZipInfo) -> bool:
+            """
+            task_idディレクトリ配下の入力データJSONかどうか
+            """
+            paths = [p for p in zip_info.filename.split("/") if len(p) != 0]
+            if len(paths) != 2:
+                return False
+            if paths[0] != task_id:
+                return False
+            if not paths[1].endswith(".json"):
+                return False
+            return True
+
+        return [
+            zip_info.filename
+            for zip_info in self.__zip_file.infolist()
+            if _match_task_id_and_contain_input_data_json(zip_info)
+        ]
+
+    def __init__(self, zip_file: zipfile.ZipFile, task_id: str, json_path_list: Optional[List[str]] = None):
         self.__zip_file = zip_file
-        self.__json_path_list = json_path_list
+        if json_path_list is not None:
+            self.__json_path_list = json_path_list
+        else:
+            self.__json_path_list = self.__get_json_file_path_list(task_id)
         super().__init__(task_id)
 
     def lazy_parse(self) -> Iterator[SimpleAnnotationZipParser]:
         return (SimpleAnnotationZipParser(self.__zip_file, e) for e in self.__json_path_list)
 
+    @property
+    def json_file_path_list(self) -> List[str]:
+        return self.__json_path_list
+
+    def get_parser(self, json_file_path: str) -> SimpleAnnotationParser:
+        if json_file_path in self.__json_path_list:
+            return SimpleAnnotationZipParser(self.__zip_file, json_file_path)
+        else:
+            raise ValueError(f"json_file_path '{json_file_path}' は `json_file_path_list` に含まれていません。")
+
 
 class SimpleAnnotationDirParserByTask(SimpleAnnotationParserByTask):
-    def __init__(self, task_id: str, json_path_list: List[Path]):
-        self.__json_path_list = json_path_list
+    """
+    Simple Annotation zipを展開したディレクトリのparserをタスクごとにまとめたもの。
+
+    Args:
+        task_id: Simple Annotation zipのzipfileオブジェクト
+        task_id: タスクID
+        json_path_list: タスク配下のJSONパスのリスト。パスにはtask_idを含む。
+
+    Examples:
+        JSONファイルをパースする::
+
+            with zipfile.ZipFile("simple-annotation.zip", "r") as zip_file:
+                p = SimpleAnnotationZipParserByTask(zip_file, "task1", ["task1/input1.json","task1/input2.json"])
+
+    """
+
+    def __init__(self, task_dir_path: Path):
+        self.__task_dir_path = task_dir_path
+        task_id = task_dir_path.name
         super().__init__(task_id)
 
     def lazy_parse(self) -> Iterator[SimpleAnnotationDirParser]:
-        return (SimpleAnnotationDirParser(e) for e in self.__json_path_list)
+        return (
+            SimpleAnnotationDirParser(e) for e in self.__task_dir_path.iterdir() if e.is_file() and e.suffix == ".json"
+        )
+
+    @property
+    def json_file_path_list(self) -> List[str]:
+        return [str(e) for e in self.__task_dir_path.iterdir() if e.is_file() and e.suffix == ".json"]
+
+    def get_parser(self, json_file_path: str) -> SimpleAnnotationParser:
+        if json_file_path in self.json_file_path_list:
+            return SimpleAnnotationDirParser(Path(json_file_path))
+        else:
+            raise ValueError(f"json_file_path '{json_file_path}' は `json_file_path_list` に含まれていません。")
 
 
 def __parse_annotation_dir(annotaion_dir_path: Path, clazz) -> Iterator[Any]:
@@ -405,6 +510,9 @@ def is_input_data_json(zip_info: zipfile.ZipInfo) -> bool:
         return True
 
     def create_task_dict(arg_info_list: List[zipfile.ZipInfo]) -> Dict[str, List[str]]:
+        """
+        task_idとJSONパスリストの辞書を取得する。
+        """
         task_dict: Dict[str, List[str]] = {}
         sorted_path_list = sorted([e.filename for e in arg_info_list if is_input_data_json(e)])
 
@@ -447,9 +555,7 @@ def lazy_parse_simple_annotation_dir_by_task(annotaion_dir_path: Path) -> Iterat
         if not task_dir.is_dir():
             continue
 
-        task_id = task_dir.name
-        json_path_list = [e for e in task_dir.iterdir() if e.is_file() and e.suffix == ".json"]
-        yield SimpleAnnotationDirParserByTask(task_id=task_id, json_path_list=json_path_list)
+        yield SimpleAnnotationDirParserByTask(task_dir)
 
 
 def __parse_annotation_zip(zip_file_path: Path, clazz) -> Iterator[Any]:

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "annofabapi"
-version = "0.42.0"
+version = "0.42.1"
 description = "Python Clinet Library of AnnoFab WebAPI (https://annofab.com/docs/api/)"
 authors = ["yuji38kwmt"]
 license = "MIT"

tests/test_local_parser.py

@@ -14,7 +14,9 @@
     FullAnnotationDirParser,
     FullAnnotationZipParser,
     SimpleAnnotationDirParser,
+    SimpleAnnotationDirParserByTask,
     SimpleAnnotationZipParser,
+    SimpleAnnotationZipParserByTask,
 )
 
 # プロジェクトトップに移動する
@@ -25,7 +27,61 @@
 test_dir = Path("./tests/data")
 
 
-class TestSimpleAnnotationV2:
+class TestSimpleAnnotationParser:
+    def test_SimpleAnnotationZipParser(self):
+        zip_path = Path(test_dir / "simple-annotation.zip")
+        with zipfile.ZipFile(zip_path) as zip_file:
+            parser = SimpleAnnotationZipParser(zip_file, "sample_1/c86205d1-bdd4-4110-ae46-194e661d622b.json")
+            assert parser.task_id == "sample_1"
+            assert parser.input_data_id == "c86205d1-bdd4-4110-ae46-194e661d622b"
+            assert parser.json_file_path == "sample_1/c86205d1-bdd4-4110-ae46-194e661d622b.json"
+            with pytest.raises(AnnotationOuterFileNotFoundError):
+                parser.open_outer_file("foo")
+
+    def test_SimpleAnnotationDirParser(self):
+        dir_path = Path(test_dir / "simple-annotation")
+
+        parser = SimpleAnnotationDirParser(dir_path / "sample_1/c86205d1-bdd4-4110-ae46-194e661d622b.json")
+        assert parser.task_id == "sample_1"
+        assert parser.input_data_id == "c86205d1-bdd4-4110-ae46-194e661d622b"
+        assert parser.json_file_path == str(dir_path / "sample_1/c86205d1-bdd4-4110-ae46-194e661d622b.json")
+        with pytest.raises(AnnotationOuterFileNotFoundError):
+            parser.open_outer_file("foo")
+
+
+class TestSimpleAnnotationParserByTask:
+    def test_SimpleAnnotationDirParserByTask(self):
+        annotation_dir = test_dir / "simple-annotation"
+        task_parser = SimpleAnnotationDirParserByTask(annotation_dir / "sample_1")
+        assert task_parser.task_id == "sample_1"
+        json_file_path_list = task_parser.json_file_path_list
+        assert str(annotation_dir / "sample_1/c6e1c2ec-6c7c-41c6-9639-4244c2ed2839.json") in json_file_path_list
+        assert str(annotation_dir / "sample_1/c86205d1-bdd4-4110-ae46-194e661d622b.json") in json_file_path_list
+
+        input_data_parser = task_parser.get_parser(
+            str(annotation_dir / "sample_1/c6e1c2ec-6c7c-41c6-9639-4244c2ed2839.json")
+        )
+        assert input_data_parser.input_data_id == "c6e1c2ec-6c7c-41c6-9639-4244c2ed2839"
+        assert input_data_parser.json_file_path == str(
+            test_dir / "simple-annotation/sample_1/c6e1c2ec-6c7c-41c6-9639-4244c2ed2839.json"
+        )
+
+
+    def test_SimpleAnnotationZipParserByTask(self):
+        with zipfile.ZipFile(test_dir / "simple-annotation.zip") as zip_file:
+            task_parser = SimpleAnnotationZipParserByTask(zip_file, "sample_1")
+
+            assert task_parser.task_id == "sample_1"
+            json_file_path_list = task_parser.json_file_path_list
+            assert "sample_1/c6e1c2ec-6c7c-41c6-9639-4244c2ed2839.json" in json_file_path_list
+            assert "sample_1/c86205d1-bdd4-4110-ae46-194e661d622b.json" in json_file_path_list
+
+            input_data_parser = task_parser.get_parser("sample_1/c6e1c2ec-6c7c-41c6-9639-4244c2ed2839.json")
+            assert input_data_parser.input_data_id == "c6e1c2ec-6c7c-41c6-9639-4244c2ed2839"
+            assert input_data_parser.json_file_path == str("sample_1/c6e1c2ec-6c7c-41c6-9639-4244c2ed2839.json")
+
+
+class TestSimpleAnnotation:
     def test_simple_annotation_zip(self):
         zip_path = Path(test_dir / "simple-annotation.zip")
         iter_parser = annofabapi.parser.lazy_parse_simple_annotation_zip(zip_path)
@@ -42,13 +98,6 @@ def test_simple_annotation_zip(self):
 
         assert index == 4
 
-        with zipfile.ZipFile(zip_path) as zip_file:
-            parser = SimpleAnnotationZipParser(zip_file, "sample_1/c86205d1-bdd4-4110-ae46-194e661d622b.json")
-            assert parser.task_id == "sample_1"
-            assert parser.input_data_id == "c86205d1-bdd4-4110-ae46-194e661d622b"
-            with pytest.raises(AnnotationOuterFileNotFoundError):
-                parser.open_outer_file("foo")
-
     def test_simple_annotation_dir(self):
         dir_path = Path(test_dir / "simple-annotation")
         iter_parser = annofabapi.parser.lazy_parse_simple_annotation_dir(dir_path)
@@ -61,12 +110,6 @@ def test_simple_annotation_dir(self):
 
         assert index == 4
 
-        parser = SimpleAnnotationDirParser(dir_path / "sample_1/c86205d1-bdd4-4110-ae46-194e661d622b.json")
-        assert parser.task_id == "sample_1"
-        assert parser.input_data_id == "c86205d1-bdd4-4110-ae46-194e661d622b"
-        with pytest.raises(AnnotationOuterFileNotFoundError):
-            parser.open_outer_file("foo")
-
     def test_lazy_parse_simple_annotation_zip_by_task(self):
         zip_path = Path(test_dir / "simple-annotation.zip")
         task_parser_list = list(annofabapi.parser.lazy_parse_simple_annotation_zip_by_task(zip_path))
@@ -81,6 +124,8 @@ def test_lazy_parse_simple_annotation_zip_by_task(self):
         assert len([e for e in parser_list if e.input_data_id == "c6e1c2ec-6c7c-41c6-9639-4244c2ed2839"]) == 1
         assert len([e for e in parser_list if e.input_data_id == "c86205d1-bdd4-4110-ae46-194e661d622b"]) == 1
 
+
+
     def test_lazy_parse_simple_annotation_dir_by_task(self):
         zip_path = Path(test_dir / "simple-annotation")
         task_parser_list = list(annofabapi.parser.lazy_parse_simple_annotation_dir_by_task(zip_path))
@@ -96,6 +141,8 @@ def test_lazy_parse_simple_annotation_dir_by_task(self):
         assert len([e for e in parser_list if e.input_data_id == "c86205d1-bdd4-4110-ae46-194e661d622b"]) == 1
 
 
+
+
 class TestFullAnnotation:
     def test_full_annotation_zip(self):
         zip_path = Path(test_dir / "full-annotation.zip")