Rename to auto.py with tests

Author: john-liu2

batch_img/defaults.py batch_img/auto.pybatch_img/defaults.py renamed to batch_img/auto.py

@@ -1,4 +1,4 @@
-"""class Defaults: apply default actions to the image file(s):
+"""class Auto - do auto actions to the image file(s):
     * Resize to 1280 pixels as the max length
     * Add the border of 5 pixel width in green color
     * Auto-rotate if upside down or sideways
@@ -20,7 +20,7 @@
 pillow_heif.register_heif_opener()  # allow Pillow to open HEIC files
 
 
-class Defaults:
+class Auto:
     @staticmethod
     def resize_add_border(in_path: Path, out_path: Path) -> tuple:
         """Resize and add border to an image file:
@@ -105,8 +105,8 @@ def do_actions(in_path: Path, out_path: Path) -> tuple:
         Returns:
             tuple: bool, str
         """
-        _, file = Defaults.rotate_if_needed(in_path, out_path)
-        return Defaults.resize_add_border(file, out_path)
+        _, file = Auto.rotate_if_needed(in_path, out_path)
+        return Auto.resize_add_border(file, out_path)
 
     @staticmethod
     def run_on_all(in_path: Path, out_path: Path) -> bool:
@@ -128,7 +128,7 @@ def run_on_all(in_path: Path, out_path: Path) -> bool:
 
         logger.info(f"Do default actions on {files_cnt} files in multiprocess ...")
         success_cnt = Common.multiprocess_progress_bar(
-            Defaults.do_actions, "Default action on image files", tasks
+            Auto.do_actions, "Default action on image files", tasks
         )
         logger.info(f"\nFinished default actions on {success_cnt}/{files_cnt} files")
         return True

batch_img/orientation.py

@@ -14,15 +14,11 @@
 
 pillow_heif.register_heif_opener()  # allow Pillow to open HEIC files
 
