init

Dockerfile

@@ -0,0 +1,22 @@
+# Read the doc: https://huggingface.co/docs/hub/spaces-sdks-docker
+# you will also find guides on how best to write your Dockerfile
+
+FROM python:3.12
+
+RUN useradd -m -u 1000 user
+USER user
+ENV PATH="/home/user/.local/bin:$PATH"
+
+WORKDIR /app
+
+COPY --chown=user . /app
+
+RUN pip install --no-cache-dir --upgrade -r /app/requirements.txt
+RUN curl "https://jsonbin.pages.dev/_download/50ZIDESFfSyk7Xbb%3AQvIUprVbtLUWafGlyhnBTHlLQjhsstEgQQcVJ8x%2Fy%2FUKD7s3HOpGKCbt%2B5AByDIwOl13QWmvwWnZlOXP0DyRqUdbtB3AfgOlbX8%3D?share=py" -o /app/iotcore-3.0.3-cp312-cp312-manylinux_2_38_x86_64.whl
+RUN pip install /app/iotcore-3.0.3-cp312-cp312-manylinux_2_38_x86_64.whl
+
+
+RUN python /app/pyx/setup.py build_ext -b /app/
+
+
+CMD ["python", "run_broker.py"]

broker_app.py

@@ -0,0 +1,594 @@
+import asyncio
+from datetime import datetime
+import socket
+from pydantic import BaseModel, Field
+import os
+# 1. IMPORT THE COMPILED CYTHON MODULE
+try:
+    from proxy_core import optimized_ws_to_tcp, optimized_tcp_to_ws
+    print("🚀 Running with CYTHON optimizations")
+except ImportError:
+    print("⚠️ Cython module not found. Run 'python setup.py build_ext --inplace'")
+    # Fallback to Python definitions if compilation failed
+    async def optimized_ws_to_tcp(ws, writer): pass 
+    async def optimized_tcp_to_ws(reader, ws): pass
+
+from fastapi.middleware.cors import CORSMiddleware
+import uuid
+from concurrent_collections import ConcurrentDictionary
+
+from fastapi import FastAPI, WebSocket, WebSocketDisconnect
+from contextlib import asynccontextmanager
+
+# Assuming 'iotcore' is your custom library or wrapper around an MQTT broker
+from iotcore import IotCore 
+
+# --- CRITICAL: Install and use uvloop for high performance ---
+# pip install uvloop
+try:
+    import uvloop
+    asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
+    print("🚀 High-Performance uvloop enabled")
+except ImportError:
+    print("⚠️ uvloop not found. Falling back to standard asyncio (Slower)")
+
+
+
+
+# --- MOCKING IotCore for the fix demonstration (Uncomment your import above) ---
+# class IotCore:
+#     def background_loop_forever(self): pass
+#     def subscribe(self, topic, cb): pass
+#     def publish(self, topic, payload): pass
+#     def accept(self, topic):
+#         def decorator(func): return func
+#         return decorator
+iot = IotCore()
+# ---------------------------------------------------------------------------
+# import logging
+# logging.basicConfig(level=logging.INFO) # This will show Broker logs too
+
+BROKER_PORT = 1883
+BROKER_HOST = '127.0.0.1'
+
+BROKER_APIKEY = os.environ.get("BROKER_APIKEY","BROKER_APIKEY")
+
+CHUNK_SIZE = 65536  # 64KB Read Buffer
+BUFFER_SIZE = 65536
+MESSAGE_LOG_COUNT = 10000
+MESSAGE_COUNT=0
+MESSAGE_START_TIME = None
+
+def on_message(msg):
+    print(f"📩 Recieve [{msg}]")
+def on_message_benchmark(msg):
+    # Only print every 100th message to save CPU
+    COUNT = int(msg.split()[-1])
+    global MESSAGE_COUNT, MESSAGE_START_TIME
+    if MESSAGE_START_TIME is None:
+        MESSAGE_START_TIME = datetime.now()
+    MESSAGE_COUNT += 1
+
+    if MESSAGE_COUNT % MESSAGE_LOG_COUNT == 0:
+        print(f"📩 Recieve [{msg}]")
+        elapsed = (datetime.now() - MESSAGE_START_TIME).total_seconds()
+        
+        mps = float(MESSAGE_COUNT) / float(elapsed) 
+        print(f"📊 Processed {MESSAGE_COUNT} messages in {elapsed:.2f} sec,  {mps:.2f} m/sec") 
+        
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    print("🚀 Starting IoT Broker...")
+    iot.background_loop_forever()
+    
+    # Quick connectivity check
+    try:
+        _, w = await asyncio.open_connection(BROKER_HOST, BROKER_PORT)
+        w.close()
+        await w.wait_closed()
+        print("✅ Internal Broker UP")
+    except:
+        pass
+
+    iot.unsubscribe("iot")
+    iot.subscribe("iot", on_message)
+    # iot.unsubscribe("iot")
+    yield
+    print("🛑 Broker stopping...")
+
+
+app = FastAPI(lifespan=lifespan)
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+async def ws_to_tcp(ws: WebSocket, writer: asyncio.StreamWriter):
+    pending_bytes = 0
+    # Increase from 128KB to 512KB for Localhost testing
+    # Try Increasing the Flush Threshold for localhost. Since localhost is extremely fast, flushing too often (CPU bound) is worse than flushing less often (Memory bound).
+    FLUSH_THRESHOLD = 524288  # 512KB
+
+    try:
+        async for data in ws.iter_bytes():
+            if data:
+                writer.write(data)
+                n = len(data)
+                pending_bytes += n
+                
+                # LOGIC CHANGE:
+                # If the packet is tiny (< 100 bytes), it is likely an MQTT ACK or PINGREQ.
+                # Flush IMMEDIATELY to keep latency low.
+                # If it is big (payload), buffer it to keep bandwidth high.
+                if n < 100 or pending_bytes > FLUSH_THRESHOLD:
+                    await writer.drain()
+                    pending_bytes = 0
+        
+        if pending_bytes > 0:
+            await writer.drain()
+
+    except (WebSocketDisconnect, ConnectionResetError):
+        pass 
+    except Exception as e:
+        print(f"⚠️ WS->TCP Error: {e}")
+
+async def tcp_to_ws(reader: asyncio.StreamReader, ws: WebSocket):
+    """
+    Optimized Reader: Reads large chunks.
+    """
+    try:
+        while True:
+            data = await reader.read(BUFFER_SIZE)
+            if not data:
+                break # EOF from Broker
+            await ws.send_bytes(data)
+    except (RuntimeError, ConnectionResetError):
+        pass # WS closed
+    except Exception as e:
+        print(f"⚠️ TCP->WS Error: {e}")
+
+
+@app.websocket("/mqtt")
+async def mqtt_websocket_proxy_opt(client_ws: WebSocket):
+    
+    # --- FIX FOR BROWSER COMPATIBILITY ---
+    # 1. Connect to Internal Broker
+    try:
+        reader, writer = await asyncio.open_connection(BROKER_HOST, BROKER_PORT)
+        # --- OPTIMIZATION ---
+        # Get the raw socket object
+        sock = writer.get_extra_info('socket')
+        if sock:
+            # Disable Nagle's Algorithm (No Delay)
+            sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
+            
+    except Exception as e:
+        print(f"❌ Internal Broker Down: {e}")
+        await client_ws.close(code=1011)
+        return
+
+    # 2. Negotiate Subprotocol Smartly
+    # Browsers might send: "mqtt", "mqttv3.1", or a list.
+    requested_protocols = client_ws.scope.get('subprotocols', [])
+    print(f"🔄 Client requested subprotocols: {requested_protocols}")
+    
+    # If client asks for 'mqtt', give it 'mqtt'.
+    if "mqtt" in requested_protocols:
+        await client_ws.accept(subprotocol="mqtt")
+    # If client asks for nothing (some default browser behaviors), accept anyway.
+    elif not requested_protocols:
+        await client_ws.accept()
+    else:
+        # Fallback: Just accept the first one they asked for to keep connection alive
+        await client_ws.accept(subprotocol=requested_protocols[0])
+
+
+
+    # 2. USE THE CYTHON FUNCTIONS
+    task_ws = asyncio.create_task(optimized_ws_to_tcp(client_ws, writer))
+    task_tcp = asyncio.create_task(optimized_tcp_to_ws(reader, client_ws))
+
+    await asyncio.wait([task_ws, task_tcp], return_when=asyncio.FIRST_COMPLETED)
+
+    for task in [task_ws, task_tcp]:
+        task.cancel()
+    
+    writer.close()
+    try:
+        await writer.wait_closed()
+    except:
+        pass
+
+@app.websocket("/mqtt_normal")
+async def mqtt_websocket_proxy(client_ws: WebSocket):
+    
+    # --- FIX FOR BROWSER COMPATIBILITY ---
+    # 1. Connect to Internal Broker
+
+    # get reader, writer from connection pool
+
+    try:
+        reader, writer = await asyncio.open_connection(BROKER_HOST, BROKER_PORT)
+        # --- OPTIMIZATION ---
+        # Get the raw socket object
+        sock = writer.get_extra_info('socket')
+        if sock:
+            # Disable Nagle's Algorithm (No Delay)
+            sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
+    except Exception as e:
+        print(f"❌ Internal Broker Down: {e}")
+        await client_ws.close(code=1011)
+        return
+
+    print("✅ Connected to Internal Broker")
+    
+
+    # get id of the client
+    client_id = client_ws.headers.get("sec-websocket-key","unknown")
+    print(f"🔑 Client ID: {client_id}")
+
+
+    # 2. Negotiate Subprotocol Smartly
+    # Browsers might send: "mqtt", "mqttv3.1", or a list.
+    requested_protocols = client_ws.scope.get('subprotocols', [])
+    print(f"🔄 Client requested subprotocols: {requested_protocols}")
+    
+    # If client asks for 'mqtt', give it 'mqtt'.
+    if "mqtt" in requested_protocols:
+        await client_ws.accept(subprotocol="mqtt")
+    # If client asks for nothing (some default browser behaviors), accept anyway.
+    elif not requested_protocols:
+        await client_ws.accept()
+    else:
+        # Fallback: Just accept the first one they asked for to keep connection alive
+        await client_ws.accept(subprotocol=requested_protocols[0])
+
+
+
+    # 3. Create Tasks
+    task_ws_to_tcp = asyncio.create_task(ws_to_tcp(client_ws, writer))
+    task_tcp_to_ws = asyncio.create_task(tcp_to_ws(reader, client_ws))
+
+    # 4. CRITICAL: Wait for FIRST_COMPLETED
+    # If WebSocket dies, we MUST kill the TCP task immediately.
+    # If TCP dies, we MUST kill the WebSocket task immediately.
+    done, pending = await asyncio.wait(
+        [task_ws_to_tcp, task_tcp_to_ws],
+        return_when=asyncio.FIRST_COMPLETED
+    )
+
+    # 5. AGGRESSIVE CLEANUP (Fixes the "Zombie" connection issue)
+    for task in pending:
+        task.cancel()
+    
+    # Close TCP socket immediately so Broker releases the Client ID
+    writer.close()
+    try:
+        await writer.wait_closed()
+    except:
+        pass
+    
+    # print("🔌 Clean Disconnect")
+
+# def main():
+#     import uvicorn
+#     # ws_ping_interval=None  -> Disables sending Pings (Prevents 20s disconnect)
+#     # ws_ping_timeout=None   -> Disables waiting for Pongs (Prevents timeouts)
+#     uvicorn.run(
+#         "broker:app", 
+#         host="127.0.0.1", 
+#         port=8000, 
+#         loop="uvloop",
+#         ws_ping_interval=None, 
+#         ws_ping_timeout=None,
+#         log_level="info"
+#     )
+
+# if __name__ == "__main__":
+#     main()
+
+# --------------------------------------------------------------------------- #
+# Imports added for the bridge fix
+# --------------------------------------------------------------------------- #
+from fastapi import FastAPI, WebSocket, WebSocketDisconnect, HTTPException
+from fastapi.responses import JSONResponse, StreamingResponse
+from pydantic import BaseModel, Field
+from datetime import datetime
+from concurrent_collections import ConcurrentDictionary
+import uuid, asyncio, json
+
+# --------------------------------------------------------------------------- #
+# Pydantic bodies
+# --------------------------------------------------------------------------- #
+class ConnectBody(BaseModel):
+    brokerUrl: str | None = f"mqtt://{BROKER_HOST}:{BROKER_PORT}"
+    username: str | None = None
+    password: str | None = None
+    keepalive: int = 60
+    apiKey: str | None = None
+
+
+class PublishBody(BaseModel):
+    topic: str = "iot"
+    payload: str = "test is send"
+    qos: int = 0
+    retain: bool = False
+
+
+class SubscribeBody(BaseModel):
+    topic: str = "iot"
+    qos: int = 0
+    stream: bool = False               # ← this controls /messages behavior
+
+
+class UnsubscribeBody(BaseModel):
+    topic: str = "iot"
+
+
+class PollBody(BaseModel):
+    topic: str | None = None            # None → all topics
+    limit: int = Field(default=50, ge=1, le=500)
+
+
+# --------------------------------------------------------------------------- #
+# State
+# --------------------------------------------------------------------------- #
+MAX_QUEUED_MESSAGES = 1000
+
+CONNECTION_CLIENTS = ConcurrentDictionary({})
+SUBSCRIPTIONS      = ConcurrentDictionary({})
+
+
+# --------------------------------------------------------------------------- #
+# Helpers
+# --------------------------------------------------------------------------- #
+def _require_client(client_id: str):
+    if CONNECTION_CLIENTS.get(client_id) is None:
+        raise HTTPException(status_code=404, detail="Client not found")
+
+
+def _mqtt_callback(client_id: str, topic: str, msg):
+    """Thread-safe broker callback. Drops silently if client/topic gone."""
+    print(f"✅ Callback: client={client_id}, topic={topic}, msg={msg}")
+    try:
+        if CONNECTION_CLIENTS.get(client_id) is None:
+            print(f"⚠️ callback error (client={client_id}, topic={topic}): invalid clinet")
+            return
+        with CONNECTION_CLIENTS.get_locked(client_id) as v:
+            store = v.get("messages", {}).get(topic)
+            if store is None:
+                print(f"⚠️ callback error (client={client_id}, topic={topic}): invalid store")
+                return
+            queue = store["queue"]
+            if len(queue) >= MAX_QUEUED_MESSAGES:
+                queue.pop(0)
+            queue.append({
+                "topic":     topic,
+                "payload":   msg,
+                "timestamp": datetime.now().isoformat(),
+            })
+    except Exception as e:
+        print(f"⚠️ callback error (client={client_id}, topic={topic}): {e}")
+
+
+def _is_stream_mode(client_id: str, topic: str | None) -> bool:
+    """Check if the requested scope has ANY stream-enabled subscription."""
+    with CONNECTION_CLIENTS.get_locked(client_id) as v:
+        messages = v.get("messages", {})
+        if topic:
+            store = messages.get(topic)
+            return store is not None and store.get("stream", False)
+        return any(s.get("stream", False) for s in messages.values())
+
+
+# --------------------------------------------------------------------------- #
+# POST /clients/connect
+# --------------------------------------------------------------------------- #
+@app.post("/clients/connect")
+async def connect_client(body: ConnectBody):
+    if body.apiKey != BROKER_APIKEY:
+        print("Invalid API Key")
+        raise HTTPException(status_code=401, detail="Invalid API Key")
+
+    client_id = str(uuid.uuid4())
+    stamp = datetime.now()
+
+    CONNECTION_CLIENTS.assign_atomic(client_id, {
+        "created_at":   stamp,
+        "ping_at":      stamp,
+        "ping_timeout": 0,
+        "messages":     {},
+    })
+
+    return {
+        "success":    True,
+        "client_id":  client_id,
+        "created_at": stamp.isoformat(),
+    }
+
+
+# --------------------------------------------------------------------------- #
+# GET /health
+# --------------------------------------------------------------------------- #
+@app.get("/health")
+async def health():
+    return {"status": "ok", "activeClients": len(CONNECTION_CLIENTS)}
+
+
+# --------------------------------------------------------------------------- #
+# POST /clients/{client_id}/ping
+# --------------------------------------------------------------------------- #
+@app.post("/clients/{client_id}/ping")
+async def ping(client_id: str):
+    _require_client(client_id)
+    with CONNECTION_CLIENTS.get_locked(client_id) as v:
+        now = datetime.now()
+        v["ping_timeout"] = (now - v["ping_at"]).total_seconds()
+        v["ping_at"] = now
+    return {"success": True}
+
+
+# --------------------------------------------------------------------------- #
+# POST /clients/{client_id}/disconnect
+# --------------------------------------------------------------------------- #
+@app.post("/clients/{client_id}/disconnect")
+async def disconnect_client(client_id: str):
+    _require_client(client_id)
+    try:
+        with CONNECTION_CLIENTS.get_locked(client_id) as v:
+            for topic in list(v.get("messages", {})):
+                try:
+                    iot.unsubscribe(topic)
+                except Exception:
+                    pass
+    except Exception:
+        pass
+    CONNECTION_CLIENTS.pop(client_id, None)
+    SUBSCRIPTIONS.pop(client_id, None)
+    return {"success": True}
+
+
+# --------------------------------------------------------------------------- #
+# POST /clients/{client_id}/publish
+# --------------------------------------------------------------------------- #
+@app.post("/clients/{client_id}/publish")
+async def publish(client_id: str, body: PublishBody):
+    _require_client(client_id)
+    print(f"ℹ️ Publish {body}")
+    iot.publish(body.topic, body.payload)
+    return {"success": True}
+
+
+# --------------------------------------------------------------------------- #
+# POST /clients/{client_id}/subscribe
+# --------------------------------------------------------------------------- #
+@app.post("/clients/{client_id}/subscribe")
+async def subscribe(client_id: str, body: SubscribeBody):
+    _require_client(client_id)
+    topic = body.topic
+
+    with CONNECTION_CLIENTS.get_locked(client_id) as v:
+        messages = v.setdefault("messages", {})
+        if topic in messages:
+            # update stream flag even if already subscribed
+            messages[topic]["stream"] = body.stream
+            return {"success": True, "topic": topic, "stream": body.stream,
+                    "note": "already subscribed, stream flag updated"}
+        messages[topic] = {"stream": body.stream, "queue": []}
+
+    iot.subscribe(
+        topic,
+        lambda msg, _cid=client_id, _t=topic: _mqtt_callback(_cid, _t, msg),
+    )
+    return {"success": True, "topic": topic, "stream": body.stream}
+
+
+# --------------------------------------------------------------------------- #
+# POST /clients/{client_id}/unsubscribe
+# --------------------------------------------------------------------------- #
+@app.post("/clients/{client_id}/unsubscribe")
+async def unsubscribe(client_id: str, body: UnsubscribeBody):
+    _require_client(client_id)
+    topic = body.topic
+
+    with CONNECTION_CLIENTS.get_locked(client_id) as v:
+        messages = v.get("messages", {})
+        if topic not in messages:
+            raise HTTPException(status_code=404,
+                                detail=f"Not subscribed to '{topic}'")
+        del messages[topic]
+
+    iot.unsubscribe(topic)
+    return {"success": True}
+
+
+# --------------------------------------------------------------------------- #
+# POST /clients/{client_id}/messages   ← SINGLE ENDPOINT, DYNAMIC MODE
+# --------------------------------------------------------------------------- #
+@app.post("/clients/{client_id}/messages")
+async def get_messages(client_id: str, body: PollBody):
+    """
+    • If the topic (or any topic) was subscribed with `stream: true`
+      → returns an **SSE stream** that pushes messages in real-time.
+
+    • Otherwise
+      → returns a **JSON array** of drained messages (classic poll).
+    """
+    _require_client(client_id)
+
+    # ── decide mode from the flag set at subscribe time ──
+    if _is_stream_mode(client_id, body.topic):
+        return _stream_response(client_id, body.topic)
+    else:
+        return _poll_response(client_id, body.topic, body.limit)
+
+
+# ── poll path ──────────────────────────────────────────────────────────────── #
+def _poll_response(client_id: str, topic: str | None, limit: int):
+    collected: list[dict] = []
+
+
+    with CONNECTION_CLIENTS.get_locked(client_id) as v:
+        messages = v.get("messages", {})
+        topics = [topic] if topic else list(messages)
+
+        for t in topics:
+            store = messages.get(t)
+            if store is None:
+                print(f"⚠️ topic:{t}, store is none ")
+                continue
+            queue: list = store["queue"]
+            take = min(len(queue), limit - len(collected))
+            if take <= 0:
+                print(f"⚠️ topic:{t}, take is {take} ")
+                continue
+            collected.extend(queue[:take])
+            del queue[:take]
+
+    return {
+        "success":  True,
+        "mode":     "poll",
+        "count":    len(collected),
+        "messages": collected,
+    }
+
+
+# ── stream path (SSE) ─────────────────────────────────────────────────────── #
+def _stream_response(client_id: str, topic: str | None):
+
+    async def _event_generator():
+        while True:
+            # client gone → stop
+            if CONNECTION_CLIENTS.get(client_id) is None:
+                break
+
+            batch: list[dict] = []
+            with CONNECTION_CLIENTS.get_locked(client_id) as v:
+                messages = v.get("messages", {})
+                targets = [topic] if topic else list(messages)
+                for t in targets:
+                    store = messages.get(t)
+                    if store is None:
+                        continue
+                    # only drain stream-enabled topics
+                    if not store.get("stream", False):
+                        continue
+                    batch.extend(store["queue"])
+                    store["queue"].clear()
+
+            for item in batch:
+                yield f"data: {json.dumps(item)}\n\n"
+
+            if not batch:
+                await asyncio.sleep(0.1)        # 100ms idle backoff
+
+    return StreamingResponse(
+        _event_generator(),
+        media_type="text/event-stream",
+        headers={
+            "Cache-Control":    "no-cache",
+            "X-Accel-Buffering": "no",
+        },
+    )

