Example traffic app

.github/workflows/ci_linux.yaml

@@ -29,7 +29,7 @@ jobs:
       run: sudo apt-get update
 
     - name: Get apt dependencies
-      run: sudo apt-get install protobuf-compiler
+      run: sudo apt-get install protobuf-compiler libgtk-3-dev libasound2-dev libudev-dev
 
     - name: Setup rust
       run: rustup default ${{ matrix.rust-version }}

Cargo.toml

@@ -33,6 +33,7 @@ features = ["maximal"]
 [workspace.dependencies]
 anyhow = "1.0"
 axum = { version = "0.8.4", default-features = false }
+bevy = "0.16"
 bevy_app = "0.16"
 bevy_derive = "0.16"
 bevy_diagnostic = "0.16"
@@ -42,6 +43,7 @@ bevy_time = "0.16"
 bevy_utils = "0.16"
 clap = { version = "4.5.23", features = ["derive"] }
 futures-concurrency = "7.7"
+glam = "0.29.3"
 mime_guess = "2.0.5"
 schemars = "1.2.0"
 serde = "1.0.219"
@@ -57,6 +59,7 @@ tonic-prost = "0.14"
 tonic-prost-build = "0.14"
 prost-build = "0.14"
 prost-reflect = "0.16"
+rand = "0.9.0"
 tracing = "0.1.41"
 tracing-subscriber = "0.3.19"
 futures = "0.3.31"
@@ -212,6 +215,8 @@ members = [
   "examples/diagram/calculator_ops_catalog",
   "examples/native",
   "examples/handbook_snippets",
+  "examples/diagram/traffic_app",
+  "examples/diagram/traffic_ops_catalog",
 ]
 
 [[bin]]

diagram-editor/server/basic_executor.rs

@@ -19,7 +19,8 @@ use crate::{ServerOptions, new_router};
 use bevy_app::App;
 use clap::Parser;
 use crossflow::{
-    CrossflowExecutorApp, Diagram, DiagramError, Outcome, RequestExt, RunCommandsOnWorldExt,
+    CrossflowExecutorApp, CrossflowPlugin, Diagram, DiagramError, Outcome, RequestExt,
+    RunCommandsOnWorldExt,
 };
 use std::thread;
 use std::{fs::File, str::FromStr};
@@ -60,12 +61,59 @@ pub struct RunArgs {
     request: String,
 }
 
-#[derive(Parser, Debug)]
+#[derive(Parser, Debug, Clone, Copy)]
 pub struct ServeArgs {
     #[arg(short, long, default_value_t = 3000)]
     port: u16,
 }
 
+impl Default for ServeArgs {
+    fn default() -> Self {
+        Self { port: 3000 }
+    }
+}
+
+#[derive(Default, Clone, Copy)]
+pub enum PluginSelection {
+    /// Include the [`CrossflowExecutorApp`] plugins. Use this if you will not
+    /// be adding any Bevy plugins yourself.
+    #[default]
+    App,
+    /// Only include the basic [`CrossflowPlugin`]. Use this if you will be
+    /// adding the standard Bevy plugins yourself.
+    Minimal,
+}
+
+#[derive(Default)]
+pub struct CustomRun {
+    pub plugins: PluginSelection,
+    pub args: Option<Args>,
+}
+
+impl From<()> for CustomRun {
+    fn from(_: ()) -> Self {
+        Default::default()
+    }
+}
+
+impl From<Args> for CustomRun {
+    fn from(args: Args) -> Self {
+        CustomRun {
+            plugins: Default::default(),
+            args: Some(args),
+        }
+    }
+}
+
+impl From<PluginSelection> for CustomRun {
+    fn from(plugins: PluginSelection) -> Self {
+        CustomRun {
+            plugins,
+            args: Default::default(),
+        }
+    }
+}
+
 pub fn headless(
     args: RunArgs,
     setup: impl FnOnce() -> BasicExecutorSetup + 'static,
@@ -111,6 +159,43 @@ pub async fn serve(
         app.run()
     });
 
