MPR Preview first mockup after updating image

Preview first mockup after updating image

mockup_package/mockup/__init__.py

@@ -1 +1,2 @@
 from .image import mockup as startMockup  # noqa: F401
+from .image import previewMockup  # noqa: F401

mockup_package/mockup/image.py

@@ -1,49 +1,84 @@
 import base64
 import sys
 from pathlib import Path
+from typing import Any
 from pyodide.http import pyfetch
 from mockup.image_generator import ImageGenerator as IG
 import os
 
 
-async def mockup(location, device_id, original_img_path_list, device_info):
-    device_path_prefix = f"{location.split('/')[0]}//{location.split('/')[2]}"
-    device_mask_path_prefix = device_path_prefix + "/images/mockup_mask_templates/"
+async def generate(
+    location: str,
+    original_img_path: str,
+    spec: dict[str, Any],
+    ig: IG,
+) -> tuple[str, str, str]:
+    device_path_prefix: str = f"{location.split('/')[0]}//{location.split('/')[2]}"
+    device_mask_path_prefix: str = device_path_prefix + "/images/mockup_mask_templates/"
     device_path_prefix += "/images/mockup_templates/"
-    device_path = "./device.png"
-    device_mask_path = "./device_mask.png"
-    output_img_path_list = []
+    device_path: str = "./device.png"
+    device_mask_path: str = "./device_mask.png"
+
+    try:
+        await process_response(
+            device_path_prefix + str(spec["image"]),
+            device_path,
+        )
+        await process_response(
+            device_mask_path_prefix + str(spec["image"]),
+            device_mask_path,
+        )
+    except Exception as e:
+        print(e, file=sys.stderr)
+        # js.errorBox(e)
+        raise
+    ig.create_fit_coord_image(spec)
+    deviceView = str(spec["image"]).split("-")[-1].split(".")[0]
+    path = (
+        f"{os.path.splitext(os.path.basename(original_img_path))[0]}"
+        + f"-{deviceView}.png"
+    )
+    ig.create_mockup_image(device_path, device_mask_path, path)
+    return (path, original_img_path, deviceView)
+
+
+async def mockup(
+    location: str,
+    device_id: str,
+    original_img_path_list: list[str],
+    device_info: dict[str, Any],
+):
+    output_img_path_list: list[tuple[str, str, str]] = []
     for original_img_path in original_img_path_list:
         ig = IG(original_img_path, device_id, device_info)
         ig.create_fit_resolution_image()
         for spec in ig.phone_models.get(device_id).get("mockups").values():
