feature(flask): 增加后端测试项

.idea/cellpose-web.iml

@@ -1,7 +1,7 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <module type="PYTHON_MODULE" version="4">
   <component name="NewModuleRootManager">
-    <content url="file://$MODULE_DIR$" />
+    <content url="file://$MODULE_DIR$/../cellpose-web" />
     <orderEntry type="jdk" jdkName="cellpose" jdkType="Python SDK" />
     <orderEntry type="sourceFolder" forTests="false" />
   </component>

.idea/modules.xml

@@ -2,7 +2,7 @@
 <project version="4">
   <component name="ProjectModuleManager">
     <modules>
-      <module fileurl="file://$PROJECT_DIR$/.idea/cellpose-web-backend.iml" filepath="$PROJECT_DIR$/.idea/cellpose-web-backend.iml" />
+      <module fileurl="file://$PROJECT_DIR$/.idea/cellpose-web.iml" filepath="$PROJECT_DIR$/.idea/cellpose-web.iml" />
     </modules>
   </component>
 </project>

cp_run.py

@@ -3,10 +3,20 @@ from cellpose.io import imread, save_masks
 from PIL import Image
 import numpy as np
 import os, datetime
+from typing import Literal
 
-def test():
+from sympy import false
+
+
+class Cprun:
+    # def __init__(self, model: str | Literal["cpsam"], images: list[str] | str):
+    #     self.model = model
+    #     self.images = images
+
+    @classmethod
+    def run_test(cls):
         model = models.CellposeModel(gpu=True)
-    files = ['img.png']
+        files = ['test_tif/img.png']
         imgs = [imread(f) for f in files]
         masks, flows, styles = model.eval(
             imgs, flow_threshold=0.4, cellprob_threshold=0.0
@@ -29,3 +39,47 @@ def test():
             rgb = plot.image_to_rgb(img, channels=[0, 0])  # 原图转 RGB
             over = plot.mask_overlay(rgb, masks=mask, colors=None)  # 叠加彩色实例
             Image.fromarray(over).save(base + "_overlay.png")
+
+    @classmethod
+    async def run(cls,
+                  images: list[str] | str | None = None,
+                  time: datetime.datetime | None = None,
+                  model: str | str = "cpsam",
+                  diameter: float | None = None,
+                  flow_threshold: float | float = 0.4,
+                  cellprob_threshold: float | float = 0.0, ):
+
+        if time is None:
+            return [False, "No time received"]
+
+        if images is None:
+            return [False, "No images received"]
+
+        message = [f"Using {model} model"]
+
+        model = models.CellposeModel(gpu=True, model_type=model)
+        files = images
+        imgs = [imread(f) for f in files]
+        masks, flows, styles = model.eval(
+            imgs,
+            flow_threshold=flow_threshold,
+            cellprob_threshold=cellprob_threshold,
+            diameter=diameter
+        )
+
+        ts = datetime.datetime.now().strftime("%Y-%m-%d-%H-%M-%S")
+        outdir = os.path.join(os.path.dirname(__file__), "run_output", ts)
+        os.makedirs(outdir, exist_ok=True)  # 自动创建目录
+        for img, mask, flow, name in zip(imgs, masks, flows, files):
+            base = os.path.join(outdir, os.path.splitext(os.path.basename(name))[0])
+            # 使用内置绘图生成蒙版
+            out = base + "_output"
+            save_masks(imgs, mask, flow, out, tif=True)
+
+            # 用 plot 生成彩色叠加图（不依赖 skimage）
+            rgb = plot.image_to_rgb(img, channels=[0, 0])  # 原图转 RGB
+            over = plot.mask_overlay(rgb, masks=mask, colors=None)  # 叠加彩色实例
+            Image.fromarray(over).save(base + "_overlay.png")
+
+        message.append(f"Output saved to: {outdir}")
+        return [True, message]

flaskApp.py

@@ -0,0 +1,40 @@
+from flask import Flask, send_from_directory, request, jsonify
+import os, shutil, time, threading
+from werkzeug.utils import secure_filename
+
+app = Flask(__name__)
+UPLOAD_DIR = "./uploads"
+os.makedirs(UPLOAD_DIR, exist_ok=True)
+
+def run_flask():
+    app.run(host="10.147.18.141", port=5000)
+
+@app.route("/")
+def index():
+    return "<h1>Hello</h1><p>This is the backend of our cellpose server, please visit our website.</p>"
+
+@app.route("/testdl")
+def test_download():
+    return send_from_directory("./test_output/2025-09-16-20-03-51", "img_overlay.png", as_attachment=True)
+
+@app.route("/dl/<timestamp>")
+def download(timestamp):
+    input_dir = os.path.join("./output", timestamp)
+    output_dir = os.path.join("./output/tmp", timestamp)  # 不要加 .zip，make_archive 会自动加
+    os.makedirs("./output/tmp", exist_ok=True)  # 确保 tmp 存在
+    shutil.make_archive(output_dir, 'zip', input_dir)
+    print(f"压缩完成: {output_dir}.zip")
+    return send_from_directory("./output/tmp", f"{timestamp}.zip", as_attachment=True)
+
+
+@app.post("/upload")
+def upload():
+    files = request.files.getlist("files")  # ← 前端用同一个键名多次 append
+    saved = []
+    for f in files:
+        if not f or f.filename == "":
+            continue
+        name = secure_filename(f.filename)
+        f.save(os.path.join(UPLOAD_DIR, name))
+        saved.append(name)
+    return jsonify({"ok": True, "count": len(saved), "files": saved})

main.py

@@ -1,4 +1,10 @@
-import cp_run
+from cp_run import Cprun
+from flaskApp import run_flask
+from multiprocessing import Process
+
 
 if __name__ == "__main__":
-    cp_run.test()
+    # Cprun.run_test()
+    p = Process(target=run_flask)
+    p.start()
+    print(f"Flask running in PID {p.pid}")