Get video streaming actually working

README.org

@@ -41,7 +41,9 @@ Options:
       --ssh-port <ssh-port>            SSH Port used by the reMarkable tablet (default: 22)
       --ssh-key <ssh-key>              Private SSH key file path
       --tcp-port <tcp-port>            TCP port for video stream (default: 6680)
+      --framerate <framerate>          Framerate (default: 120)
       --dark-mode                      Dark mode - invert colors (default: false)
+      --show-cursor                    Show cursor (default: false)
   -h, --help                           Print help
   -V, --version                        Print version
 #+end_src
@@ -53,7 +55,9 @@ Corresponding keys:
 - =REMARKABLE_SSH_PORT=
 - =REMARKABLE_SSH_KEY_PATH=
 - =REMARKABLE_TCP_PORT=
+- =REMARKABLE_FRAMERATE=
 - =REMARKABLE_DARK_MODE=
+- =REMARKABLE_SHOW_CURSOR=
 
 * Building
 

client/src/config.rs

@@ -5,19 +5,7 @@ use std::{
 
 use anyhow::{Context, Error};
 use clap::Parser;
-use gstreamer_video::VideoFormat;
-
-// TODO: get this info from reMarkable tablet directly
-// first parse firmware version (on server)
-// second set all settings accordingly
-// third send out height, width, pixel format, and bytes per pixel to client
-pub const HEIGHT: u32 = 1872;
-pub const WIDTH: u32 = 1404;
-pub const PIXEL_FORMAT: &str = "bgra";
-pub const VIDEO_FORMAT: VideoFormat = VideoFormat::Bgra;
-pub const BYTES_PER_PIXEL: u32 = 4;
-pub const FILE: &str = ":mem:";
-pub const SKIP_OFFSET: usize = 2629636;
+use review_server::config::ServerOptions;
 
 const DEFAULT_IP: &str = "10.11.99.1";
 const DEFAULT_SSH_PORT: u16 = 22;
@@ -42,18 +30,28 @@ pub struct CliOptions {
     #[arg(long, name = "tcp-port")]
     tcp_port: Option<u16>,
 
+    /// Framerate (default: 120)
+    #[arg(long, name = "framerate")]
+    framerate: Option<f32>,
+
     /// Dark mode - invert colors (default: false)
     #[arg(long, name = "dark-mode")]
     dark_mode: bool,
+
+    /// Show cursor (default: false)
+    #[arg(long, name = "show-cursor")]
+    show_cursor: bool,
 }
 
 #[derive(Debug, Clone)]
 pub struct ClientOptions {
     pub remarkable_ip: String,
     pub ssh_port: u16,
     pub ssh_key: PathBuf,
-    pub tcp_port: u16,
     pub dark_mode: bool,
+    pub tcp_port: u16,
+    pub show_cursor: bool,
+    pub framerate: f32,
 }
 
 impl From<CliOptions> for ClientOptions {
@@ -67,17 +65,46 @@ impl From<CliOptions> for ClientOptions {
             ssh_port: resolve_with(
                 value.ssh_port,
                 "REMARKABLE_SSH_PORT",
-                |string| string.parse().context("could not parse"),
+                |string| {
+                    string
+                        .parse()
+                        .context("could not parse SSH port from environment")
+                },
                 DEFAULT_SSH_PORT,
             ),
             ssh_key: must_resolve_option(value.ssh_key, "REMARKABLE_SSH_KEY_PATH"),
             tcp_port: resolve_with(
                 value.tcp_port,
                 "REMARKABLE_TCP_PORT",
-                |string| string.parse().context("could not parse"),
+                |string| {
+                    string
+                        .parse()
+                        .context("could not parse TCP port from environment")
+                },
                 DEFAULT_TCP_PORT,
             ),
+            framerate: resolve_with(
+                value.framerate,
+                "REMARKABLE_FRAMERATE",
+                |string| {
+                    string
+                        .parse()
+                        .context("could not parse framerate from environment")
+                },
+                120.,
+            ),
             dark_mode: resolve_boolean_option(value.dark_mode, "REMARKABLE_DARK_MODE", false),
+            show_cursor: resolve_boolean_option(value.show_cursor, "REMARKABLE_SHOW_CURSOR", false),
+        }
+    }
+}
+
+impl Into<ServerOptions> for ClientOptions {
+    fn into(self) -> ServerOptions {
+        ServerOptions {
+            port: self.tcp_port,
+            show_cursor: self.show_cursor,
+            framerate: self.framerate,
         }
     }
 }