-            try:
-                await process_response(
-                    device_path_prefix + str(spec["image"]),
-                    device_path,
-                )
-                await process_response(
-                    device_mask_path_prefix + str(spec["image"]),
-                    device_mask_path,
-                )
-            except Exception as e:
-                print(e, file=sys.stderr)
-                # js.errorBox(e)
-                raise
-            ig.create_fit_coord_image(spec)
-            deviceView = str(spec["image"]).split("-")[-1].split(".")[0]
-            path = (
-                f"{os.path.splitext(os.path.basename(original_img_path))[0]}"
-                + f"-{deviceView}.png"
+            output_img_path_list.append(
+                await generate(location, original_img_path, spec, ig)
             )
-            ig.create_mockup_image(device_path, device_mask_path, path)
-            output_img_path_list.append([path, original_img_path, deviceView])
 
     original_img_path_list.clear()
     return output_img_path_list
 
 
-async def download(url):
+async def previewMockup(
+    location: str,
+    device_id: str,
+    original_img_path: str,
+    device_info: dict[str, Any],
+    preview_orientation_index: int = 0,
+):
+    ig = IG(original_img_path, device_id, device_info)
+    ig.create_fit_resolution_image()
+    spec = list(ig.phone_models.get(device_id).get("mockups").values())[
+        preview_orientation_index
+    ]
+    output_img_path = await generate(location, original_img_path, spec, ig)
+
+    return output_img_path
+
+
+async def download(url: str):
     filename = Path(url).name
     response = await pyfetch(url)
     if response.status == 200:
@@ -57,7 +92,7 @@ async def download(url):
         return filename, status
 
 
-async def process_response(url, path):
+async def process_response(url: str, path: str):
     response_content = await download(url)
     if response_content[1] == 200:
         data = base64.b64encode(open(response_content[0], "rb").read())

public/image_process.py

@@ -2,11 +2,13 @@
 import io
 import os
 
-from js import Uint8Array, imageUploadList
+from js import Uint8Array, imageUploadList, imageUpload
 from PIL import Image
 
 
-async def upload_single_image(origin_image, file_name, original_img_path_list):
+async def upload_single_image_and_save_to_list(
+    origin_image, file_name, original_img_path_list
+):
     array_buf = Uint8Array.new(await origin_image.arrayBuffer())
     bytes_list = bytearray(array_buf)
     origin_bytes = io.BytesIO(bytes_list)
@@ -16,17 +18,51 @@ async def upload_single_image(origin_image, file_name, original_img_path_list):
     my_image.save(filePath)
 
 
+async def upload_single_image(origin_image, file_name):
+    array_buf = Uint8Array.new(await origin_image.arrayBuffer())
+    bytes_list = bytearray(array_buf)
+    origin_bytes = io.BytesIO(bytes_list)
+    my_image = Image.open(origin_bytes)
+    filePath = f"./{file_name}.png"
+    my_image.save(filePath)
+    return filePath
+
+
+async def upload_file():
+    basename, ext = os.path.splitext(imageUpload.name)
+    if ext.lower() not in [".psd", ".jpg", ".jpeg", ".png"]:
+        return
+    original_img_path = await upload_single_image(imageUpload, basename)
+    return original_img_path
+
+
 async def upload_files():
     original_img_path_list = []
     for fileItem in imageUploadList:
         basename, ext = os.path.splitext(fileItem.name)
         if ext.lower() not in [".psd", ".jpg", ".jpeg", ".png"]:
             return
-        await upload_single_image(fileItem, basename, original_img_path_list)
+        await upload_single_image_and_save_to_list(
+            fileItem, basename, original_img_path_list
+        )
     return original_img_path_list
 
 
-def save_image(imageList):
+def save_image(image):
+    print("image", image)
+    path = image[0]
+    my_image = Image.open(path)
+    my_stream = io.BytesIO()
+    my_image.save(my_stream, format="PNG")
+    binary_fc = open(path, "rb").read()
+    base64_utf8_str = base64.b64encode(binary_fc).decode("utf-8")
+    basename, ext = os.path.splitext(path)
+    dataurl = f"data:image/{ext};base64,{base64_utf8_str}"
+    print(basename)
+    return [f"img{basename}", dataurl]
+
+
+def save_images(imageList):
     returnList = []
     for image in imageList:
         path = image[0]

public/scripts/models/image-upload.js

@@ -20,6 +20,8 @@ class ImageUpload {
   signedData = null;
   readState = ReadState.ReadyForRead;
   message = null;
+  ulid = null;
+  previewUrl = null;
 
   loadDimensionPromise = null;
 

public/scripts/preview_worker.js

@@ -0,0 +1,69 @@
+importScripts("https://cdn.jsdelivr.net/pyodide/v0.23.4/full/pyodide.js");
+
+async function initianPyodide() {
+  console.log("start startup");
+  const pyodide = await loadPyodide();
+  await pyodide.loadPackage(["numpy", "opencv-python", "pillow", "micropip"]);
+  let zipResponse = await fetch("/mockup.zip");
+  let zipBinary = await zipResponse.arrayBuffer();
+  pyodide.unpackArchive(zipBinary, "zip");
+  await pyodide.runPythonAsync(
+    `
+      from pyodide.http import pyfetch
+      response = await pyfetch("/image_process.py")
+      with open("./image_process.py", "wb") as f:
+        f.write(await response.bytes())
+    `,
+    (output) => console.log(output),
+    (output) => console.log(output),
+  );
+  console.log("end up");
+  return pyodide;
+}
+
+// Now only the first orientation model is generated for preview
+async function runPreviewMockup(pyodide) {
+  let pythonNamespace = pyodide.globals.get("dict")();
+  await pyodide.runPythonAsync(
+    `
+      import mockup
+      import image_process
+      from js import locationKey, imageUpload, deviceInfo, deviceId
+      origin_image_path = await image_process.upload_file()
+      print("start preview", origin_image_path)
+      output_img = await mockup.previewMockup(locationKey, deviceId, origin_image_path, deviceInfo)
+    `,
+    { globals: pythonNamespace },
+  );
+  pyodide.runPython(
+    `
+        temp = image_process.save_image(output_img)
+    `,
+    { globals: pythonNamespace },
+  );
+  return pythonNamespace.get("temp").toJs();
+}
+
+async function main() {
+  let pyodideObject = initianPyodide();
+  self.onmessage = async (event) => {
+    pyodideObject = await pyodideObject;
+
+    self["imageUploadList"] = undefined;
+    self["imageUpload"] = event.data.imageUpload;
+    self["locationKey"] = event.data.location;
+    self["deviceId"] = event.data.deviceId;
+    self["deviceInfo"] = event.data.deviceInfo;
+
+    try {
+      // TODO: Handle preview loading state in widget
+      let results = await runPreviewMockup(pyodideObject);
+      console.log("preview results", results);
+      self.postMessage(results);
+    } catch (error) {
+      self.postMessage({ error: error.message });
+    }
+  };
+}
+
+main();