chore: add experimental image transform functions to BQ bytes (#1397)

* chore: add experimental image transform functions to BQ bytes

* revert unrelated files

Author: GarrettWu

bigframes/blob/_functions.py

@@ -130,6 +130,32 @@ def image_blur_func(
 image_blur_def = FunctionDef(image_blur_func, ["opencv-python", "numpy", "requests"])
 
 
+def image_blur_to_bytes_func(src_obj_ref_rt: str, ksize_x: int, ksize_y: int) -> bytes:
+    import json
+
+    import cv2 as cv  # type: ignore
+    import numpy as np
+    import requests
+
+    src_obj_ref_rt_json = json.loads(src_obj_ref_rt)
+    src_url = src_obj_ref_rt_json["access_urls"]["read_url"]
+
+    response = requests.get(src_url)
+    bts = response.content
+
+    nparr = np.frombuffer(bts, np.uint8)
+    img = cv.imdecode(nparr, cv.IMREAD_UNCHANGED)
+    img_blurred = cv.blur(img, ksize=(ksize_x, ksize_y))
+    bts = cv.imencode(".jpeg", img_blurred)[1].tobytes()
+
+    return bts
+
+
+image_blur_to_bytes_def = FunctionDef(
+    image_blur_to_bytes_func, ["opencv-python", "numpy", "requests"]
+)
+
+
 def image_resize_func(
     src_obj_ref_rt: str,
     dst_obj_ref_rt: str,
@@ -174,6 +200,38 @@ def image_resize_func(
 )
 
 
+def image_resize_to_bytes_func(
+    src_obj_ref_rt: str,
+    dsize_x: int,
+    dsize_y: int,
+    fx: float,
+    fy: float,
+) -> bytes:
+    import json
+
+    import cv2 as cv  # type: ignore
+    import numpy as np
+    import requests
+
+    src_obj_ref_rt_json = json.loads(src_obj_ref_rt)
+    src_url = src_obj_ref_rt_json["access_urls"]["read_url"]
+
+    response = requests.get(src_url)
+    bts = response.content
+
+    nparr = np.frombuffer(bts, np.uint8)
+    img = cv.imdecode(nparr, cv.IMREAD_UNCHANGED)
+    img_resized = cv.resize(img, dsize=(dsize_x, dsize_y), fx=fx, fy=fy)
+    bts = cv.imencode(".jpeg", img_resized)[1].tobytes()
+
+    return bts
+
+
+image_resize_to_bytes_def = FunctionDef(
+    image_resize_to_bytes_func, ["opencv-python", "numpy", "requests"]
+)
+
+
 def image_normalize_func(
     src_obj_ref_rt: str, dst_obj_ref_rt: str, alpha: float, beta: float, norm_type: str
 ) -> str:
@@ -222,6 +280,43 @@ def image_normalize_func(
 )
 
 
+def image_normalize_to_bytes_func(
+    src_obj_ref_rt: str, alpha: float, beta: float, norm_type: str
+) -> bytes:
+    import json
+
+    import cv2 as cv  # type: ignore
+    import numpy as np
+    import requests
+
+    norm_type_mapping = {
+        "inf": cv.NORM_INF,
+        "l1": cv.NORM_L1,
+        "l2": cv.NORM_L2,
+        "minmax": cv.NORM_MINMAX,
+    }
+
+    src_obj_ref_rt_json = json.loads(src_obj_ref_rt)
+    src_url = src_obj_ref_rt_json["access_urls"]["read_url"]
+
+    response = requests.get(src_url)
+    bts = response.content
+
+    nparr = np.frombuffer(bts, np.uint8)
+    img = cv.imdecode(nparr, cv.IMREAD_UNCHANGED)
+    img_normalized = cv.normalize(
+        img, None, alpha=alpha, beta=beta, norm_type=norm_type_mapping[norm_type]
+    )
+    bts = cv.imencode(".jpeg", img_normalized)[1].tobytes()
+
+    return bts
+
+
+image_normalize_to_bytes_def = FunctionDef(
+    image_normalize_to_bytes_func, ["opencv-python", "numpy", "requests"]
+)
+
+
 # Extracts all text from a PDF url
 def pdf_extract_func(src_obj_ref_rt: str) -> str:
     import io

bigframes/operations/blob.py

@@ -276,7 +276,7 @@ def image_blur(
         self,
         ksize: tuple[int, int],
         *,
-        dst: Union[str, bigframes.series.Series],
+        dst: Optional[Union[str, bigframes.series.Series]] = None,
         connection: Optional[str] = None,
     ) -> bigframes.series.Series:
         """Blurs images.
@@ -286,7 +286,7 @@ def image_blur(
 
         Args:
             ksize (tuple(int, int)): Kernel size.
-            dst (str or bigframes.series.Series): Destination GCS folder str or blob series.
+            dst (str or bigframes.series.Series or None, default None): Destination GCS folder str or blob series. If None, output to BQ as bytes.
             connection (str or None, default None): BQ connection used for function internet transactions, and the output blob if "dst" is str. If None, uses default connection of the session.
 
         Returns:
@@ -295,6 +295,19 @@ def image_blur(
         import bigframes.blob._functions as blob_func
 
         connection = self._resolve_connection(connection)
+        df = self._get_runtime_json_str(mode="R").to_frame()
+
+        if dst is None:
+            image_blur_udf = blob_func.TransformFunction(
+                blob_func.image_blur_to_bytes_def,
+                session=self._block.session,
+                connection=connection,
+            ).udf()
+
+            df["ksize_x"], df["ksize_y"] = ksize
+            res = df.apply(image_blur_udf, axis=1)
+
+            return res
 
         if isinstance(dst, str):
             dst = os.path.join(dst, "")
@@ -311,10 +324,9 @@ def image_blur(
             connection=connection,
         ).udf()
 
-        src_rt = self._get_runtime_json_str(mode="R")
         dst_rt = dst.blob._get_runtime_json_str(mode="RW")
 
-        df = src_rt.to_frame().join(dst_rt.to_frame(), how="outer")
+        df = df.join(dst_rt, how="outer")
         df["ksize_x"], df["ksize_y"] = ksize
 
         res = df.apply(image_blur_udf, axis=1)
@@ -328,7 +340,7 @@ def image_resize(
         *,
         fx: float = 0.0,
         fy: float = 0.0,
-        dst: Union[str, bigframes.series.Series],
+        dst: Optional[Union[str, bigframes.series.Series]] = None,
         connection: Optional[str] = None,
     ):
         """Resize images.
@@ -340,7 +352,7 @@ def image_resize(
             dsize (tuple(int, int), default (0, 0)): Destination size. If set to 0, fx and fy parameters determine the size.
             fx (float, default 0.0): scale factor along the horizontal axis. If set to 0.0, dsize parameter determines the output size.
             fy (float, defalut 0.0): scale factor along the vertical axis. If set to 0.0, dsize parameter determines the output size.
-            dst (str or bigframes.series.Series): Destination GCS folder str or blob series.
+            dst (str or bigframes.series.Series or None, default None): Destination GCS folder str or blob series. If None, output to BQ as bytes.
             connection (str or None, default None): BQ connection used for function internet transactions, and the output blob if "dst" is str. If None, uses default connection of the session.
 
         Returns:
@@ -356,6 +368,20 @@ def image_resize(
         import bigframes.blob._functions as blob_func
 
         connection = self._resolve_connection(connection)
+        df = self._get_runtime_json_str(mode="R").to_frame()
+
+        if dst is None:
+            image_resize_udf = blob_func.TransformFunction(
+                blob_func.image_resize_to_bytes_def,
+                session=self._block.session,
+                connection=connection,
+            ).udf()
+
+            df["dsize_x"], df["dsizye_y"] = dsize
+            df["fx"], df["fy"] = fx, fy
+            res = df.apply(image_resize_udf, axis=1)
+
+            return res
 
         if isinstance(dst, str):
             dst = os.path.join(dst, "")
@@ -372,10 +398,9 @@ def image_resize(
             connection=connection,
         ).udf()
 
-        src_rt = self._get_runtime_json_str(mode="R")
         dst_rt = dst.blob._get_runtime_json_str(mode="RW")
 
-        df = src_rt.to_frame().join(dst_rt.to_frame(), how="outer")
+        df = df.join(dst_rt, how="outer")
         df["dsize_x"], df["dsizye_y"] = dsize
         df["fx"], df["fy"] = fx, fy
 
@@ -390,7 +415,7 @@ def image_normalize(
         alpha: float = 1.0,
         beta: float = 0.0,
         norm_type: str = "l2",
-        dst: Union[str, bigframes.series.Series],
+        dst: Optional[Union[str, bigframes.series.Series]] = None,
         connection: Optional[str] = None,
     ) -> bigframes.series.Series:
         """Normalize images.
@@ -402,7 +427,7 @@ def image_normalize(
             alpha (float, default 1.0): Norm value to normalize to or the lower range boundary in case of the range normalization.
             beta (float, default 0.0): Upper range boundary in case of the range normalization; it is not used for the norm normalization.
             norm_type (str, default "l2"): Normalization type. Accepted values are "inf", "l1", "l2" and "minmax".
-            dst (str or bigframes.series.Series): Destination GCS folder str or blob series.
+            dst (str or bigframes.series.Series or None, default None): Destination GCS folder str or blob series. If None, output to BQ as bytes.
             connection (str or None, default None): BQ connection used for function internet transactions, and the output blob if "dst" is str. If None, uses default connection of the session.
 
         Returns:
@@ -411,6 +436,21 @@ def image_normalize(
         import bigframes.blob._functions as blob_func
 
         connection = self._resolve_connection(connection)
+        df = self._get_runtime_json_str(mode="R").to_frame()
+
+        if dst is None:
+            image_normalize_udf = blob_func.TransformFunction(
+                blob_func.image_normalize_to_bytes_def,
+                session=self._block.session,
+                connection=connection,
+            ).udf()
+
+            df["alpha"] = alpha
+            df["beta"] = beta
+            df["norm_type"] = norm_type
+            res = df.apply(image_normalize_udf, axis=1)
+
+            return res
 
         if isinstance(dst, str):
             dst = os.path.join(dst, "")
@@ -427,10 +467,9 @@ def image_normalize(
             connection=connection,
         ).udf()
 
-        src_rt = self._get_runtime_json_str(mode="R")
         dst_rt = dst.blob._get_runtime_json_str(mode="RW")
 
-        df = src_rt.to_frame().join(dst_rt.to_frame(), how="outer")
+        df = df.join(dst_rt, how="outer")
         df["alpha"] = alpha
         df["beta"] = beta
         df["norm_type"] = norm_type