client/src/display/mod.rs

@@ -5,8 +5,9 @@ use std::{io::Read as _, thread::sleep, time::Duration};
 use anyhow::{Context, Error};
 use futures::stream::StreamExt;
 use gstreamer_app::AppSrc;
-use lz4_flex::frame::FrameDecoder;
-use review_server::config::CommunicatedConfig;
+use gstreamer_video::VideoFormat;
+use lz4_flex::decompress_size_prepended;
+use review_server::config::{CommunicatedConfig, PixelFormat};
 use tokio::net::TcpStream;
 use tokio_util::codec::{Framed, LengthDelimitedCodec};
 use tracing::{debug, info};
@@ -23,63 +24,74 @@ pub async fn gstreamer_thread(opts: ClientOptions) -> Result<(), Error> {
         .await
         .context("could not connect to TCP stream")?;
 
-    let (stream, communicated_config) = get_communicated_config(stream)
+    let mut framed_stream = Framed::new(stream, LengthDelimitedCodec::new());
+    let communicated_config = get_communicated_config(&mut framed_stream)
         .await
         .context("could not get communicated config from TCP stream")?;
 
     debug!("received communicated config: {:?}", &communicated_config);
 
-    let stream = stream
-        .into_std()
-        .context("could not convert stream into std")?;
-
-    let mut decoded_video_data = FrameDecoder::new(stream);
-
-    let (pipeline, appsrc) = build_pipeline().context("could not build gstreamer pipeline")?;
+    let (pipeline, appsrc) =
+        build_pipeline(&communicated_config).context("could not build gstreamer pipeline")?;
     pipeline
         .set_state(gstreamer::State::Playing)
         .context("could not start playing gstreamer pipeline")?;
 
-    let mut buffer = vec![0u8; (BYTES_PER_PIXEL * HEIGHT * WIDTH) as usize];
     loop {
-        let n = decoded_video_data
-            .read(&mut buffer)
+        debug!("attempting to read data from TCP stream");
+
+        let compressed_frame = framed_stream
+            .next()
+            .await
+            .context("TCP stream was closed")?
             .context("could not read from TCP stream")?;
 
-        let slice = buffer[..n].to_vec();
+        debug!(
+            "read one compressed frame from TCP stream ({} bytes)",
+            compressed_frame.len(),
+        );
+
+        let frame = decompress_size_prepended(&compressed_frame)
+            .context("could not decompress received frame")?;
 
-        debug!("read {} bytes:\n\n{:?}", n, &slice);
+        debug!("decompressed: {} bytes", frame.len());
 
-        let buffer = gstreamer::Buffer::from_slice(slice);
+        let buffer = gstreamer::Buffer::from_mut_slice(frame);
         appsrc
             .push_buffer(buffer)
             .context("could not push buffer to app source")?;
-
-        sleep(Duration::from_secs_f64(1.));
     }
 }
 
 async fn get_communicated_config(
-    tcp_stream: TcpStream,
-) -> Result<(TcpStream, CommunicatedConfig), Error> {
-    let mut framed_stream = Framed::new(tcp_stream, LengthDelimitedCodec::new());
-
+    framed_stream: &mut Framed<TcpStream, LengthDelimitedCodec>,
+) -> Result<CommunicatedConfig, Error> {
     let config_bytes = framed_stream
         .next()
         .await
         .context("received None as config bytes")?
         .context("could not receive config bytes")?;
 
-    let config = bson::deserialize_from_slice(&config_bytes)
-        .context("could not deserialize config from bytes")?;
+    bson::deserialize_from_slice(&config_bytes).context("could not deserialize config from bytes")
+}
 
-    Ok((framed_stream.into_inner(), config))
+fn to_video_format(pixel_format: &PixelFormat) -> VideoFormat {
+    match pixel_format {
+        PixelFormat::Rgb565le => todo!("not sure what the video format for RGB 565 LE is"),
+        PixelFormat::Gray8 => VideoFormat::Gray8,
+        PixelFormat::Gray16be => VideoFormat::Gray16Be,
+        PixelFormat::Bgra => VideoFormat::Bgra,
+    }
 }
 
