Mypy fixes

Author: munrojm

emmet-builders/emmet/builders/materials/elasticity.py

@@ -73,9 +73,7 @@ def ensure_index(self):
         self.elasticity.ensure_index("material_id")
         self.elasticity.ensure_index("last_updated")
 
-    def get_items(
-        self,
-    ) -> Generator[Tuple[str, Dict[str, str], List[Dict]], None, None]:
+    def get_items(self,) -> Generator[Tuple[str, Dict[str, str], List[Dict]], None, None]:
         """
         Gets all items to process into elasticity docs.
 
@@ -89,9 +87,7 @@ def get_items(
 
         self.ensure_index()
 
-        cursor = self.materials.query(
-            criteria=self.query, properties=["material_id", "calc_types", "task_ids"]
-        )
+        cursor = self.materials.query(criteria=self.query, properties=["material_id", "calc_types", "task_ids"])
 
         # query for tasks
         query = self.query.copy()
@@ -120,9 +116,7 @@ def get_items(
 
             yield material_id, calc_types, tasks
 
-    def process_item(
-        self, item: Tuple[MPID, Dict[str, str], List[Dict]]
-    ) -> Union[Dict, None]:
+    def process_item(self, item: Tuple[MPID, Dict[str, str], List[Dict]]) -> Union[Dict, None]:
         """
         Process all tasks belong to the same material into an elasticity doc.
 
@@ -161,23 +155,15 @@ def process_item(
 
         # select one task for each set of optimization tasks with the same lattice
         opt_grouped_tmp = group_by_parent_lattice(opt_tasks, mode="opt")
-        opt_grouped = [
-            (lattice, filter_opt_tasks_by_time(tasks, self.logger))
-            for lattice, tasks in opt_grouped_tmp
-        ]
+        opt_grouped = [(lattice, filter_opt_tasks_by_time(tasks, self.logger)) for lattice, tasks in opt_grouped_tmp]
 
         # for deformed tasks with the same lattice, select one if there are multiple
         # tasks with the same deformation
         deform_grouped = group_by_parent_lattice(deform_tasks, mode="deform")
-        deform_grouped = [
-            (lattice, filter_deform_tasks_by_time(tasks))
-            for lattice, tasks in deform_grouped
-        ]
+        deform_grouped = [(lattice, filter_deform_tasks_by_time(tasks)) for lattice, tasks in deform_grouped]
 
         # select opt and deform tasks for fitting
-        final_opt, final_deform = select_final_opt_deform_tasks(
-            opt_grouped, deform_grouped, self.logger
-        )
+        final_opt, final_deform = select_final_opt_deform_tasks(opt_grouped, deform_grouped, self.logger)
         if final_opt is None or final_deform is None:
             return None
 
@@ -187,11 +173,7 @@ def process_item(
         deform_task_ids = []
         deform_dir_names = []
         for doc in final_deform:
-            deforms.append(
-                Deformation(
-                    doc["transmuter"]["transformation_params"][0]["deformation"]
-                )
-            )
+            deforms.append(Deformation(doc["transmuter"]["transformation_params"][0]["deformation"]))
             # -0.1 to convert to GPa from kBar and s
             stresses.append(-0.1 * Stress(doc["output"]["stress"]))
             deform_task_ids.append(doc["task_id"])
@@ -226,9 +208,7 @@ def update_targets(self, items: List[Dict]):
 
 
 def filter_opt_tasks(
-    tasks: List[Dict],
-    calc_types: Dict[str, str],
-    target_calc_type: str = CalcType.GGA_Structure_Optimization,
+    tasks: List[Dict], calc_types: Dict[str, str], target_calc_type: str = CalcType.GGA_Structure_Optimization,
 ) -> List[Dict]:
     """
     Filter optimization tasks, by
@@ -240,9 +220,7 @@ def filter_opt_tasks(
 
 
 def filter_deform_tasks(
-    tasks: List[Dict],
-    calc_types: Dict[str, str],
-    target_calc_type: str = CalcType.GGA_Deformation,
+    tasks: List[Dict], calc_types: Dict[str, str], target_calc_type: str = CalcType.GGA_Deformation,
 ) -> List[Dict]:
     """
     Filter deformation tasks, by
@@ -254,18 +232,13 @@ def filter_deform_tasks(
     for t in tasks:
         if calc_types[str(t["task_id"])] == target_calc_type:
             transforms = t["transmuter"]["transformations"]
-            if (
-                len(transforms) == 1
-                and transforms[0] == "DeformStructureTransformation"
-            ):
+            if len(transforms) == 1 and transforms[0] == "DeformStructureTransformation":
                 deform_tasks.append(t)
 
     return deform_tasks
 
 
-def filter_by_incar_settings(
-    tasks: List[Dict], incar_settings: Dict[str, Any] = None
-) -> List[Dict]:
+def filter_by_incar_settings(tasks: List[Dict], incar_settings: Optional[Dict[str, Any]] = None) -> List[Dict]:
     """
     Filter tasks by incar parameters.
     """
@@ -338,9 +311,7 @@ def filter_opt_tasks_by_time(tasks: List[Dict], logger) -> Dict:
         return selected
 
 
-def filter_deform_tasks_by_time(
-    tasks: List[Dict], deform_comp_tol: float = 1e-5
-) -> List[Dict]:
+def filter_deform_tasks_by_time(tasks: List[Dict], deform_comp_tol: float = 1e-5) -> List[Dict]:
     """
     For deformation tasks with the same deformation, select the latest completed one.
 
@@ -416,10 +387,7 @@ def select_final_opt_deform_tasks(
             tasks.extend(pair[1])
 
         ids = [t["task_id"] for t in tasks]
-        logger.warning(
-            f"Cannot find optimization and deformation tasks that match by lattice "
-            f"for tasks {ids}"
-        )
+        logger.warning(f"Cannot find optimization and deformation tasks that match by lattice " f"for tasks {ids}")
 
         final_opt_task = None
         final_deform_tasks = None