-ORIENTATION_MAP = {
-    1: "normal",
-    2: "mirrored_horizontal",
-    3: "upside_down",
-    4: "mirrored_vertical",
-    5: "rotated_left_mirrored",
-    6: "rotated_left",
-    7: "rotated_right_mirrored",
-    8: "rotated_right",
+ROTATION_MAP = {  # map to the clockwise angle to correct
+    "bottom": 0,
+    "top": 180,
+    "left": 270,  # rotated right
+    "right": 90,  # rotated left
 }
 EXIF_CW_ANGLE = {
     1: 0,
@@ -81,7 +77,7 @@ def _rotate_image(img, angle: int):
         return img
 
     def get_cw_angle_by_face(self, file: Path) -> int:
-        """Detect orientation by face in mage by Haar Cascades:
+        """Get image orientation by face using Haar Cascades:
         * Fastest but least accurate
         * Works best with frontal faces
         * May produce false positives
@@ -105,8 +101,134 @@ def get_cw_angle_by_face(self, file: Path) -> int:
                 faces = face_cascade.detectMultiScale(
                     gray, scaleFactor=1.2, minNeighbors=6
                 )
-                # logger.info(f"{len(faces)=}")
                 if len(faces) > 0:
                     return angle_cw
         logger.warning(f"Found no face in {file}")
         return -1
+
+    @staticmethod
+    def get_orientation_by_floor(file: Path) -> int:
+        """Get image orientation by floor
+
+        Args:
+            file: image file path
+
+        Returns:
+            int: clockwise angle: 0, 90, 180, 270
+        """
+        with Image.open(file) as img:
+            opencv_img = np.array(img)
+            if opencv_img is None:
+                raise ValueError(f"Failed to load {file}")
+
+        # Convert to HSV for color-based floor detection
+        hsv = cv2.cvtColor(opencv_img, cv2.COLOR_BGR2HSV)
+        # Heuristic: floors are often low saturation, medium-low value (gray/brown)
+        lower_floor = np.array([0, 0, 30])
+        upper_floor = np.array([180, 100, 180])
+        mask = cv2.inRange(hsv, lower_floor, upper_floor)
+
+        h, w, _ = opencv_img.shape
+        regions = {  # Divide image into 4 regions
+            "top": mask[0 : h // 3, :],
+            "bottom": mask[2 * h // 3 :, :],
+            "left": mask[:, 0 : w // 3],
+            "right": mask[:, 2 * w // 3 :],
+        }
+        counts = {k: cv2.countNonZero(v) for k, v in regions.items()}
+        logger.info(f"Floor pixels cnt: {counts=}")
+
+        max_region = max(counts, key=counts.get)
+        cw_angle = ROTATION_MAP.get(max_region, -1)
+        return cw_angle
+
+    @staticmethod
+    def get_cw_angle_by_sky(file: Path) -> int:
+        """Get image orientation by sky, clouds
+
+        Args:
+            file: image file path
+
+        Returns:
+            int: clockwise angle: 0, 90, 180, 270
+        """
+        with Image.open(file) as img:
+            opencv_img = np.array(img)
+            if opencv_img is None:
+                raise ValueError(f"Failed to load {file}")
+
+        # Convert to HSV to detect sky (blue-ish) and cloud (white-ish)
+        hsv = cv2.cvtColor(opencv_img, cv2.COLOR_BGR2HSV)
+
+        # sky_lower = np.array([90, 20, 70])
+        # sky_upper = np.array([140, 255, 255])
+        sky_lower = np.array([80, 40, 100])
+        sky_upper = np.array([140, 200, 255])
+
+        # cloud_lower = np.array([0, 0, 180])
+        # cloud_upper = np.array([180, 70, 255])
+        cloud_lower = np.array([0, 0, 180])
+        cloud_upper = np.array([180, 70, 255])
+
+        # Masks
+        sky_mask = cv2.inRange(hsv, sky_lower, sky_upper)
+        cloud_mask = cv2.inRange(hsv, cloud_lower, cloud_upper)
+        sky_cloud_mask = cv2.bitwise_or(sky_mask, cloud_mask)
+
+        h, w, _ = opencv_img.shape
+        regions = {  # Divide image into 4 regions
+            "top": sky_cloud_mask[0 : h // 3, :],
+            "bottom": sky_cloud_mask[2 * h // 3 :, :],
+            "left": sky_cloud_mask[:, 0 : w // 3],
+            "right": sky_cloud_mask[:, 2 * w // 3 :],
+        }
+        counts = {k: cv2.countNonZero(v) for k, v in regions.items()}
+        logger.info(f"Sky/Cloud pixels cnt: {counts=}")
+        return ROTATION_MAP.get(max(counts, key=counts.get), -1)
+
+    @staticmethod
+    def get_sky_score(img):
+        h, _ = img.shape[:2]
+        hsv = cv2.cvtColor(img, cv2.COLOR_BGR2HSV)
+
+        # Sky: blue range
+        sky_lower = np.array([90, 20, 70])
+        sky_upper = np.array([140, 255, 255])
+
+        # Clouds: low saturation, high value (white)
+        cloud_lower = np.array([0, 0, 180])
+        cloud_upper = np.array([180, 50, 255])
+
+        sky_mask = cv2.inRange(hsv, sky_lower, sky_upper)
+        cloud_mask = cv2.inRange(hsv, cloud_lower, cloud_upper)
+
+        combined_mask = cv2.bitwise_or(sky_mask, cloud_mask)
+
+        # Only consider the top 1/3 region of the image
+        top_region = combined_mask[0 : h // 3, :]
+        score = cv2.countNonZero(top_region)
+        return score
+
+    def get_cw_angle_sky_by_rotate(self, file: Path) -> int:
+        """Get image orientation by sky, clouds
+
+        Args:
+            file: image file path
+
+        Returns:
+            int: clockwise angle: 0, 90, 180, 270
+        """
+        with Image.open(file) as img:
+            opencv_img = np.array(img)
+            if opencv_img is None:
+                raise ValueError(f"Failed to load {file}")
+        scores = {}
+        for angle in [0, 90, 180, 270]:
+            rotated = self._rotate_image(opencv_img, angle)
+            score = self.get_sky_score(rotated)
+            scores[angle] = score
+        logger.info(f"Sky/Cloud: {scores=}")
+
+        # Best orientation is the one with most sky at the top
+        best_angle = max(scores, key=scores.get)
+        return best_angle

tests/data/HEIC/IMG_2529_180cw.HEIC

@@ -1,5 +1,5 @@
-"""Test defaults.py
-pytest -sv tests/test_defaults.py
+"""Test auto.py
+pytest -sv tests/test_auto.py
 Copyright © 2025 John Liu
 """
 
@@ -8,7 +8,7 @@
 from unittest.mock import patch
 
 import pytest
-from batch_img.defaults import Defaults
+from batch_img.auto import Auto
 
 
 @pytest.fixture(
@@ -26,24 +26,23 @@ def data_resize_add_border(request):
 
 def test_resize_add_border(data_resize_add_border):
     in_path, out_path, expected = data_resize_add_border
-    actual = Defaults.resize_add_border(in_path, out_path)
+    actual = Auto.resize_add_border(in_path, out_path)
     assert actual == expected
 
 
 @patch("PIL.Image.open")
 def test_error_resize_add_border(mock_open):
     mock_open.side_effect = ValueError("VE")
-    actual = Defaults.resize_add_border(Path("img/file"), Path("out/path"))
+    actual = Auto.resize_add_border(Path("img/file"), Path("out/path"))
     assert "img/file" in actual[1]
 
 
 @pytest.fixture(
     params=[
-        # JL 2025-08-18: race condition when pytest --cov-report=term --cov=batch_img tests
         # (
-        #     Path(f"{dirname(__file__)}/data/HEIC/chef_orientation_3.heic"),
+        #     Path(f"{dirname(__file__)}/data/HEIC/chef_180cw.heic"),
         #     Path(f"{dirname(__file__)}/.out/"),
-        #     (True, Path(f"{dirname(__file__)}/.out/chef_orientation_3_180cw.heic")),
+        #     (True, Path(f"{dirname(__file__)}/.out/chef_180cw_180cw.heic")),
         # ),
         (
             Path(f"{dirname(__file__)}/data/HEIC/chef_show2.heic"),
@@ -58,7 +57,7 @@ def data_rotate_if_needed(request):
 
 def test_rotate_if_needed(data_rotate_if_needed):
     in_path, out_path, expected = data_rotate_if_needed
-    actual = Defaults.rotate_if_needed(in_path, out_path)
+    actual = Auto.rotate_if_needed(in_path, out_path)
     assert actual == expected
 
 
@@ -80,11 +79,11 @@ def data_do_actions(request):
     return request.param
 
 
-# JL 2025-08-18: race condition when pytest --cov-report=term --cov=batch_img tests
-# def test_do_actions(data_do_actions):
-#     in_path, out_path, expected = data_do_actions
-#     actual = Defaults.do_actions(in_path, out_path)
-#     assert actual == expected
+@pytest.mark.slow(reason="This test modifies test data file.")
+def test_do_actions(data_do_actions):
+    in_path, out_path, expected = data_do_actions
+    actual = Auto.do_actions(in_path, out_path)
+    assert actual == expected
 
 
 @pytest.fixture(
@@ -105,8 +104,8 @@ def data_run_on_all(request):
     return request.param
 
 
-# JL 2025-08-18: race condition when pytest --cov-report=term --cov=batch_img tests
-# def test_run_on_all(data_run_on_all):
-#     in_path, out_path, expected = data_run_on_all
-#     actual = Defaults.run_on_all(in_path, out_path)
-#     assert actual == expected
+@pytest.mark.slow(reason="This test modifies test data file.")
+def test_run_on_all(data_run_on_all):
+    in_path, out_path, expected = data_run_on_all
+    actual = Auto.run_on_all(in_path, out_path)
+    assert actual == expected