-fn build_pipeline() -> Result<(Pipeline, AppSrc), Error> {
-    let video_info = gstreamer_video::VideoInfo::builder(VIDEO_FORMAT, WIDTH, HEIGHT)
-        .build()
-        .context("could not build video info")?;
+fn build_pipeline(communicated_config: &CommunicatedConfig) -> Result<(Pipeline, AppSrc), Error> {
+    let video_info = gstreamer_video::VideoInfo::builder(
+        to_video_format(&communicated_config.video_config.pixel_format),
+        communicated_config.video_config.width as u32,
+        communicated_config.video_config.height as u32,
+    )
+    .build()
+    .context("could not build video info")?;
 
     let appsrc = gstreamer_app::AppSrc::builder()
         .caps(

client/src/start/mod.rs

@@ -33,12 +33,7 @@ pub async fn start_server(
 > {
     let (stdout_tx, stdout_rx) = mpsc::channel(10);
 
-    let server_options = ServerOptions {
-        port: 6680,
-        show_cursor: false,
-        framerate: 10,
-    };
-
+    let server_options: ServerOptions = opts.into();
     let restream_command = Box::leak(Box::new(format!(
         "RUST_LOG=trace ./review-server '{}'",
         serde_json::to_string(&server_options)

dagger/main.go

@@ -34,7 +34,10 @@ func (m *ReView) CheckAndTestAll(ctx context.Context, source *dagger.Directory)
 	return "ok", nil
 }
 
-func (m *ReView) BuildClient(source *dagger.Directory) *dagger.File {
+func (m *ReView) BuildClient(
+	// +ignore=["target"]
+	source *dagger.Directory,
+) *dagger.File {
 	return linuxContainer(source).
 		WithExec([]string{
 			"cargo", "build", "--release",
@@ -59,17 +62,19 @@ func linuxContainer(source *dagger.Directory) *dagger.Container {
 		WithWorkdir("/source").
 
 		// Cache
-		WithMountedCache("/cache/cargo", dag.CacheVolume("rust-packages")).
-		WithEnvVariable("CARGO_HOME", "/cache/cargo").
+		WithMountedCache("/root/.cargo/registry", dag.CacheVolume("rust-packages")).
 		WithMountedCache("target", dag.CacheVolume("rust-target"))
 }
 
-func (m *ReView) BuildServer(source *dagger.Directory) *dagger.File {
+func (m *ReView) BuildServer(
+	// +ignore=["target"]
+	source *dagger.Directory,
+) *dagger.File {
 	return toltecContainer(source).
 		WithExec([]string{
 			"bash", "-c",
 			"source /opt/x-tools/switch-arm.sh; " +
-				"cargo build --release --bin review-server --target " + RemarkableTarget,
+				"cargo build --release --bin review-server",
 		}).
 		WithExec(
 			[]string{"cp", "target/" + RemarkableTarget + "/release/review-server", "review-server"},
@@ -83,7 +88,9 @@ func toltecContainer(source *dagger.Directory) *dagger.Container {
 
 		// Sources
 		WithDirectory("/source", source).
-		WithWorkdir("/source")
+		WithWorkdir("/source").
 
-	// Sadly caching breaks compile :(
+		// Cache
+		WithMountedCache("/root/.cargo/registry", dag.CacheVolume("rust-packages-toltec")).
+		WithMountedCache("target", dag.CacheVolume("rust-target-toltec"))
 }

server/src/config/mod.rs

@@ -1,10 +1,11 @@
 use std::path::PathBuf;
 
-use crate::version::{ConfigVersion, FirmwareVersion, HardwareVersion, get_config_version};
 use anyhow::{Context, Error, anyhow};
 use clap::Parser;
 use serde::{Deserialize, Serialize};
 
+use crate::version::{ConfigVersion, FirmwareVersion, HardwareVersion, get_config_version};
+
 #[derive(Parser, Debug)]
 #[command(author, version)]
 pub struct CliOptions {
@@ -16,7 +17,7 @@ pub struct CliOptions {
 pub struct ServerOptions {
     pub port: u16,
     pub show_cursor: bool,
-    pub framerate: u8,
+    pub framerate: f32,
 }
 
 impl TryFrom<CliOptions> for ServerOptions {