mqtt.toml

@@ -0,0 +1,22 @@
+id = 0
+
+[router]
+id = 0
+max_connections = 10010
+max_outgoing_packet_count = 200
+max_segment_size = 104857600
+max_segment_count = 10
+
+# Standard TCP Listener (We will proxy to this)
+[v4.1]
+name = "v4-1"
+listen = "127.0.0.1:1883" 
+next_connection_delay_ms = 1
+[v4.1.connections]
+max_payload_size = 20480
+dynamic_filters = true
+connection_timeout_ms = 60000   # Increase to 2 minutes
+max_inflight_count = 1000        # Allow more unacknowledged messages
+
+[console]
+listen = "0.0.0.0:3030"

pyx/proxy_core.pyx

@@ -0,0 +1,51 @@
+# proxy_core.pyx
+# cython: language_level=3
+
+import asyncio
+from cpython.bytes cimport PyBytes_GET_SIZE
+
+# We declare types for the loop variables to speed up math
+async def optimized_ws_to_tcp(object ws, object writer):
+    cdef int pending_bytes = 0
+    cdef int n = 0
+    cdef int FLUSH_THRESHOLD = 131072 # 128KB
+    # Increase from 128KB to 512KB for Localhost testing
+    # Try Increasing the Flush Threshold for localhost. Since localhost is extremely fast, flushing too often (CPU bound) is worse than flushing less often (Memory bound).
+    FLUSH_THRESHOLD = 524288  # 512KB
+    cdef bytes data
+
+    try:
+        # async for is handled by Cython's generator support
+        async for data in ws.iter_bytes():
+            if data:
+                writer.write(data)
+                
+                # C-Level Optimization: Get size instantly
+                n = PyBytes_GET_SIZE(data)
+                pending_bytes += n
+                
+                # C-Level Comparison (No Python Object overhead)
+                if pending_bytes > FLUSH_THRESHOLD or n < 100:
+                    await writer.drain()
+                    pending_bytes = 0
+        
+        if pending_bytes > 0:
+            await writer.drain()
+
+    except Exception:
+        pass
+
+async def optimized_tcp_to_ws(object reader, object ws):
+    cdef int BUFFER_SIZE = 65536
+    cdef bytes data
+
+    try:
+        while True:
+            # We cannot optimize the await call itself (it's I/O)
+            # But we minimize the overhead AROUND it
+            data = await reader.read(BUFFER_SIZE)
+            if not data:
+                break
+            await ws.send_bytes(data)
+    except Exception:
+        pass

