use pyo3::types::PyString;
use serde::Serialize;
use serde_json::{to_value, Value};
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use tokio::runtime::Runtime;
use tokio::sync::{mpsc, oneshot};
use tokio::sync::Mutex as AsyncMutex;
use uuid::Uuid;
use pyo3::{prelude::*, IntoPyObjectExt};
#[derive(Debug, Clone)]
pub struct PythonData {
pub id: String,
pub method: String,
pub args: Value,
}
#[derive(Debug)]
pub struct APIRequest {
pub data: PythonData,
pub responder: oneshot::Sender<Value>,
}
#[derive(Clone, Debug)]
pub struct APIConnector {
sender: mpsc::Sender<APIRequest>,
_pending: Arc<Mutex<HashMap<String, oneshot::Sender<Value>>>>,
receiver: Arc<AsyncMutex<mpsc::Receiver<APIRequest>>>,
runtime: Arc<Runtime>,
}
impl APIConnector {
pub fn new() -> Self {
let runtime = Arc::new(Runtime::new().unwrap());
let (tx, rx) = mpsc::channel::<APIRequest>(100000);
let receiver = Arc::new(AsyncMutex::new(rx));
Self {
sender: tx,
_pending: Arc::new(Mutex::new(HashMap::new())),
receiver,
runtime,
}
}
pub fn send_request<S: Serialize>(&self, method: &str, args: S) -> oneshot::Receiver<Value> {
let id = Uuid::new_v4().to_string();
let (tx, rx) = oneshot::channel();
let args_value = to_value(args).expect("Failed to serialize arguments");
let request = APIRequest {
data: PythonData {
id: id.clone(),
method: method.to_string(),
args: args_value,
},
responder: tx,
};
let _ = self.sender.blocking_send(request);
rx
}
pub fn on(&self, proxy:&tao::event_loop::EventLoopProxy<crate::FrameEvent>) {
let rx_clone = self.receiver.clone();
let main_proxy = proxy.clone();
self.runtime.spawn(async move {
let mut rx = rx_clone.lock().await;
while let Some(req) = rx.recv().await {
if let Err(e) = main_proxy.send_event(crate::FrameEvent::APIRequest {
data: req.data,
responder: Some(req.responder),
}){
eprintln!("Event Loop already closed: {:?}", e);
}
tokio::time::sleep(std::time::Duration::from_millis(200)).await;
}
});
}
}
#[pyclass]
pub struct WindowAPI {
api: Option<APIConnector>,
}
#[pymethods]
impl WindowAPI {
#[new]
pub fn new() -> Self {
Self { api: None }
}
#[pyo3(name = "emit_event")]
fn emit_event_py<'py>(&self, py: Python<'py>, event:String, args:String) -> PyResult<Bound<'py, PyAny>> {
let Some(api) = &self.api else {
return Err(PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(
"API Connector not connected",
));
};
let event = event.to_string();
let args_value: Value = serde_json::from_str(&args)
.map_err(|e| PyErr::new::<pyo3::exceptions::PyValueError, _>(e.to_string()))?;
let api_clone = api.clone();
let fut = async move {
let rx = api_clone.send_request(&event, args_value);
let response = rx.await.map_err(|e| {
PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string())
})?;
Python::with_gil(|py| {
let py_value:Bound<'_, PyString> = match response {
Value::String(s) => PyString::new(py, &s).into(),
_ => {
let json = serde_json::to_string(&response)
.unwrap_or_else(|_| "null".to_string());
PyString::new(py, &json).into()
}
};
let payload = py_value.into_any();
Ok(payload)
})
};
let d = pyo3_async_runtimes::tokio::future_into_py(py, fut)?.into_bound_py_any(py)?;
Ok(d)
}
}
impl WindowAPI {
pub fn conect_api(&mut self, api:APIConnector){
self.api = Some(api)
}
}
#[pyfunction]
pub fn create_webframe(window_connector:&mut WindowAPI)->anyhow::Result<()>{
let proxy= todo!();
let api = APIConnector::new();
window_connector.conect_api(api);
loop {
api.on(proxy);
}
}it should actually look something like this in python domain
import json
from webruntime import create_webframe
class WindowAPI:
def __init__(self):
pass
async def emit_event(self, event:str, args:str):...
window_api = WindowAPI()
async def some_window_command():
args = json.dumps({})
command_result = await window_api.emit_event("window.setTitle", args)
print(command_result)
if __name__ == "__main__":
create_webframe(window_api)My goal is to implement something similar to this
import asyncio
import uuid
from typing import Optional, Any, Dict
class WindowHandel:
def __init__(self):
self.sio = None
self.state_scope = asyncio.Queue()
self.methods_scope = asyncio.Queue()
self._pending_responses: Dict[str, asyncio.Future] = {}
def runtime_scope(self, sio):
"""Connect or update the socket.io server instance."""
self.sio = sio
async def handle_window_response(self, response: dict):
req_id = response.get("id")
if req_id and req_id in self._pending_responses:
future = self._pending_responses.pop(req_id)
if "error" in response:
future.set_exception(Exception(response["error"]))
else:
future.set_result(response.get("result"))
async def endless_state_loop(self):
while True:
task = None
queue_name = None
queue = None
if not self.state_scope.empty():
queue = self.state_scope
queue_name = "state_scope"
elif not self.methods_scope.empty():
queue = self.methods_scope
queue_name = "methods_scope"
if queue:
task = await queue.get()
if task is None:
await asyncio.sleep(0.01)
continue
future = task.pop("future", None)
data = task.get("data", {})
try:
if hasattr(self, "sio") and self.sio:
await self.sio.emit("window_request", data)
except Exception as e:
print(f"[SocketIO Error in {queue_name}] {e}")
if future:
future.set_exception(e)
async def _request(self, method: str, args: dict, scope: bool = True) -> Any:
req_id = str(uuid.uuid4())
future = asyncio.get_event_loop().create_future()
self._pending_responses[req_id] = future
queue = self.state_scope if scope == True else self.methods_scope
await queue.put({
"event": "window_request",
"data": {
"id": req_id,
"method": method,
"args": args
},
"future": future
})
return await future
async def current(self) -> Any:
return await self._request("window.current", {})
async def close(self, id: Optional[int] = None) -> Any:
return await self._request("window.close", {"id": id}, scope=False)
async def list(self) -> Any:
return await self._request("window.list", {})
async def sendMessage(self, message: str, id: int) -> Any:
return await self._request("window.sendMessage", {"message": message, "id": id}, scope=False)
async def setMenu(self, options: Optional[dict] = None, id: Optional[int] = None) -> Any:
return await self._request("window.setMenu", {"options": options, "id": id}, scope=False)
async def hideMenu(self, id: Optional[int] = None) -> Any:
return await self._request("window.hideMenu", {"id": id}, scope=False)
async def showMenu(self, id: Optional[int] = None) -> Any:
return await self._request("window.showMenu", {"id": id}, scope=False)
async def isMenuVisible(self, id: Optional[int] = None) -> Any:
return await self._request("window.isMenuVisible", {"id": id})
async def scaleFactor(self, id: Optional[int] = None) -> Any:
return await self._request("window.scaleFactor", {"id": id})lifetime may not live long enough
returning this value requires that `'1` must outlive `'2`rustc[Click for full compiler diagnostic](rust-analyzer-diagnostics-view:/diagnostic%20message%20[1]?1#file:///c%3A/Users/MalekAli/Desktop/simple_loop/cha/src/runtime_handel.rs)
runtime_handel.rs(123, 27): has type `pyo3::Python<'1>`
runtime_handel.rs(123, 29): return type of closure is Result<pyo3::Bound<'2, pyo3::PyAny>, pyo3::PyErr>
let payload: Bound<'_, PyAny>
Go to [Bound](vscode-file://vscode-app/c:/Users/MalekAli/AppData/Local/Programs/Microsoft%20VS%20Code/resources/app/out/vs/code/electron-sandbox/workbench/workbench.html) | [PyAny](vscode-file://vscode-app/c:/Users/MalekAli/AppData/Local/Programs/Microsoft%20VS%20Code/resources/app/out/vs/code/electron-sandbox/workbench/workbench.html)
implementation of `IntoPyObject` is not general enough
`IntoPyObject<'0>` would have to be implemented for the type `pyo3::Bound<'_, pyo3::PyAny>`, for any lifetime `'0`...
...but `IntoPyObject<'1>` is actually implemented for the type `pyo3::Bound<'1, pyo3::PyAny>`, for some specific lifetime `'1`rustc[Click for full compiler diagnostic](rust-analyzer-diagnostics-view:/diagnostic%20message%20[4]?4#file:///c%3A/Users/MalekAli/Desktop/simple_loop/cha/src/runtime_handel.rs)
