Adding Rust Python Server

Cargo.toml

@@ -5,10 +5,17 @@ edition = "2021"
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
-[workspace]
-
 [dependencies]
 eyre = "0.6.7"
 serde = { version = "1.0", features = ["derive"] }
 serde_yaml = "0.8.23"
 structopt = "0.3.26"
+async-std = { version = "1.10.0", default-features = false, features = [
+	"attributes",
+] }
+zenoh = { git="https://github.com/eclipse-zenoh/zenoh.git" }
+env_logger = "0.9.0"
+tokio = { version="1.17.0", features=["full"]}
+pyo3 = "0.16.1"
+pyo3-asyncio = { version = "0.16", features = ["tokio-runtime", "attributes"] }
+futures = "0.3.12"

app.py

@@ -0,0 +1,3 @@
+async def return_1(x):
+    print(x)
+    return 1

src/lib.rs

@@ -1 +1,2 @@
 pub mod descriptor;
+pub mod server;

src/main.rs

@@ -1,28 +1,64 @@
-use dora_rs::descriptor::Descriptor;
-use eyre::Context;
+use dora_rs::{descriptor::Descriptor, server::start_server};
+use eyre::{Context, ContextCompat};
+use pyo3::prelude::*;
 use std::{fs::File, path::PathBuf};
 use structopt::StructOpt;
 
 #[derive(Debug, Clone, StructOpt)]
-struct Args {
-    file: PathBuf,
+#[structopt(about = "Dora control")]
+enum Command {
+    #[structopt(about = "Print Graph")]
+    Graph { file: PathBuf },
+    #[structopt(about = "Print Runner")]
+    Runner { file: PathBuf },
+    #[structopt(about = "Run Python server")]
+    StartPython { server: String },
 }
 
-fn main() -> eyre::Result<()> {
-    let args = Args::from_args();
-    let descriptor_file = File::open(&args.file).context("failed to open given file")?;
+#[pyo3_asyncio::tokio::main]
+async fn main() -> PyResult<()> {
+    let command = Command::from_args();
+    match command {
+        Command::Graph { file } => {
+            let descriptor_file = File::open(&file)
+                .context("failed to open given file")
+                .unwrap();
 
-    let descriptor: Descriptor =
-        serde_yaml::from_reader(descriptor_file).context("failed to parse given descriptor")?;
-
-    let visualized = descriptor
-        .visualize_as_mermaid()
-        .context("failed to visualize descriptor")?;
-    println!("{visualized}");
-    println!(
-        "Paste the above output on https://mermaid.live/ or in a \
+            let descriptor: Descriptor = serde_yaml::from_reader(descriptor_file)
+                .context("failed to parse given descriptor")
+                .unwrap();
+            let visualized = descriptor
+                .visualize_as_mermaid()
+                .context("failed to visualize descriptor")
+                .unwrap();
+            println!("{visualized}");
+            println!(
+                "Paste the above output on https://mermaid.live/ or in a \
         ```mermaid code block on GitHub to display it."
-    );
+            );
+        }
+        Command::Runner { file } => {
+            let descriptor_file = File::open(&file)
+                .context("failed to open given file")
+                .unwrap();
+
+            let descriptor: Descriptor = serde_yaml::from_reader(descriptor_file)
+                .context("failed to parse given descriptor")
+                .unwrap();
+            let commands = descriptor
+                .print_commands()
+                .context("Failed to generate commands.")
+                .unwrap();
+            println!("{commands}");
+        }
+        Command::StartPython { server } => {
+            let mut server = server.split(":");
+            let file = server.next().context("Server string is empty.").unwrap();
+            let app = server.next().context("No app found").unwrap();
+
+            let _result = start_server(file, app).await;
+        }
+    }
 
     Ok(())
 }

src/server.rs

@@ -0,0 +1,36 @@
+use futures::prelude::*;
+use pyo3::prelude::*;
+use zenoh::config::Config;
+use zenoh::net::protocol::io::SplitBuffer;
+
+pub async fn start_server(file: &str, app: &str) -> eyre::Result<()> {
+    // Subscribe
+    env_logger::init();
+    let config = Config::default();
+    let session = zenoh::open(config).await.unwrap();
+    let mut subscriber = session.subscribe("a").await.unwrap();
+    let identity = initialize(file, app).await.unwrap();
+
+    loop {
+        let data = subscriber.next().await.unwrap().value.payload;
+        let binary = data.contiguous();
+        println!("recieved data");
+
+        let result = Python::with_gil(|py| {
+            let a = (binary.into_py(py),);
+            pyo3_asyncio::tokio::into_future(identity.call(py, a, None).unwrap().as_ref(py))
+        })
+        .unwrap()
+        .await
+        .unwrap();
+    }
+}
+
+pub async fn initialize(file: &str, app: &str) -> PyResult<Py<PyAny>> {
+    Ok(Python::with_gil(|py| {
+        let file = py.import(file).unwrap();
+        // convert Function into a PyObject
+        let identity = file.getattr(app).unwrap();
+        identity.to_object(py)
+    }))
+}