pyx/setup.py

@@ -0,0 +1,39 @@
+# setup.py
+
+# numpy header not found https://stackoverflow.com/questions/14657375/cython-fatal-error-numpy-arrayobject-h-no-such-file-or-directory
+# wasm 
+# from wasmpy_build import build_ext
+from setuptools import setup, Extension
+from setuptools import setup
+from Cython.Build import cythonize
+from pathlib import Path
+from distutils.core import setup, Extension
+from Cython.Build import cythonize
+import numpy as np
+
+
+
+
+pwd = Path(__file__).parent
+filepath = str(pwd/"proxy_core.pyx")
+
+for  root, dirs, files in pwd.walk():
+    # print(f"root {root}")
+    # print(f"dirs:{dirs}")
+    # print(f"files:{files}")
+    for file in files:
+        # print(f"file:{file}")
+        if file.endswith(".pyx"):
+            print(f"Compiling Cython file: {root/file}")
+            setup(
+                ext_modules = cythonize(str(root/file),
+                build_dir=str(pwd/"build")),
+                include_dirs=[np.get_include()]
+                )
+            # setup(
+            #     ext_modules=cythonize([
+            #         Extension(file.split(".")[0], [str(root/file)])
+            #         ]),
+            #     cmdclass={"build_ext": build_ext},
+            #     )
+    # setup(ext_modules = cythonize(str(file),build_dir=str(pwd/"build")))

requirements.txt

@@ -0,0 +1,26 @@
+fastapi
+uvicorn
+uvloop
+websockets
+
+gmqtt
+google-genai
+python-dotenv
+jinja2
+PersistDict
+python-multipart
+python-jose[cryptography]
+# passlib
+sqlalchemy
+# opencv-python
+paho-mqtt
+cython
+setuptools
+
+sqlitedict
+concurrent_collections
+
+# mqtt
+# iotcore
+# for pyo3
+maturin

run_broker.py

@@ -0,0 +1,20 @@
+import uvicorn
+import asyncio
+try:
+    import uvloop
+    asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
+except ImportError:
+    pass
+
+if __name__ == "__main__":
+    # ws_ping_interval=None  -> Disables sending Pings (Prevents 20s disconnect)
+    # ws_ping_timeout=None   -> Disables waiting for Pongs (Prevents timeouts)
+    uvicorn.run(
+        "broker_app:app", 
+        host="127.0.0.1", 
+        port=7860, 
+        loop="uvloop",
+        ws_ping_interval=None, 
+        ws_ping_timeout=None,
+        log_level="info"
+    )