+    axum_serve(router_receiver, args).await
+}
+
+pub fn custom_serve(
+    plugins: PluginSelection,
+    args: ServeArgs,
+    setup: impl FnOnce() -> BasicExecutorSetup + 'static,
+) -> Result<(), Box<dyn Error>> {
+    println!("Serving diagram editor at http://localhost:{}", args.port);
+
+    let BasicExecutorSetup { mut app, registry } = setup();
+    // If WinitPlugin is added, add CrossflowPlugin instead of
+    // CrossflowExecutorApp to prevent overlapping plugins
+    match plugins {
+        PluginSelection::App => {
+            app.add_plugins(CrossflowExecutorApp::default());
+        }
+        PluginSelection::Minimal => {
+            app.add_plugins(CrossflowPlugin::default());
+        }
+    }
+
+    let (router_sender, router_receiver) = tokio::sync::oneshot::channel();
+    thread::spawn(move || {
+        let rt = tokio::runtime::Runtime::new().unwrap();
+        let _ = rt.block_on(axum_serve(router_receiver, args));
+    });
+    let router = new_router(&mut app, registry, ServerOptions::default());
+    let _ = router_sender.send(router);
+    app.run();
+    Ok(())
+}
+
+pub async fn axum_serve(
+    router_receiver: tokio::sync::oneshot::Receiver<axum::routing::Router>,
+    args: ServeArgs,
+) -> Result<(), Box<dyn Error>> {
     let router = router_receiver.await?;
 
     let listener = tokio::net::TcpListener::bind(("localhost", args.port))
@@ -186,22 +271,23 @@ impl BasicExecutorSetup {
 ///   structure cannot be moved between threads. This closure will be moved between
 ///   threads so it must have the Send trait.
 pub fn run_custom_setup(
-    args: Option<Args>,
+    settings: impl Into<CustomRun>,
     setup: impl FnOnce() -> BasicExecutorSetup + Send + 'static,
 ) -> Result<(), Box<dyn Error>> {
-    let rt = tokio::runtime::Runtime::new().unwrap();
-    rt.block_on(run_custom_setup_async(args, setup))
+    let CustomRun { args, plugins } = settings.into();
+    let args = args.unwrap_or_else(|| Args::parse());
+    match args.command {
+        Commands::Run(args) => {
+            let rt = tokio::runtime::Runtime::new().unwrap();
+            rt.block_on(async_headless(args, setup))
+        }
+        Commands::Serve(args) => custom_serve(plugins, args, setup),
+    }
 }
 
-/// Run a custom setup of a basic executor asynchronously. For more information,
-/// see [`run_custom_setup`].
-pub async fn run_custom_setup_async(
-    args: Option<Args>,
+pub async fn async_headless(
+    args: RunArgs,
     setup: impl FnOnce() -> BasicExecutorSetup + Send + 'static,
 ) -> Result<(), Box<dyn Error>> {
-    let args = args.unwrap_or_else(|| Args::parse());
-    match args.command {
-        Commands::Run(args) => headless(args, setup),
-        Commands::Serve(args) => serve(args, setup).await,
-    }
+    headless(args, setup)
 }

examples/diagram/traffic_app/.gitignore

@@ -0,0 +1,2 @@
+# Downloaded from Gazebo Fuel by build.rs
+/assets/

examples/diagram/traffic_app/Cargo.toml

@@ -0,0 +1,23 @@
+[package]
+name = "traffic_app"
+version = "0.1.0"
+edition = "2024"
+
+[lib]
+name = "traffic_app"
+
+[[bin]]
+path = "src/main.rs"
+name = "traffic_app"
+
+[dependencies]
+bevy = { workspace = true }
+bevy_ecs = { workspace = true }
+crossflow = { version = "0.0.7", path = "../../..", features = ["diagram", "python"] }
+crossflow_diagram_editor = { path = "../../../diagram-editor", features = ["basic_executor"] }
+rand = { workspace = true }
+traffic_ops_catalog = { path = "../traffic_ops_catalog" }
+
+[build-dependencies]
+ureq = "2"
+zip = { version = "2", default-features = false, features = ["deflate"] }

examples/diagram/traffic_app/README.md

@@ -0,0 +1,88 @@
+# Traffic app example
+
+This is an example that enables users to build workflows via the diagram editor
+and watch how their node connections result in different behaviors in a simple
+traffic simulator. It is designed to support and demonstrate various `crossflow`
+operations via the diagram editor.
+
+The simulator is a kinematic simulation of a vehicle driving down a road. The
+compiled nodes are intentionally minimal: inputs resemble vehicle controls
+(throttle and steering) and outputs resemble sensor data (obstacle locations,
+traffic signals, dashboard readings). The actual decision-making logic lives in
+the diagram itself, typically inside a Python script operation that reads
+sensor data out of [Buffers](https://open-rmf.github.io/crossflow-handbook/buffers.html)
+and streams control commands back to the vehicle.
+
+The simulation uses metric units throughout: positions and distances are in
+meters, wheel angles are in degrees, and speeds are shown in km/h as the
+user-facing unit.
+
+## Vehicle controls
+
+Nodes that command the vehicle:
+
+| Node   | Use case    | Input   | Output   |
+| ------ |------------ | ------- | -------- |
+| `set_throttle` | Sets the target speed of the vehicle. Pass in a number for the target speed in km/h, or a dict to set both `target_speed` (km/h) and `max_acceleration` (km/h per second). The vehicle accelerates toward the target speed within its acceleration limit. | `f32` or `ThrottleCommand` | `Result<(), String>` |
+| `steer` | Sets the target angle of the front wheels. Pass in a number for the target turn angle in degrees (positive angles steer left), or a dict to set both `target_turn_angle` and `max_steer_speed` (degrees per second). | `f32` or `SteeringCommand` | `Result<(), String>` |
+
+## Sensors
+
+Continuous service nodes that stream out data about the vehicle and its
+surroundings. Their streams are typically connected to a
+[Buffer](https://open-rmf.github.io/crossflow-handbook/buffers.html) so that a
+script can fetch the newest value on its own schedule:
+
+| Node   | Use case    | Streams   |
+| ------ |------------ | --------- |
+| `dashboard` | Streams the vehicle's dashboard instruments every update. | `speed` (km/h), `steering_wheel` (degrees) |
+| `detect_traffic_signal` | Monitors the upcoming traffic signal via events and streams out changes. | `traffic_signal` (`red`/`yellow`/`green`/`empty`) |
+| `detect_speed_limit` | Streams the speed limit posted by the road sign nearest to the vehicle. | `speed_limit` (km/h) |
+| `detect_obstacles` | Monitors obstacles ahead of the vehicle via query and streams out their positions relative to the vehicle, in meters. | `obstacles` (list of `{x, y}`) |
+| `detect_lane_position` | Streams the vehicle's current x position within the lane, in meters. | `position` (meters) |
+| `detect_stop_request` | A "user cancellation sensor" that emits each time the STOP button in the simulator UI is pressed. Use this to let the user end an active workflow early. | `stop` (elapsed seconds) |
+
+## Controllers
+
+| Node   | Use case    | Input   | Streams   |
+| ------ |------------ | ------- | --------- |
+| `lane_controller` | Continuously steers the vehicle toward a target x position within the lane. The target is read from a `ScriptMessage` buffer via [buffer access](https://open-rmf.github.io/crossflow-handbook/buffer-access.html), so a script can update the target while the controller runs. Its steering commands are streamed out and typically connected to the `steer` node. Optionally configure the controller gains (`err_gain`, `dir_gain`, `max_yaw`). | `((), BufferKey<ScriptMessage>)` | `steer` (degrees) |
+
+## Example workflows
+
+Ready-made JSON workflows live in `traffic_app/diagrams/`. Each one carries a
+description and input examples, and they are worth exploring in this order:
+
+| Workflow | What it demonstrates | Input |
+| -------- | -------------------- | ----- |
+| `drive.json` | The simplest possible workflow: set the throttle and terminate. The vehicle keeps driving at the target speed. | Target speed in km/h, e.g. `10` |
+| `donuts.json` | [Split](https://open-rmf.github.io/crossflow-handbook/parallelism.html#split) and [Join](https://open-rmf.github.io/crossflow-handbook/join.html) operations routing one input to both vehicle controls, which makes the vehicle spin in circles. | e.g. `{"throttle": 20, "steer": -45}` |
+| `stoplight.json` | A Python script control loop that fetches the latest traffic signal from a buffer and stops the vehicle at red lights. | Duration in seconds, e.g. `30` |
+| `stoplight_and_obstacles.json` | The same control loop extended to also brake for obstacles ahead of the vehicle. | Duration in seconds, e.g. `30` |
+| `speed_limit.json` | A control loop that follows the speed limit posted on road signs as they pass by. | Duration in seconds, e.g. `60` |
+| `change_lane.json` | Splitting responsibilities between the diagram and compiled nodes: a script decides which lane to drive in and streams the target into a buffer, while the `lane_controller` node steers toward it. | Duration in seconds, e.g. `60` |
+
+All of the timed workflows also connect a `detect_stop_request` sensor, so you
+can press the STOP button in the simulator's user panel to end the trip early.
+
+Try experimenting with the various settings, such as buffer sizes and fetch
+types (clone vs. pull), or edit the scripts and controller gains to see how
+they affect the vehicle's behavior.
+
+## Try it out!
+
+From the current directory, run
+
+```bash
+cargo run -- serve
+```
+
+The first build downloads the app's sprite and font assets from
+[Gazebo Fuel](https://app.gazebosim.org/Open-RMF/fuel/models/crossflow_traffic_app_assets)
+into `assets/`, so it needs an internet connection; later builds reuse the
+downloaded files.
+
+Then open http://localhost:3000 to run the diagram editor app from your web
+browser. Load one of the workflows from `traffic_app/diagrams/`, click
+`Run Workflow`, enter an input (each workflow's input examples are listed in
+its side panel), and watch the vehicle react in the simulator window.