stat_visualization : リファクタリング (#1294)

* りふぁたリング

* 不具合修正

* 不具合修正

annofabcli/stat_visualization/mask_visualization_dir.py

@@ -151,10 +151,9 @@ def mask_visualization_dir(
     not_masked_biography_set: Optional[Set[str]] = None,
     not_masked_user_id_set: Optional[Set[str]] = None,
     minimal_output: bool = False,
-    custom_production_volume_list: Optional[list[ProductionVolumeColumn]] = None,
 ) -> None:
     worktime_per_date = project_dir.read_worktime_per_date_user()
-    task_worktime_by_phase_user = project_dir.read_task_worktime_list(custom_production_volume_list=custom_production_volume_list)
+    task_worktime_by_phase_user = project_dir.read_task_worktime_list()
     df_user = create_df_user(worktime_per_date, task_worktime_by_phase_user)
     replacement_dict = create_replacement_dict(
         df_user,
@@ -182,7 +181,7 @@ def mask_visualization_dir(
     # メンバのパフォーマンスを散布図で出力する
     output_project_dir.write_user_performance_scatter_plot(masked_user_performance)
 
-    masked_task = project_dir.read_task_list(custom_production_volume_list=custom_production_volume_list).mask_user_info(
+    masked_task = project_dir.read_task_list().mask_user_info(
         to_replace_for_user_id=replacement_dict.user_id, to_replace_for_username=replacement_dict.username
     )
     output_project_dir.write_task_list(masked_task)
@@ -223,15 +222,20 @@ def main(args: argparse.Namespace) -> None:
         create_custom_production_volume_list(args.custom_production_volume) if args.custom_production_volume is not None else None
     )
 
-    input_project_dir = ProjectDir(args.dir)
-    output_project_dir = ProjectDir(args.output_dir, metadata=input_project_dir.read_metadata())
+    input_project_dir = ProjectDir(
+        args.dir,
+        custom_production_volume_list=custom_production_volume_list,
+    )
+    output_project_dir = ProjectDir(
+        args.output_dir,
+        metadata=input_project_dir.read_metadata(),
+    )
     mask_visualization_dir(
         project_dir=input_project_dir,
         output_project_dir=output_project_dir,
         not_masked_biography_set=not_masked_biography_set,
         not_masked_user_id_set=not_masked_user_id_set,
         minimal_output=args.minimal,
-        custom_production_volume_list=custom_production_volume_list,
     )
 
 

annofabcli/stat_visualization/merge_visualization_dir.py

@@ -67,22 +67,20 @@ def write_whole_performance(self, whole_performance: WholePerformance):  # noqa:
     @_catch_exception
     def write_cumulative_line_graph(self, task: Task) -> None:
         """ユーザごとにプロットした累積折れ線グラフを出力する。"""
-        df = task.df.copy()
-
         self.output_project_dir.write_cumulative_line_graph(
-            AnnotatorCumulativeProductivity(df),
+            AnnotatorCumulativeProductivity.from_task(task),
             phase=TaskPhase.ANNOTATION,
             user_id_list=self.user_id_list,
             minimal_output=self.minimal_output,
         )
         self.output_project_dir.write_cumulative_line_graph(
-            InspectorCumulativeProductivity(df),
+            InspectorCumulativeProductivity.from_task(task),
             phase=TaskPhase.INSPECTION,
             user_id_list=self.user_id_list,
             minimal_output=self.minimal_output,
         )
         self.output_project_dir.write_cumulative_line_graph(
-            AcceptorCumulativeProductivity(df),
+            AcceptorCumulativeProductivity.from_task(task),
             phase=TaskPhase.ACCEPTANCE,
             user_id_list=self.user_id_list,
             minimal_output=self.minimal_output,

annofabcli/stat_visualization/write_graph.py

@@ -33,32 +33,28 @@ def __init__(
         output_project_dir: ProjectDir,
         *,
         user_id_list: Optional[List[str]] = None,
-        custom_production_volume_list: Optional[List[ProductionVolumeColumn]] = None,
         minimal_output: bool = False,
     ) -> None:
         self.project_dir = project_dir
         self.output_project_dir = output_project_dir
         self.user_id_list = user_id_list
-        self.custom_production_volume_list = custom_production_volume_list
         self.minimal_output = minimal_output
 
     def write_line_graph(self, task: Task) -> None:
-        df = task.df.copy()
-
         self.output_project_dir.write_cumulative_line_graph(
-            AnnotatorCumulativeProductivity(df),
+            AnnotatorCumulativeProductivity.from_task(task),
             phase=TaskPhase.ANNOTATION,
             user_id_list=self.user_id_list,
             minimal_output=self.minimal_output,
         )
         self.output_project_dir.write_cumulative_line_graph(
-            InspectorCumulativeProductivity(df),
+            InspectorCumulativeProductivity.from_task(task),
             phase=TaskPhase.INSPECTION,
             user_id_list=self.user_id_list,
             minimal_output=self.minimal_output,
         )
         self.output_project_dir.write_cumulative_line_graph(
-            AcceptorCumulativeProductivity(df),
+            AcceptorCumulativeProductivity.from_task(task),
             phase=TaskPhase.ACCEPTANCE,
             user_id_list=self.user_id_list,
             minimal_output=self.minimal_output,
@@ -82,14 +78,12 @@ def write_line_graph(self, task: Task) -> None:
     def main(self) -> None:
         try:
             # メンバのパフォーマンスを散布図で出力する
-            self.output_project_dir.write_user_performance_scatter_plot(
-                self.project_dir.read_user_performance(custom_production_volume_list=self.custom_production_volume_list)
-            )
+            self.output_project_dir.write_user_performance_scatter_plot(self.project_dir.read_user_performance())
         except Exception:
             logger.warning("'メンバごとの生産性と品質.csv'から生成できるグラフの出力に失敗しました。", exc_info=True)
 
         try:
-            task = self.project_dir.read_task_list(custom_production_volume_list=self.custom_production_volume_list)
+            task = self.project_dir.read_task_list()
             # ヒストグラムを出力
             self.output_project_dir.write_task_histogram(task)
             # ユーザごとにプロットした折れ線グラフを出力
@@ -98,17 +92,13 @@ def main(self) -> None:
             logger.warning("'タスクlist.csv'から生成できるグラフの出力に失敗しました。", exc_info=True)
 
         try:
-            self.output_project_dir.write_whole_productivity_line_graph_per_date(
-                self.project_dir.read_whole_productivity_per_date(custom_production_volume_list=self.custom_production_volume_list)
-            )
+            self.output_project_dir.write_whole_productivity_line_graph_per_date(self.project_dir.read_whole_productivity_per_date())
         except Exception:
             logger.warning("'日毎の生産量と生産性.csv'から生成できるグラフの出力に失敗しました。", exc_info=True)
 
         try:
             self.output_project_dir.write_whole_productivity_line_graph_per_annotation_started_date(
-                self.project_dir.read_whole_productivity_per_first_annotation_started_date(
-                    custom_production_volume_list=self.custom_production_volume_list
-                )
+                self.project_dir.read_whole_productivity_per_first_annotation_started_date()
             )
         except Exception:
             logger.warning("'教師付者_教師付開始日list.csv'から生成できるグラフの出力に失敗しました。", exc_info=True)
@@ -138,14 +128,13 @@ def main(args: argparse.Namespace) -> None:
         create_custom_production_volume_list(args.custom_production_volume) if args.custom_production_volume is not None else None
     )
 
-    input_project_dir = ProjectDir(args.dir)
+    input_project_dir = ProjectDir(args.dir, custom_production_volume_list=custom_production_volume_list)
     output_project_dir = ProjectDir(args.output_dir, metadata=input_project_dir.read_metadata())
     main_obj = WritingGraph(
         project_dir=input_project_dir,
         output_project_dir=output_project_dir,
         minimal_output=args.minimal,
         user_id_list=user_id_list,
-        custom_production_volume_list=custom_production_volume_list,
     )
     main_obj.main()
 

annofabcli/statistics/visualization/dataframe/cumulative_productivity.py

@@ -187,7 +187,8 @@ def from_task(cls, task: Task) -> AnnotatorCumulativeProductivity:
         """
         `タスクlist.csv`に相当する情報から、インスタンスを生成します。
         """
-        return cls(task.df, custom_production_volume_list=task.custom_production_volume_list)
+        # task.df
+        return cls(task.df.copy(), custom_production_volume_list=task.custom_production_volume_list)
 
     def _get_cumulative_dataframe(self) -> pandas.DataFrame:
         """
@@ -308,7 +309,7 @@ def from_task(cls, task: Task) -> InspectorCumulativeProductivity:
         """
         `タスクlist.csv`に相当する情報から、インスタンスを生成します。
         """
-        return cls(task.df, custom_production_volume_list=task.custom_production_volume_list)
+        return cls(task.df.copy(), custom_production_volume_list=task.custom_production_volume_list)
 
     def _get_cumulative_dataframe(self) -> pandas.DataFrame:
         """
@@ -416,7 +417,7 @@ def from_task(cls, task: Task) -> AcceptorCumulativeProductivity:
         """
         `タスクlist.csv`に相当する情報から、インスタンスを生成します。
         """
-        return cls(task.df, custom_production_volume_list=task.custom_production_volume_list)
+        return cls(task.df.copy(), custom_production_volume_list=task.custom_production_volume_list)
 
     def _get_cumulative_dataframe(self) -> pandas.DataFrame:
         """

annofabcli/statistics/visualization/dataframe/productivity_per_date.py

@@ -455,7 +455,6 @@ def plot_production_volume_metrics(
         アノテーション単位の生産性を受入作業者ごとにプロットする。
 
         """
-
         if not self._validate_df_for_output(output_file):
             return
 
@@ -532,6 +531,7 @@ def plot_production_volume_metrics(
             color = get_color_from_palette(user_index)
             username = df_subset.iloc[0]["first_inspection_username"]
 
+            line_count += 1
             for line_graph, (x_column, y_column) in zip(line_graph_list, columns_list):
                 if y_column.endswith(WEEKLY_MOVING_AVERAGE_COLUMN_SUFFIX):
                     line_graph.add_moving_average_line(

annofabcli/statistics/visualization/project_dir.py

@@ -50,9 +50,16 @@ class ProjectDir(DataClassJsonMixin):
     FILENAME_PROJECT_INFO = "project_info.json"
     FILENAME_MERGE_INFO = "merge_info.json"
 
-    def __init__(self, project_dir: Path, *, metadata: Optional[dict[str, Any]] = None) -> None:
+    def __init__(
+        self,
+        project_dir: Path,
+        *,
+        metadata: Optional[dict[str, Any]] = None,
+        custom_production_volume_list: Optional[list[ProductionVolumeColumn]] = None,
+    ) -> None:
         self.project_dir = project_dir
         self.metadata = metadata
+        self.custom_production_volume_list = custom_production_volume_list
 
     def __repr__(self) -> str:
         return f"ProjectDir(project_dir={self.project_dir!r})"
@@ -77,27 +84,27 @@ def is_merged(self) -> bool:
         """
         return (self.project_dir / self.FILENAME_MERGE_INFO).exists()
 
-    def read_task_list(self, *, custom_production_volume_list: Optional[list[ProductionVolumeColumn]] = None) -> Task:
+    def read_task_list(self) -> Task:
         """`タスクlist.csv`を読み込む。"""
         file = self.project_dir / self.FILENAME_TASK_LIST
         if file.exists():
-            return Task.from_csv(file, custom_production_volume_list=custom_production_volume_list)
+            return Task.from_csv(file, custom_production_volume_list=self.custom_production_volume_list)
         else:
             logger.warning(f"'{file!s}'を読み込もうとしましたが、ファイルは存在しません。")
-            return Task.empty(custom_production_volume_list=custom_production_volume_list)
+            return Task.empty(custom_production_volume_list=self.custom_production_volume_list)
 
     def write_task_list(self, obj: Task) -> None:
         """`タスクlist.csv`を書き込む。"""
         obj.to_csv(self.project_dir / self.FILENAME_TASK_LIST)
 
-    def read_task_worktime_list(self, *, custom_production_volume_list: Optional[list[ProductionVolumeColumn]] = None) -> TaskWorktimeByPhaseUser:
+    def read_task_worktime_list(self) -> TaskWorktimeByPhaseUser:
         """`task-worktime-list.csv`を読み込む。"""
         file = self.project_dir / self.FILENAME_TASK_WORKTIME_LIST
         if file.exists():
-            return TaskWorktimeByPhaseUser.from_csv(file, custom_production_volume_list=custom_production_volume_list)
+            return TaskWorktimeByPhaseUser.from_csv(file, custom_production_volume_list=self.custom_production_volume_list)
         else:
             logger.warning(f"'{file!s}'を読み込もうとしましたが、ファイルは存在しません。")
-            return TaskWorktimeByPhaseUser.empty(custom_production_volume_list=custom_production_volume_list)
+            return TaskWorktimeByPhaseUser.empty(custom_production_volume_list=self.custom_production_volume_list)
 
     def write_task_worktime_list(self, obj: TaskWorktimeByPhaseUser) -> None:
         """`task-worktime-list.csv`を書き込む。"""
@@ -196,32 +203,30 @@ def write_performance_line_graph_per_date(
                 metadata=self.metadata,
             )
 
-    def read_whole_performance(self, *, custom_production_volume_list: Optional[list[ProductionVolumeColumn]] = None) -> WholePerformance:
+    def read_whole_performance(self) -> WholePerformance:
         """`全体の生産性と品質.csv`を読み込む。"""
         file = self.project_dir / self.FILENAME_WHOLE_PERFORMANCE
         if file.exists():
-            return WholePerformance.from_csv(file, custom_production_volume_list=custom_production_volume_list)
+            return WholePerformance.from_csv(file, custom_production_volume_list=self.custom_production_volume_list)
         else:
             logger.warning(f"'{file!s}'を読み込もうとしましたが、ファイルは存在しません。")
-            return WholePerformance.empty(custom_production_volume_list=custom_production_volume_list)
+            return WholePerformance.empty(custom_production_volume_list=self.custom_production_volume_list)
 
     def write_whole_performance(self, whole_performance: WholePerformance) -> None:
         """`全体の生産性と品質.csv`を出力します。"""
         whole_performance.to_csv(self.project_dir / self.FILENAME_WHOLE_PERFORMANCE)
 
-    def read_whole_productivity_per_date(
-        self, *, custom_production_volume_list: Optional[list[ProductionVolumeColumn]] = None
-    ) -> WholeProductivityPerCompletedDate:
+    def read_whole_productivity_per_date(self) -> WholeProductivityPerCompletedDate:
         """
         日ごとの生産性と品質の情報を読み込みます。
         """
         file = self.project_dir / self.FILENAME_WHOLE_PRODUCTIVITY_PER_DATE
         if file.exists():
             return WholeProductivityPerCompletedDate.from_csv(
-                self.project_dir / self.FILENAME_WHOLE_PRODUCTIVITY_PER_DATE, custom_production_volume_list=custom_production_volume_list
+                self.project_dir / self.FILENAME_WHOLE_PRODUCTIVITY_PER_DATE, custom_production_volume_list=self.custom_production_volume_list
             )
         else:
-            return WholeProductivityPerCompletedDate.empty(custom_production_volume_list=custom_production_volume_list)
+            return WholeProductivityPerCompletedDate.empty(custom_production_volume_list=self.custom_production_volume_list)
 
     def write_whole_productivity_per_date(self, obj: WholeProductivityPerCompletedDate) -> None:
         """
@@ -236,43 +241,41 @@ def write_whole_productivity_line_graph_per_date(self, obj: WholeProductivityPer
         obj.plot(self.project_dir / "line-graph/折れ線-横軸_日-全体.html", metadata=self.metadata)
         obj.plot_cumulatively(self.project_dir / "line-graph/累積折れ線-横軸_日-全体.html", metadata=self.metadata)
 
-    def read_whole_productivity_per_first_annotation_started_date(
-        self, custom_production_volume_list: Optional[list[ProductionVolumeColumn]] = None
-    ) -> WholeProductivityPerFirstAnnotationStartedDate:
+    def read_whole_productivity_per_first_annotation_started_date(self) -> WholeProductivityPerFirstAnnotationStartedDate:
         """
         教師付開始日ごとの生産性と品質の情報を読み込みます。
         """
         file = self.project_dir / self.FILENAME_WHOLE_PRODUCTIVITY_PER_FIRST_ANNOTATION_STARTED_DATE
         if file.exists():
             return WholeProductivityPerFirstAnnotationStartedDate.from_csv(
                 file,
-                custom_production_volume_list=custom_production_volume_list,
+                custom_production_volume_list=self.custom_production_volume_list,
             )
         else:
-            return WholeProductivityPerFirstAnnotationStartedDate.empty(custom_production_volume_list=custom_production_volume_list)
+            return WholeProductivityPerFirstAnnotationStartedDate.empty(custom_production_volume_list=self.custom_production_volume_list)
 
-    def write_whole_productivity_per_first_annotation_started_date(self, obj: WholeProductivityPerFirstAnnotationStartedDate):  # noqa: ANN201
+    def write_whole_productivity_per_first_annotation_started_date(self, obj: WholeProductivityPerFirstAnnotationStartedDate) -> None:
         """
         教師付開始日ごとの生産性と品質の情報を書き込みます。
         """
         obj.to_csv(self.project_dir / self.FILENAME_WHOLE_PRODUCTIVITY_PER_FIRST_ANNOTATION_STARTED_DATE)
 
-    def write_whole_productivity_line_graph_per_annotation_started_date(self, obj: WholeProductivityPerFirstAnnotationStartedDate):  # noqa: ANN201
+    def write_whole_productivity_line_graph_per_annotation_started_date(self, obj: WholeProductivityPerFirstAnnotationStartedDate) -> None:
         """
         横軸が教師付開始日、縦軸が全体の生産性などをプロットした折れ線グラフを出力します。
         """
         obj.plot(self.project_dir / "line-graph/折れ線-横軸_教師付開始日-全体.html", metadata=self.metadata)
 
-    def read_user_performance(self, *, custom_production_volume_list: Optional[list[ProductionVolumeColumn]] = None) -> UserPerformance:
+    def read_user_performance(self) -> UserPerformance:
         """
         メンバごとの生産性と品質の情報を読み込みます。
         """
         file = self.project_dir / self.FILENAME_USER_PERFORMANCE
         if file.exists():
-            return UserPerformance.from_csv(file, custom_production_volume_list=custom_production_volume_list)
+            return UserPerformance.from_csv(file, custom_production_volume_list=self.custom_production_volume_list)
         else:
             logger.warning(f"'{file!s}'を読み込もうとしましたが、ファイルは存在しません。")
-            return UserPerformance.empty(custom_production_volume_list=custom_production_volume_list)
+            return UserPerformance.empty(custom_production_volume_list=self.custom_production_volume_list)
 
     def write_user_performance(self, user_performance: UserPerformance) -> None:
         """