init

server.py

@@ -0,0 +1,100 @@
+import fastapi
+import uvicorn
+import gradio
+import os
+import io
+import numpy
+import base64
+import traceback
+from PIL import Image
+
+
+def extract(data):
+    def check(data):
+        try:
+            image = base64.b64decode(data)
+            image = Image.open(io.BytesIO(image))
+            if image.format == "JPEG":
+                image = numpy.array(image)
+                return image
+        except Exception as e:
+            return False
+
+    images = []
+
+    def check_json(data):
+        if isinstance(data, dict):
+            for value in data.values():
+                check_json(value)
+        elif isinstance(data, list):
+            for item in data:
+                check_json(item)
+        elif isinstance(data, str):
+            image = check(data)
+            if image is not False:
+                images.append(image)
+        else:
+            pass
+
+    check_json(data)
+    return images
+
+
+class Server:
+    def __init__(self):
+        self.images = [Image.fromarray(numpy.random.randint(0, 256, (360, 640, 3), dtype=numpy.uint8), 'RGB') for _ in range(8)]
+
+    def route(self, app):
+        async def error_handler(request: fastapi.Request, exc):
+            return fastapi.responses.PlainTextResponse(status_code=500, content=str(exc))
+
+        app.add_exception_handler(Exception, handler=error_handler)
+            
+        @app.post("/update")
+        async def update(data: dict):
+            for image in extract(data=data):
+                self.images.insert(0, image)
+                self.images = self.images[:8]
+            return fastapi.responses.JSONResponse(status_code=200, content={})
+
+    def webui(self, app, path):
+        with gradio.Blocks() as webui:
+            with gradio.Row():
+                image_0 = gradio.Image(type="pil", streaming=True)
+                image_1 = gradio.Image(type="pil", streaming=True)
+                image_2 = gradio.Image(type="pil", streaming=True)
+                image_3 = gradio.Image(type="pil", streaming=True)
+            with gradio.Row():
+                image_4 = gradio.Image(type="pil", streaming=True)
+                image_5 = gradio.Image(type="pil", streaming=True)
+                image_6 = gradio.Image(type="pil", streaming=True)
+                image_7 = gradio.Image(type="pil", streaming=True)
+            update = gradio.Button(value="update")
+
+            @update.click(inputs=[], outputs=[image_0, image_1, image_2, image_3, image_4, image_5, image_6, image_7])
+            def update():
+                while True:
+                    yield (gradio.update(value=self.images[0]), 
+                            gradio.update(value=self.images[1]), 
+                            gradio.update(value=self.images[2]), 
+                            gradio.update(value=self.images[3]), 
+                            gradio.update(value=self.images[4]), 
+                            gradio.update(value=self.images[5]), 
+                            gradio.update(value=self.images[6]), 
+                            gradio.update(value=self.images[7]))
+
+        gradio.mount_gradio_app(app, webui, path=path)
+        
+    def run(self, SERVER_HOST, SERVER_PORT):
+        app = fastapi.FastAPI()
+        self.route(app=app)
+        self.webui(app=app, path="/")
+        uvicorn.run(app, host=SERVER_HOST, port=int(SERVER_PORT))
+
+
+if __name__ == "__main__":
+    server = Server()
+    server.run(
+        SERVER_HOST=os.getenv("SERVER_HOST", "0.0.0.0"), 
+        SERVER_PORT=os.getenv("SERVER_PORT", 60000),
+    )

test.ipynb

@@ -0,0 +1,91 @@
+{
+ "cells": [
+  {
+   "cell_type": "code",
+   "execution_count": 68,
+   "metadata": {},
+   "outputs": [
+    {
+     "data": {
+      "text/plain": [
+       "<module 'server' from '/media/ict/hky3535/data/chinatelecom/pypi/b64-viewer/server.py'>"
+      ]
+     },
+     "execution_count": 68,
+     "metadata": {},
+     "output_type": "execute_result"
+    }
+   ],
+   "source": [
+    "import numpy\n",
+    "import base64\n",
+    "import io\n",
+    "import time\n",
+    "import requests\n",
+    "from PIL import Image, ImageDraw, ImageFont\n",
+    "\n",
+    "import server\n",
+    "import importlib\n",
+    "importlib.reload(server)"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 103,
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "<Response [200]> {}\n"
+     ]
+    }
+   ],
+   "source": [
+    "font = ImageFont.load_default()\n",
+    "\n",
+    "def random_image_b64(width=640, height=480):\n",
+    "    image = Image.fromarray(\n",
+    "        numpy.random.randint(0, 1, (height, width, 3), dtype=numpy.uint8), 'RGB')\n",
+    "\n",
+    "    buffer = io.BytesIO()\n",
+    "    image.save(buffer, format=\"JPEG\")\n",
+    "    image_b64 = base64.b64encode(buffer.getvalue()).decode('utf-8')\n",
+    "    return image_b64\n",
+    "\n",
+    "while True:\n",
+    "    image_b64 = random_image_b64()\n",
+    "\n",
+    "    response = requests.post(\n",
+    "        url=\"http://localhost:60000/update\", \n",
+    "        json={\"image\": image_b64, \"image_extra\": image_b64})\n",
+    "    \n",
+    "    print(response, response.json())\n",
+    "\n",
+    "    break"
+   ]
+  }
+ ],
+ "metadata": {
+  "kernelspec": {
+   "display_name": "Python 3",
+   "language": "python",
+   "name": "python3"
+  },
+  "language_info": {
+   "codemirror_mode": {
+    "name": "ipython",
+    "version": 3
+   },
+   "file_extension": ".py",
+   "mimetype": "text/x-python",
+   "name": "python",
+   "nbconvert_exporter": "python",
+   "pygments_lexer": "ipython3",
+   "version": "3.10.12"
+  }
+ },
+ "nbformat": 4,
+ "nbformat_minor": 2
+}