From 7a1c37fc8612fe7e7641d7d55b010388a80dccd4 Mon Sep 17 00:00:00 2001 From: "Jan-Peter v. Hunnius" Date: Thu, 4 Jun 2026 08:48:35 +0200 Subject: [PATCH] refactor: TUI multi-feed monitor with binary parsers - Add ratatui/crossterm TUI replacing stdout data dump - Add binary parsers for KSUDP RTCarInfo (port 9996) and Telemetry Tool (port 10101) - Add shared binary_utils module for safe packet reading - Add ring-buffer frequency calculation and feed health state machine - Add parse_first_driver() to avoid unnecessary multi-driver allocation - Add 27 unit tests (14 feed_state + 6 ksudp + 7 telemetry_tool) - Move parsing outside mutex lock to minimize TUI contention - Replace expect() on env var with graceful error handling - Remove unused serde dependency - Delete obsolete stats.rs and telemetry.rs --- Cargo.lock | 464 +++++++++++++++++++++++++++++---- Cargo.toml | 6 +- src/binary_utils.rs | 97 +++++++ src/feed_state.rs | 389 ++++++++++++++++++++++++++++ src/main.rs | 489 ++++++++++++++++++++++++++--------- src/parser_ksudp.rs | 241 +++++++++++++++++ src/parser_telemetry_tool.rs | 363 ++++++++++++++++++++++++++ src/stats.rs | 70 ----- src/telemetry.rs | 56 ---- src/telemetry_frame.rs | 218 ++++++++++++++++ src/ui.rs | 222 ++++++++++++++++ 11 files changed, 2308 insertions(+), 307 deletions(-) create mode 100644 src/binary_utils.rs create mode 100644 src/feed_state.rs create mode 100644 src/parser_ksudp.rs create mode 100644 src/parser_telemetry_tool.rs delete mode 100644 src/stats.rs delete mode 100644 src/telemetry.rs create mode 100644 src/telemetry_frame.rs create mode 100644 src/ui.rs diff --git a/Cargo.lock b/Cargo.lock index ad3a8ef..b1e278f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2,6 +2,12 @@ # It is not intended for manual editing. version = 4 +[[package]] +name = "allocator-api2" +version = "0.2.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "683d7910e743518b0e34f1186f92494becacb047c7b6bf616c96772180fef923" + [[package]] name = "android_system_properties" version = "0.1.5" @@ -35,6 +41,21 @@ version = "1.11.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1e748733b7cbc798e1434b6ac524f0c1ff2ab456fe201501e6497c8417a4fc33" +[[package]] +name = "cassowary" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df8670b8c7b9dae1793364eafadf7239c40d669904660c5960d74cfd80b46a53" + +[[package]] +name = "castaway" +version = "0.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dec551ab6e7578819132c713a93c022a05d60159dc86e7a7050223577484c55a" +dependencies = [ + "rustversion", +] + [[package]] name = "cc" version = "1.2.63" @@ -64,12 +85,97 @@ dependencies = [ "windows-link", ] +[[package]] +name = "compact_str" +version = "0.8.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7fd622ebbb56a5b2ccb651b32b911cdeb2a9b4b11776b2473bf26a26a286244e" +dependencies = [ + "castaway", + "cfg-if", + "itoa", + "rustversion", + "ryu", + "static_assertions", +] + [[package]] name = "core-foundation-sys" version = "0.8.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b" +[[package]] +name = "crossterm" +version = "0.28.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "829d955a0bb380ef178a640b91779e3987da38c9aea133b20614cfed8cdea9c6" +dependencies = [ + "bitflags", + "crossterm_winapi", + "mio", + "parking_lot", + "rustix", + "signal-hook", + "signal-hook-mio", + "winapi", +] + +[[package]] +name = "crossterm_winapi" +version = "0.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "acdd7c62a3665c7f6830a51635d9ac9b23ed385797f70a83bb8bafe9c572ab2b" +dependencies = [ + "winapi", +] + +[[package]] +name = "darling" +version = "0.23.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "25ae13da2f202d56bd7f91c25fba009e7717a1e4a1cc98a76d844b65ae912e9d" +dependencies = [ + "darling_core", + "darling_macro", +] + +[[package]] +name = "darling_core" +version = "0.23.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9865a50f7c335f53564bb694ef660825eb8610e0a53d3e11bf1b0d3df31e03b0" +dependencies = [ + "ident_case", + "proc-macro2", + "quote", + "strsim", + "syn", +] + +[[package]] +name = "darling_macro" +version = "0.23.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ac3984ec7bd6cfa798e62b4a642426a5be0e68f9401cfc2a01e3fa9ea2fcdb8d" +dependencies = [ + "darling_core", + "quote", + "syn", +] + +[[package]] +name = "either" +version = "1.16.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "91622ff5e7162018101f2fea40d6ebf4a78bbe5a49736a2020649edf9693679e" + +[[package]] +name = "equivalent" +version = "1.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "877a4ace8713b0bcf2a4e7eec82529c029f1d0619886d18145fea96c3ffe5c0f" + [[package]] name = "errno" version = "0.3.14" @@ -77,7 +183,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys", + "windows-sys 0.61.2", ] [[package]] @@ -86,6 +192,12 @@ version = "0.1.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5baebc0774151f905a1a2cc41989300b1e6fbb29aff0ceffa1064fdd3088d582" +[[package]] +name = "foldhash" +version = "0.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d9c4f5dac5e15c24eb999c26181a6ca40b39fe946cbe4c263c7209467bc83af2" + [[package]] name = "futures" version = "0.3.32" @@ -174,6 +286,23 @@ dependencies = [ "slab", ] +[[package]] +name = "hashbrown" +version = "0.15.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9229cfe53dfd69f0609a49f65461bd93001ea1ef889cd5529dd176593f5338a1" +dependencies = [ + "allocator-api2", + "equivalent", + "foldhash", +] + +[[package]] +name = "heck" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea" + [[package]] name = "iana-time-zone" version = "0.1.65" @@ -198,6 +327,43 @@ dependencies = [ "cc", ] +[[package]] +name = "ident_case" +version = "1.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b9e0384b61958566e926dc50660321d12159025e767c18e043daf26b70104c39" + +[[package]] +name = "indoc" +version = "2.0.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "79cf5c93f93228cf8efb3ba362535fb11199ac548a09ce117c9b1adc3030d706" +dependencies = [ + "rustversion", +] + +[[package]] +name = "instability" +version = "0.3.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5eb2d60ef19920a3a9193c3e371f726ec1dafc045dac788d0fb3704272458971" +dependencies = [ + "darling", + "indoc", + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "itertools" +version = "0.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "413ee7dfc52ee1a4949ceeb7dbc8a33f2d6c088194d9f922fb8318faf1f01186" +dependencies = [ + "either", +] + [[package]] name = "itoa" version = "1.0.18" @@ -222,6 +388,12 @@ version = "0.2.186" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "68ab91017fe16c622486840e4c83c9a37afeff978bd239b5293d61ece587de66" +[[package]] +name = "linux-raw-sys" +version = "0.4.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d26c52dbd32dccf2d10cac7725f8eae5296885fb5703b261f7d0a0739ec807ab" + [[package]] name = "lock_api" version = "0.4.14" @@ -237,6 +409,15 @@ version = "0.4.30" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "616ec5685824bcc94416c6d4a7a446eea774a31efd7062c8480ba6fd06d7a6e5" +[[package]] +name = "lru" +version = "0.12.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "234cf4f4a04dc1f57e24b96cc0cd600cf2af460d4161ac5ecdd0af8e1f3b2a38" +dependencies = [ + "hashbrown", +] + [[package]] name = "memchr" version = "2.8.1" @@ -250,8 +431,9 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "02bd0af71c67b473010cbbc60715ee815645a4dc942899111f494b4b737d6fda" dependencies = [ "libc", + "log", "wasi", - "windows-sys", + "windows-sys 0.61.2", ] [[package]] @@ -292,6 +474,12 @@ dependencies = [ "windows-link", ] +[[package]] +name = "paste" +version = "1.0.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "57c0d7b74b563b49d38dae00a0c37d4d6de9b432382b2892f0574ddcae73fd0a" + [[package]] name = "pin-project-lite" version = "0.2.17" @@ -316,6 +504,27 @@ dependencies = [ "proc-macro2", ] +[[package]] +name = "ratatui" +version = "0.29.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "eabd94c2f37801c20583fc49dd5cd6b0ba68c716787c2dd6ed18571e1e63117b" +dependencies = [ + "bitflags", + "cassowary", + "compact_str", + "crossterm", + "indoc", + "instability", + "itertools", + "lru", + "paste", + "strum", + "unicode-segmentation", + "unicode-truncate", + "unicode-width 0.2.0", +] + [[package]] name = "redox_syscall" version = "0.5.18" @@ -325,6 +534,19 @@ dependencies = [ "bitflags", ] +[[package]] +name = "rustix" +version = "0.38.44" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fdb5bc1ae2baa591800df16c9ca78619bf65c0488b41b96ccec5d11220d8c154" +dependencies = [ + "bitflags", + "errno", + "libc", + "linux-raw-sys", + "windows-sys 0.59.0", +] + [[package]] name = "rustversion" version = "1.0.22" @@ -333,70 +555,54 @@ checksum = "b39cdef0fa800fc44525c84ccb54a029961a8215f9619753635a9c0d2538d46d" [[package]] name = "rusty-telemetry" -version = "0.2.0" +version = "0.3.0" dependencies = [ "chrono", + "crossterm", "futures", - "serde", - "serde_json", + "ratatui", "tokio", ] +[[package]] +name = "ryu" +version = "1.0.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9774ba4a74de5f7b1c1451ed6cd5285a32eddb5cccb8cc655a4e50009e06477f" + [[package]] name = "scopeguard" version = "1.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49" -[[package]] -name = "serde" -version = "1.0.228" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9a8e94ea7f378bd32cbbd37198a4a91436180c5bb472411e48b5ec2e2124ae9e" -dependencies = [ - "serde_core", - "serde_derive", -] - -[[package]] -name = "serde_core" -version = "1.0.228" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "41d385c7d4ca58e59fc732af25c3983b67ac852c1a25000afe1175de458b67ad" -dependencies = [ - "serde_derive", -] - -[[package]] -name = "serde_derive" -version = "1.0.228" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d540f220d3187173da220f885ab66608367b6574e925011a9353e4badda91d79" -dependencies = [ - "proc-macro2", - "quote", - "syn", -] - -[[package]] -name = "serde_json" -version = "1.0.150" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e8014e44b4736ed0538adeecded0fce2a272f22dc9578a7eb6b2d9993c74cfb9" -dependencies = [ - "itoa", - "memchr", - "serde", - "serde_core", - "zmij", -] - [[package]] name = "shlex" version = "2.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f8fadd59c855ef2080decdef8ff161eb6661b86933c9d82e5ba29dc602a55aba" +[[package]] +name = "signal-hook" +version = "0.3.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d881a16cf4426aa584979d30bd82cb33429027e42122b169753d6ef1085ed6e2" +dependencies = [ + "libc", + "signal-hook-registry", +] + +[[package]] +name = "signal-hook-mio" +version = "0.2.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b75a19a7a740b25bc7944bdee6172368f988763b744e3d4dfe753f6b4ece40cc" +dependencies = [ + "libc", + "mio", + "signal-hook", +] + [[package]] name = "signal-hook-registry" version = "1.4.8" @@ -426,7 +632,41 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "52d1cfed4120b4d927bf7c0f86d2087a4a7d6027c906d9f9d525a80573b9be51" dependencies = [ "libc", - "windows-sys", + "windows-sys 0.61.2", +] + +[[package]] +name = "static_assertions" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a2eb9349b6444b326872e140eb1cf5e7c522154d69e7a0ffb0fb81c06b37543f" + +[[package]] +name = "strsim" +version = "0.11.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7da8b5736845d9f2fcb837ea5d9e2628564b3b043a70948a3f0b778838c5fb4f" + +[[package]] +name = "strum" +version = "0.26.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8fec0f0aef304996cf250b31b5a10dee7980c85da9d759361292b8bca5a18f06" +dependencies = [ + "strum_macros", +] + +[[package]] +name = "strum_macros" +version = "0.26.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4c6bee85a5a24955dc440386795aa378cd9cf82acd5f764469152d2270e581be" +dependencies = [ + "heck", + "proc-macro2", + "quote", + "rustversion", + "syn", ] [[package]] @@ -454,7 +694,7 @@ dependencies = [ "signal-hook-registry", "socket2", "tokio-macros", - "windows-sys", + "windows-sys 0.61.2", ] [[package]] @@ -474,6 +714,35 @@ version = "1.0.24" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75" +[[package]] +name = "unicode-segmentation" +version = "1.13.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c6f5d3c3b1bf09027a88a6bc961fc00497d651009560b5463668dc81b0fa87a8" + +[[package]] +name = "unicode-truncate" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b3644627a5af5fa321c95b9b235a72fd24cd29c648c2c379431e6628655627bf" +dependencies = [ + "itertools", + "unicode-segmentation", + "unicode-width 0.1.14", +] + +[[package]] +name = "unicode-width" +version = "0.1.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7dd6e30e90baa6f72411720665d41d89b9a3d039dc45b8faea1ddd07f617f6af" + +[[package]] +name = "unicode-width" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1fc81956842c57dac11422a97c3b8195a1ff727f06e85c84ed2e8aa277c9a0fd" + [[package]] name = "wasi" version = "0.11.1+wasi-snapshot-preview1" @@ -525,6 +794,28 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "winapi" +version = "0.3.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5c839a674fcd7a98952e593242ea400abe93992746761e38641405d28b00f419" +dependencies = [ + "winapi-i686-pc-windows-gnu", + "winapi-x86_64-pc-windows-gnu", +] + +[[package]] +name = "winapi-i686-pc-windows-gnu" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ac3b87c63620426dd9b991e5ce0329eff545bccbbb34f3be09ff6fb6ab51b7b6" + +[[package]] +name = "winapi-x86_64-pc-windows-gnu" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f" + [[package]] name = "windows-core" version = "0.62.2" @@ -584,6 +875,15 @@ dependencies = [ "windows-link", ] +[[package]] +name = "windows-sys" +version = "0.59.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1e38bc4d79ed67fd075bcc251a1c39b32a1776bbe92e5bef1f0bf1f8c531853b" +dependencies = [ + "windows-targets", +] + [[package]] name = "windows-sys" version = "0.61.2" @@ -594,7 +894,65 @@ dependencies = [ ] [[package]] -name = "zmij" -version = "1.0.21" +name = "windows-targets" +version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b8848ee67ecc8aedbaf3e4122217aff892639231befc6a1b58d29fff4c2cabaa" +checksum = "9b724f72796e036ab90c1021d4780d4d3d648aca59e491e6b98e725b84e99973" +dependencies = [ + "windows_aarch64_gnullvm", + "windows_aarch64_msvc", + "windows_i686_gnu", + "windows_i686_gnullvm", + "windows_i686_msvc", + "windows_x86_64_gnu", + "windows_x86_64_gnullvm", + "windows_x86_64_msvc", +] + +[[package]] +name = "windows_aarch64_gnullvm" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32a4622180e7a0ec044bb555404c800bc9fd9ec262ec147edd5989ccd0c02cd3" + +[[package]] +name = "windows_aarch64_msvc" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "09ec2a7bb152e2252b53fa7803150007879548bc709c039df7627cabbd05d469" + +[[package]] +name = "windows_i686_gnu" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8e9b5ad5ab802e97eb8e295ac6720e509ee4c243f69d781394014ebfe8bbfa0b" + +[[package]] +name = "windows_i686_gnullvm" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0eee52d38c090b3caa76c563b86c3a4bd71ef1a819287c19d586d7334ae8ed66" + +[[package]] +name = "windows_i686_msvc" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "240948bc05c5e7c6dabba28bf89d89ffce3e303022809e73deaefe4f6ec56c66" + +[[package]] +name = "windows_x86_64_gnu" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "147a5c80aabfbf0c7d901cb5895d1de30ef2907eb21fbbab29ca94c5b08b1a78" + +[[package]] +name = "windows_x86_64_gnullvm" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "24d5b23dc417412679681396f2b49f3de8c1473deb516bd34410872eff51ed0d" + +[[package]] +name = "windows_x86_64_msvc" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec" diff --git a/Cargo.toml b/Cargo.toml index c005d9f..0105245 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,12 +1,12 @@ [package] name = "rusty-telemetry" -version = "0.2.0" +version = "0.3.0" edition = "2021" description = "Sim racing telemetry receiver and analyzer" [dependencies] tokio = { version = "1", features = ["full"] } -serde = { version = "1", features = ["derive"] } -serde_json = "1" chrono = "0.4" futures = "0.3" +ratatui = "0.29" +crossterm = "0.28" diff --git a/src/binary_utils.rs b/src/binary_utils.rs new file mode 100644 index 0000000..970fb1d --- /dev/null +++ b/src/binary_utils.rs @@ -0,0 +1,97 @@ +//! Shared binary reading helpers for UDP packet parsers. +//! +//! All helpers use `get()` for safe bounds checking — they return a zero/empty +//! fallback instead of panicking when the offset is out of bounds. + +use crate::telemetry_frame::NUM_WHEELS; + +/// Read a `bool` (1 byte) from the data at the given offset. +/// Returns `false` if the offset is out of bounds. +pub fn read_bool(data: &[u8], offset: usize) -> bool { + data.get(offset).map(|&b| b != 0).unwrap_or(false) +} + +/// Read a little-endian `u32` from the data at the given offset. +/// Returns 0 if the offset is out of bounds (safe fallback for malformed packets). +pub fn read_u32(data: &[u8], offset: usize) -> u32 { + let buf: [u8; 4] = data + .get(offset..offset + 4) + .and_then(|s| s.try_into().ok()) + .unwrap_or([0u8; 4]); + u32::from_le_bytes(buf) +} + +/// Read a little-endian `i32` from the data at the given offset. +/// Returns 0 if the offset is out of bounds (safe fallback for malformed packets). +pub fn read_i32(data: &[u8], offset: usize) -> i32 { + let buf: [u8; 4] = data + .get(offset..offset + 4) + .and_then(|s| s.try_into().ok()) + .unwrap_or([0u8; 4]); + i32::from_le_bytes(buf) +} + +/// Read a little-endian `i16` from the data at the given offset. +/// Returns 0 if the offset is out of bounds (safe fallback for malformed packets). +pub fn read_i16(data: &[u8], offset: usize) -> i16 { + let buf: [u8; 2] = data + .get(offset..offset + 2) + .and_then(|s| s.try_into().ok()) + .unwrap_or([0u8; 2]); + i16::from_le_bytes(buf) +} + +/// Read a little-endian `f32` from the data at the given offset. +/// Returns 0.0 if the offset is out of bounds (safe fallback for malformed packets). +pub fn read_f32(data: &[u8], offset: usize) -> f32 { + let buf: [u8; 4] = data + .get(offset..offset + 4) + .and_then(|s| s.try_into().ok()) + .unwrap_or([0u8; 4]); + f32::from_le_bytes(buf) +} + +/// Read a little-endian `f64` from the data at the given offset. +/// Returns 0.0 if the offset is out of bounds (safe fallback for malformed packets). +pub fn read_f64(data: &[u8], offset: usize) -> f64 { + let buf: [u8; 8] = data + .get(offset..offset + 8) + .and_then(|s| s.try_into().ok()) + .unwrap_or([0u8; 8]); + f64::from_le_bytes(buf) +} + +/// Read an array of 4 little-endian `f32` from the data at the given offset. +pub fn read_f32_4(data: &[u8], offset: usize) -> [f32; NUM_WHEELS] { + let mut arr = [0.0f32; NUM_WHEELS]; + for i in 0..NUM_WHEELS { + arr[i] = read_f32(data, offset + i * 4); + } + arr +} + +/// Read an array of 2 little-endian `f32` from the data at the given offset. +pub fn read_f32_2(data: &[u8], offset: usize) -> [f32; 2] { + [read_f32(data, offset), read_f32(data, offset + 4)] +} + +/// Read an array of 3 little-endian `f32` from the data at the given offset. +pub fn read_f32_3(data: &[u8], offset: usize) -> [f32; 3] { + [read_f32(data, offset), read_f32(data, offset + 4), read_f32(data, offset + 8)] +} + +/// Read an array of 3 little-endian `i32` from the data at the given offset. +pub fn read_i32_3(data: &[u8], offset: usize) -> [i32; 3] { + [read_i32(data, offset), read_i32(data, offset + 4), read_i32(data, offset + 8)] +} + +/// Read a null-terminated/padded ASCII string. +/// Returns an empty string if the offset is out of bounds. +pub fn read_string(data: &[u8], offset: usize, len: usize) -> String { + let slice = match data.get(offset..offset + len) { + Some(s) => s, + None => return String::new(), + }; + let end = slice.iter().position(|&b| b == 0).unwrap_or(len); + String::from_utf8_lossy(&slice[..end]).to_string() +} diff --git a/src/feed_state.rs b/src/feed_state.rs new file mode 100644 index 0000000..52f6532 --- /dev/null +++ b/src/feed_state.rs @@ -0,0 +1,389 @@ +use crate::telemetry_frame::TelemetryFrame; +use std::time::Instant; + +const RING_CAPACITY: usize = 50; +const STALE_THRESHOLD_SECS: f64 = 5.0; +const MIN_TIME_SPAN_SECS: f64 = 0.001; + +#[derive(Clone, Copy, Debug, PartialEq)] +pub enum FeedHealth { + Waiting, + Alive, + Stale, + Error, +} + +#[derive(Clone, Copy, Debug, PartialEq)] +pub enum HandshakePhase { + NotStarted, + WaitingForResponse, + Subscribing, + Subscribed, +} + +#[derive(Clone, Copy, Debug, PartialEq)] +pub enum FeedMode { + BindListen, + Handshake { + target_port: u16, + recv_port: u16, + }, + Disabled, +} + +#[derive(Clone)] +pub struct FeedState { + pub health: FeedHealth, + pub handshake_phase: HandshakePhase, + pub status_text: String, + pub total_packets: u64, + pub total_bytes: u64, + pub frequency_hz: f64, + pub data_rate_kb_s: f64, + pub last_packet_ago_secs: Option, + pub start_time: Option, + pub last_packet_time: Option, + pub packet_timestamps: Vec, + pub ring_index: usize, + pub ring_count: usize, + pub last_frame: Option, +} + +impl FeedState { + pub fn new() -> Self { + Self { + health: FeedHealth::Waiting, + handshake_phase: HandshakePhase::NotStarted, + status_text: String::from("Bound, waiting for data"), + total_packets: 0, + total_bytes: 0, + frequency_hz: 0.0, + data_rate_kb_s: 0.0, + last_packet_ago_secs: None, + start_time: None, + last_packet_time: None, + packet_timestamps: vec![Instant::now(); RING_CAPACITY], + ring_index: 0, + ring_count: 0, + last_frame: None, + } + } + + pub fn update_frame(&mut self, frame: TelemetryFrame) { + self.last_frame = Some(frame); + } + + pub fn record_packet(&mut self, len: usize) { + let now = Instant::now(); + if self.start_time.is_none() { + self.start_time = Some(now); + } + + self.total_packets += 1; + self.total_bytes += len as u64; + self.last_packet_time = Some(now); + self.last_packet_ago_secs = Some(0.0); + + if self.ring_count < RING_CAPACITY { + if self.ring_count < self.packet_timestamps.len() { + self.packet_timestamps[self.ring_count] = now; + } + self.ring_count += 1; + self.ring_index = self.ring_count % RING_CAPACITY; + } else { + self.packet_timestamps[self.ring_index] = now; + self.ring_index = (self.ring_index + 1) % RING_CAPACITY; + } + + self.frequency_hz = self.calc_frequency(); + self.data_rate_kb_s = self.calc_data_rate(); + self.health = FeedHealth::Alive; + if self.status_text != "Receiving" { + self.status_text = String::from("Receiving"); + } + } + + fn calc_frequency(&self) -> f64 { + if self.ring_count < 2 { + return 0.0; + } + + let count = self.ring_count.min(RING_CAPACITY); + let oldest_idx = if self.ring_count >= RING_CAPACITY { + self.ring_index + } else { + 0 + }; + let newest_idx = if self.ring_count >= RING_CAPACITY { + if self.ring_index == 0 { + RING_CAPACITY - 1 + } else { + self.ring_index - 1 + } + } else { + self.ring_count - 1 + }; + + let oldest = self.packet_timestamps[oldest_idx]; + let newest = self.packet_timestamps[newest_idx]; + let span = newest.duration_since(oldest).as_secs_f64(); + + if span < MIN_TIME_SPAN_SECS { + return 0.0; + } + + count as f64 / span + } + + fn calc_data_rate(&self) -> f64 { + match self.start_time { + Some(start) => { + let elapsed = start.elapsed().as_secs_f64(); + if elapsed > 0.0 { + (self.total_bytes as f64 / 1024.0) / elapsed + } else { + 0.0 + } + } + None => 0.0, + } + } + + pub fn update_health(&mut self) { + match self.last_packet_ago_secs { + None => { + if self.health != FeedHealth::Error { + self.health = FeedHealth::Waiting; + self.status_text = String::from("Waiting for data"); + } + } + Some(ago) => { + if ago < STALE_THRESHOLD_SECS { + self.health = FeedHealth::Alive; + self.status_text = String::from("Receiving"); + } else { + self.health = FeedHealth::Stale; + self.status_text = format!("Stale ({:.1}s since last packet)", ago); + } + } + } + } + + pub fn tick(&mut self) { + if let Some(last) = self.last_packet_time { + self.last_packet_ago_secs = Some(last.elapsed().as_secs_f64()); + } + self.frequency_hz = self.calc_frequency(); + self.data_rate_kb_s = self.calc_data_rate(); + self.update_health(); + } + + pub fn set_handshake_phase(&mut self, phase: HandshakePhase) { + self.handshake_phase = phase; + } + + pub fn set_error(&mut self, msg: &str) { + self.health = FeedHealth::Error; + self.status_text = format!("Error: {}", msg); + } +} + +impl Default for FeedState { + fn default() -> Self { + Self::new() + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::thread; + + #[test] + fn test_new_state_is_waiting() { + let state = FeedState::new(); + assert_eq!(state.health, FeedHealth::Waiting); + assert_eq!(state.handshake_phase, HandshakePhase::NotStarted); + assert_eq!(state.total_packets, 0); + assert_eq!(state.total_bytes, 0); + assert_eq!(state.frequency_hz, 0.0); + assert!(state.last_packet_ago_secs.is_none()); + assert!(state.last_frame.is_none()); + } + + #[test] + fn test_record_packet_increments_counters() { + let mut state = FeedState::new(); + state.record_packet(100); + assert_eq!(state.total_packets, 1); + assert_eq!(state.total_bytes, 100); + assert_eq!(state.health, FeedHealth::Alive); + assert_eq!(state.status_text, "Receiving"); + + state.record_packet(200); + assert_eq!(state.total_packets, 2); + assert_eq!(state.total_bytes, 300); + } + + #[test] + fn test_frequency_zero_with_fewer_than_two_packets() { + let mut state = FeedState::new(); + state.record_packet(100); + assert_eq!(state.frequency_hz, 0.0); + + // With 2 packets at the same instant, span is < MIN_TIME_SPAN_SECS, so still 0 + state.record_packet(100); + // frequency could be 0 if the two packets are within MIN_TIME_SPAN_SECS + // This is expected behavior + } + + #[test] + fn test_frequency_nonzero_after_spaced_packets() { + let mut state = FeedState::new(); + for _ in 0..10 { + state.record_packet(100); + thread::sleep(std::time::Duration::from_millis(10)); + } + // 10 packets over ~90ms → frequency should be > 0 + assert!(state.frequency_hz > 0.0); + // Rough sanity: 10 packets / 0.09s ≈ 111 Hz, but with overhead, just check > 0 + assert!(state.frequency_hz < 10_000.0); // sanity upper bound + } + + #[test] + fn test_ring_buffer_fills_without_wraparound() { + let mut state = FeedState::new(); + for _ in 0..RING_CAPACITY { + state.record_packet(50); + } + assert_eq!(state.ring_count, RING_CAPACITY); + assert_eq!(state.ring_index, 0); // wraps to 0 after filling + } + + #[test] + fn test_ring_buffer_wraps_around() { + let mut state = FeedState::new(); + for _ in 0..RING_CAPACITY { + state.record_packet(50); + } + assert_eq!(state.ring_index, 0); + + // Add more packets to trigger wraparound + for _ in 0..10 { + state.record_packet(50); + } + assert_eq!(state.ring_count, RING_CAPACITY); + assert_eq!(state.ring_index, 10); + assert_eq!(state.total_packets, (RING_CAPACITY + 10) as u64); + } + + #[test] + fn test_health_waiting_to_alive() { + let mut state = FeedState::new(); + assert_eq!(state.health, FeedHealth::Waiting); + + state.record_packet(100); + assert_eq!(state.health, FeedHealth::Alive); + } + + #[test] + fn test_health_alive_to_stale() { + let mut state = FeedState::new(); + state.record_packet(100); + assert_eq!(state.health, FeedHealth::Alive); + + // Simulate time passing by manually setting last_packet_ago_secs + state.last_packet_ago_secs = Some(6.0); + state.update_health(); + assert_eq!(state.health, FeedHealth::Stale); + assert!(state.status_text.starts_with("Stale")); + } + + #[test] + fn test_health_stale_to_alive() { + let mut state = FeedState::new(); + state.record_packet(100); + state.last_packet_ago_secs = Some(6.0); + state.update_health(); + assert_eq!(state.health, FeedHealth::Stale); + + // New packet brings it back to alive + state.record_packet(100); + assert_eq!(state.health, FeedHealth::Alive); + } + + #[test] + fn test_health_error_is_sticky_via_update() { + let mut state = FeedState::new(); + state.set_error("test error"); + assert_eq!(state.health, FeedHealth::Error); + + // update_health should not override Error when ago is None + state.last_packet_ago_secs = None; + state.update_health(); + assert_eq!(state.health, FeedHealth::Error); + } + + #[test] + fn test_tick_updates_ago_secs() { + let mut state = FeedState::new(); + + // Before any packet, tick should not set ago + state.tick(); + assert!(state.last_packet_ago_secs.is_none()); + + state.record_packet(100); + // Immediately after, ago should be ~0 + state.tick(); + assert!(state.last_packet_ago_secs.is_some()); + assert!(state.last_packet_ago_secs.unwrap() < 1.0); + + // After sleeping, ago should be > 0 + thread::sleep(std::time::Duration::from_millis(100)); + state.tick(); + let ago = state.last_packet_ago_secs.unwrap(); + assert!(ago >= 0.09, "expected ago >= 0.09, got {}", ago); + } + + #[test] + fn test_data_rate_calculation() { + let mut state = FeedState::new(); + state.record_packet(1024); + thread::sleep(std::time::Duration::from_millis(50)); + state.record_packet(1024); + + // 2048 bytes over ~50ms → roughly 40 KB/s + state.data_rate_kb_s = state.calc_data_rate(); + assert!(state.data_rate_kb_s > 0.0); + } + + #[test] + fn test_update_frame() { + let mut state = FeedState::new(); + assert!(state.last_frame.is_none()); + + let frame = TelemetryFrame::default(); + state.update_frame(frame); + assert!(state.last_frame.is_some()); + } + + #[test] + fn test_set_handshake_phase() { + let mut state = FeedState::new(); + assert_eq!(state.handshake_phase, HandshakePhase::NotStarted); + + state.set_handshake_phase(HandshakePhase::WaitingForResponse); + assert_eq!(state.handshake_phase, HandshakePhase::WaitingForResponse); + + state.set_handshake_phase(HandshakePhase::Subscribed); + assert_eq!(state.handshake_phase, HandshakePhase::Subscribed); + } + + #[test] + fn test_set_error() { + let mut state = FeedState::new(); + state.set_error("bind failed"); + assert_eq!(state.health, FeedHealth::Error); + assert_eq!(state.status_text, "Error: bind failed"); + } +} diff --git a/src/main.rs b/src/main.rs index f7710e6..0f07f69 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,32 +1,59 @@ -mod stats; -mod telemetry; +mod binary_utils; +mod feed_state; +mod ui; +mod telemetry_frame; +mod parser_ksudp; +mod parser_telemetry_tool; -use stats::ReceiveStats; -use telemetry::TelemetryFrame; +use feed_state::*; +use ui::*; -use std::io::Write; +use std::io::{self, Stdout}; +use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; -use tokio::io::{AsyncWriteExt, BufWriter}; +use std::time::Duration; + +use crossterm::{ + event::{self, Event, KeyCode, KeyModifiers}, + execute, + terminal::{disable_raw_mode, enable_raw_mode, EnterAlternateScreen, LeaveAlternateScreen}, +}; +use ratatui::backend::CrosstermBackend; +use ratatui::Terminal; +use tokio::io::BufWriter; +use tokio::io::AsyncWriteExt; + +use crate::feed_state::HandshakePhase; +use crate::telemetry_frame::TelemetryFrame; const VERSION: &str = env!("CARGO_PKG_VERSION"); const BUFFER_SIZE: usize = 65535; +const RENDER_INTERVAL_MS: u64 = 250; +const HANDSHAKE_INTERVAL_SECS: u64 = 10; +const HANDSHAKE_RESPONSE_SIZE: usize = 408; + +const HANDSHAKE_REQUEST: [u8; 12] = [1, 0, 0, 0, 1, 0, 0, 0, 0, 0, 0, 0]; +const SUBSCRIBE_UPDATE: [u8; 12] = [1, 0, 0, 0, 1, 0, 0, 0, 1, 0, 0, 0]; + +fn decode_utf16le(bytes: &[u8]) -> String { + let chars: Vec = bytes + .chunks(2) + .map(|c| u16::from_le_bytes([c[0], c.get(1).copied().unwrap_or(0)])) + .take_while(|&c| c != 0) + .collect(); + String::from_utf16_lossy(&chars) +} fn bind_addr() -> String { std::env::var("RUSTY_TELEMETRY_BIND").unwrap_or_else(|_| "127.0.0.1".to_string()) } -#[derive(Clone)] -enum FeedFormat { - Binary, - Json, -} - #[derive(Clone)] struct FeedConfig { port: u16, name: &'static str, label: &'static str, - format: FeedFormat, + needs_handshake: bool, } static FEEDS: &[FeedConfig] = &[ @@ -34,53 +61,170 @@ static FEEDS: &[FeedConfig] = &[ port: 9996, name: "ksudp", label: "KSUDP", - format: FeedFormat::Binary, + needs_handshake: true, }, FeedConfig { port: 10101, name: "telemetry_tool", label: "Telemetry Tool", - format: FeedFormat::Binary, - }, - FeedConfig { - port: 5005, - name: "ac_plugin", - label: "AC Lua Plugin", - format: FeedFormat::Json, + needs_handshake: false, }, ]; -fn print_banner(results: &[(&FeedConfig, bool)], session_dir: &str, bind: &str) { - println!("========================================"); - println!(" rusty-telemetry v{VERSION} — Multi-feed telemetry listener"); - println!("========================================"); - for (feed, ok) in results { - let status = if *ok { "✅ Listening" } else { "❌ Failed" }; - println!( - " Feed :{} [{}] {}", - feed.port, feed.label, status - ); - } - println!("========================================"); - println!(" Bind: {bind}"); - println!(" Data dir: {session_dir}/"); - println!("========================================"); - println!(); +type Term = Terminal>; + +fn setup_terminal() -> io::Result { + enable_raw_mode()?; + let mut stdout = io::stdout(); + execute!(stdout, EnterAlternateScreen)?; + let backend = CrosstermBackend::new(stdout); + let terminal = Terminal::new(backend)?; + Ok(terminal) +} + +fn restore_terminal(terminal: &mut Term) -> io::Result<()> { + disable_raw_mode()?; + execute!(terminal.backend_mut(), LeaveAlternateScreen)?; + terminal.show_cursor()?; + Ok(()) +} + +fn install_panic_hook() { + let original_hook = std::panic::take_hook(); + std::panic::set_hook(Box::new(move |info| { + let _ = disable_raw_mode(); + let _ = execute!(io::stdout(), LeaveAlternateScreen); + original_hook(info); + })); } async fn run_feed( feed: FeedConfig, socket: tokio::net::UdpSocket, session_dir: String, + state: Arc>, shutdown: Arc, ) { - let ext = match feed.format { - FeedFormat::Binary => "bin", - FeedFormat::Json => "json", + let handshake_target: Option = if feed.needs_handshake { + let target: std::net::SocketAddr = match format!("{}:{}", bind_addr(), feed.port).parse() { + Ok(addr) => addr, + Err(e) => { + eprintln!( + "[ERROR] {} :{} Invalid target address '{}': {}", + feed.label, feed.port, bind_addr(), e + ); + if let Ok(mut s) = state.lock() { + s.set_error(&format!("Invalid target address: {}", e)); + } + return; + } + }; + + // Phase 1: Send HANDSHAKE + eprintln!( + "[INFO] {} :{} Phase 1: Sending HANDSHAKE to {}", + feed.label, feed.port, target + ); + if let Err(e) = socket.send_to(&HANDSHAKE_REQUEST, target).await { + eprintln!( + "[WARN] HANDSHAKE send failed for :{}: {}", + feed.port, e + ); + } + if let Ok(mut s) = state.lock() { + s.set_handshake_phase(HandshakePhase::WaitingForResponse); + } + + // Phase 2: Wait for 408-byte HandshakerResponse with timeout + let mut hs_buf = [0u8; HANDSHAKE_RESPONSE_SIZE]; + match tokio::time::timeout( + Duration::from_secs(5), + socket.recv_from(&mut hs_buf), + ) + .await + { + Ok(Ok((len, _src))) if len == HANDSHAKE_RESPONSE_SIZE => { + let car_name = decode_utf16le(&hs_buf[0..100]); + let driver_name = decode_utf16le(&hs_buf[100..200]); + let identifier = i32::from_le_bytes([ + hs_buf[200], hs_buf[201], hs_buf[202], hs_buf[203], + ]); + let version = i32::from_le_bytes([ + hs_buf[204], hs_buf[205], hs_buf[206], hs_buf[207], + ]); + let track_name = decode_utf16le(&hs_buf[208..308]); + let track_config = decode_utf16le(&hs_buf[308..408]); + + eprintln!( + "[INFO] {} :{} Phase 2: HandshakerResponse received", + feed.label, feed.port + ); + eprintln!( + "[INFO] car_name=\"{}\" driver_name=\"{}\" id={} ver={}", + car_name, driver_name, identifier, version + ); + eprintln!( + "[INFO] track_name=\"{}\" track_config=\"{}\"", + track_name, track_config + ); + + if identifier != 4242 { + eprintln!( + "[WARN] {} :{} Unexpected handshake identifier: {} (expected 4242)", + feed.label, feed.port, identifier + ); + } + } + Ok(Ok((len, _))) => { + eprintln!( + "[WARN] {} :{} Phase 2: Unexpected response size {} (expected {})", + feed.label, feed.port, len, HANDSHAKE_RESPONSE_SIZE + ); + } + Ok(Err(e)) => { + eprintln!( + "[WARN] {} :{} Phase 2: Recv error: {}", + feed.label, feed.port, e + ); + } + Err(_) => { + eprintln!( + "[WARN] {} :{} Phase 2: Handshake response timed out (5s). AC may not be running.", + feed.label, feed.port + ); + } + } + + // Phase 3: Send SUBSCRIBE_UPDATE + if let Ok(mut s) = state.lock() { + s.set_handshake_phase(HandshakePhase::Subscribing); + } + eprintln!( + "[INFO] {} :{} Phase 3: Sending SUBSCRIBE_UPDATE", + feed.label, feed.port + ); + if let Err(e) = socket.send_to(&SUBSCRIBE_UPDATE, target).await { + eprintln!( + "[WARN] SUBSCRIBE_UPDATE send failed for :{}: {}", + feed.port, e + ); + } + if let Ok(mut s) = state.lock() { + s.set_handshake_phase(HandshakePhase::Subscribed); + } + eprintln!( + "[INFO] {} :{} Subscribed, entering streaming loop", + feed.label, feed.port + ); + + Some(target) + } else { + None }; + let file_path = format!( - "{session_dir}/feed_{}_{}.{ext}", - feed.port, feed.name + "{}/feed_{}_{}.bin", + session_dir, feed.port, feed.name ); let file = match tokio::fs::OpenOptions::new() .create(true) @@ -94,70 +238,59 @@ async fn run_feed( "[ERROR] Cannot open dump file for :{} ({}): {}", feed.port, feed.name, e ); + if let Ok(mut s) = state.lock() { + s.set_error(&format!("Cannot open dump file: {}", e)); + } return; } }; let mut file = BufWriter::new(file); - let mut buf = vec![0u8; BUFFER_SIZE]; - let mut stats = ReceiveStats::new(); - let stdout = std::io::stdout(); + + let mut hs_interval: Option = handshake_target.map(|_| { + tokio::time::interval(Duration::from_secs(HANDSHAKE_INTERVAL_SECS)) + }); loop { tokio::select! { result = socket.recv_from(&mut buf) => { match result { Ok((len, _src)) => { - stats.record(len); + // Parse telemetry data outside the lock to minimize + // contention with the TUI render thread. + let parsed: Option = if feed.needs_handshake { + // KSUDP: parse RTCarInfo (skip 408-byte handshake responses) + if len != HANDSHAKE_RESPONSE_SIZE { + parser_ksudp::parse_rt_car_info(&buf[..len]).ok() + } else { + None + } + } else { + // Telemetry Tool: parse only the first driver + parser_telemetry_tool::parse_first_driver(&buf[..len]).ok() + }; - if let FeedFormat::Binary = feed.format { - let header = (len as u32).to_be_bytes(); - if let Err(e) = file.write_all(&header).await { - eprintln!( - "[ERROR] Write error on :{} ({}): {}", - feed.port, feed.name, e - ); + // Update state in a single brief lock acquisition. + if let Ok(mut s) = state.lock() { + s.record_packet(len); + if let Some(frame) = parsed { + s.update_frame(frame); } } + let header = (len as u32).to_be_bytes(); + if let Err(e) = file.write_all(&header).await { + eprintln!( + "[ERROR] Write error on :{} ({}): {}", + feed.port, feed.name, e + ); + } if let Err(e) = file.write_all(&buf[..len]).await { eprintln!( "[ERROR] Write error on :{} ({}): {}", feed.port, feed.name, e ); } - if let FeedFormat::Json = feed.format { - let _ = file.write_all(b"\n").await; - } - - { - let mut handle = stdout.lock(); - let fmt_tag = match feed.format { - FeedFormat::Binary => "Binary", - FeedFormat::Json => "JSON", - }; - let _ = writeln!( - handle, - "[ :{} | {} ] {fmt_tag} packet: {} bytes (#{})", - feed.port, - feed.label, - len, - stats.total_packets, - ); - if let FeedFormat::Json = feed.format { - if let Ok(text) = std::str::from_utf8(&buf[..len]) { - if let Ok(frame) = - serde_json::from_str::(text) - { - let _ = writeln!(handle, "{frame}"); - } else { - let _ = writeln!(handle, " [WARN] JSON parse failed"); - } - } - } - let _ = writeln!(handle, " {}", stats); - let _ = writeln!(handle); - } } Err(e) => { eprintln!( @@ -167,6 +300,21 @@ async fn run_feed( } } } + _ = async { + match hs_interval.as_mut() { + Some(interval) => interval.tick().await, + None => std::future::pending().await, + } + } => { + if let Some(target) = handshake_target { + if let Err(e) = socket.send_to(&SUBSCRIBE_UPDATE, target).await { + eprintln!( + "[WARN] Periodic SUBSCRIBE_UPDATE send failed for :{}: {}", + feed.port, e + ); + } + } + } _ = shutdown.notified() => { break; } @@ -193,70 +341,155 @@ async fn run_feed( #[tokio::main] async fn main() { - let bind = bind_addr(); + install_panic_hook(); + let bind = bind_addr(); let session_id = chrono::Local::now().format("%Y%m%d_%H%M%S").to_string(); - let session_dir = format!("./data/{session_id}"); + let session_dir = format!("./data/{}", session_id); if let Err(e) = tokio::fs::create_dir_all(&session_dir).await { eprintln!("[ERROR] Cannot create data dir {}: {}", session_dir, e); std::process::exit(1); } - let mut banner_data: Vec<(&FeedConfig, bool)> = Vec::new(); - let mut bound_feeds: Vec<(FeedConfig, tokio::net::UdpSocket)> = Vec::new(); + let mut terminal = match setup_terminal() { + Ok(t) => t, + Err(e) => { + eprintln!("[ERROR] Failed to setup terminal: {}", e); + std::process::exit(1); + } + }; + + let mut feed_displays: Vec<(FeedDisplayInfo, Arc>)> = Vec::new(); + let mut feed_tasks: Vec> = Vec::new(); + let shutdown = Arc::new(tokio::sync::Notify::new()); for feed in FEEDS { - let addr = format!("{bind}:{}", feed.port); - match tokio::net::UdpSocket::bind(&addr).await { - Ok(socket) => { - banner_data.push((feed, true)); - bound_feeds.push((feed.clone(), socket)); + let state = Arc::new(std::sync::Mutex::new(FeedState::new())); + + let (socket_opt, mode) = if feed.needs_handshake { + let recv_addr = format!("{}:0", bind); + match tokio::net::UdpSocket::bind(&recv_addr).await { + Ok(s) => { + let local_port = s.local_addr().map(|a| a.port()).unwrap_or(0); + (Some(s), FeedMode::Handshake { + target_port: feed.port, + recv_port: local_port, + }) + } + Err(e) => { + eprintln!( + "[WARN] Failed to bind handshake recv socket for :{} ({}): {}", + feed.port, feed.label, e + ); + if let Ok(mut s) = state.lock() { + s.set_error(&format!("Bind failed: {}", e)); + } + (None, FeedMode::Handshake { + target_port: feed.port, + recv_port: 0, + }) + } } - Err(e) => { - eprintln!( - "[WARN] Failed to bind :{} ({}): {} — skipping", - feed.port, feed.label, e - ); - banner_data.push((feed, false)); + } else { + let addr = format!("{}:{}", bind, feed.port); + match tokio::net::UdpSocket::bind(&addr).await { + Ok(s) => (Some(s), FeedMode::BindListen), + Err(e) => { + eprintln!( + "[WARN] Failed to bind :{} ({}): {} -- feed is optional, continuing without it", + feed.port, feed.label, e + ); + (None, FeedMode::Disabled) + } + } + }; + + feed_displays.push(( + FeedDisplayInfo { + label: feed.label, + port: feed.port, + mode, + }, + state.clone(), + )); + + if let Some(socket) = socket_opt { + let feed_clone = feed.clone(); + let dir = session_dir.clone(); + let sh = shutdown.clone(); + feed_tasks.push(tokio::spawn(async move { + run_feed(feed_clone, socket, dir, state, sh).await; + })); + } + } + + let (event_tx, mut event_rx) = tokio::sync::mpsc::channel::(100); + let event_running = Arc::new(AtomicBool::new(true)); + let event_running_clone = event_running.clone(); + tokio::spawn(async move { + while event_running_clone.load(Ordering::Relaxed) { + if event::poll(Duration::from_millis(100)).unwrap_or(false) { + if let Ok(ev) = event::read() { + if event_tx.send(ev).await.is_err() { + break; + } + } + } + } + }); + + let mut render_interval = tokio::time::interval(Duration::from_millis(RENDER_INTERVAL_MS)); + let running = Arc::new(AtomicBool::new(true)); + + while running.load(Ordering::Relaxed) { + tokio::select! { + maybe_event = event_rx.recv() => { + if let Some(ev) = maybe_event { + match ev { + Event::Key(key) => { + match key.code { + KeyCode::Char('q') => { + running.store(false, Ordering::Relaxed); + } + KeyCode::Char('c') + if key.modifiers.contains(KeyModifiers::CONTROL) => + { + running.store(false, Ordering::Relaxed); + } + _ => {} + } + } + Event::Resize(_, _) => { + let _ = terminal.draw(|f| { + render(f, &feed_displays, &session_dir, VERSION); + }); + } + _ => {} + } + } + } + _ = render_interval.tick() => { + let _ = terminal.draw(|f| { + render(f, &feed_displays, &session_dir, VERSION); + }); } } } - print_banner(&banner_data, &session_dir, &bind); - - if bound_feeds.is_empty() { - eprintln!("[ERROR] No feeds could bind. Exiting."); - std::process::exit(1); - } - - let shutdown = Arc::new(tokio::sync::Notify::new()); - - let mut tasks = Vec::new(); - for (feed, socket) in bound_feeds { - let dir = session_dir.clone(); - let shutdown = shutdown.clone(); - tasks.push(tokio::spawn(async move { - run_feed(feed, socket, dir, shutdown).await; - })); - } - - tokio::signal::ctrl_c().await.ok(); - println!(); - println!("[INFO] Ctrl+C received, shutting down..."); - + event_running.store(false, Ordering::Relaxed); shutdown.notify_waiters(); match tokio::time::timeout( - std::time::Duration::from_secs(3), - futures::future::join_all(tasks), + Duration::from_secs(3), + futures::future::join_all(feed_tasks), ) .await { Ok(results) => { for r in results { if let Err(e) = r { - eprintln!("[WARN] Task join error: {e}"); + eprintln!("[WARN] Task join error: {}", e); } } } @@ -264,4 +497,10 @@ async fn main() { eprintln!("[WARN] Timeout waiting for feeds to flush; forcing exit"); } } + + if let Err(e) = restore_terminal(&mut terminal) { + eprintln!("[WARN] Failed to restore terminal: {}", e); + } + + eprintln!("[INFO] rusty-telemetry shut down cleanly."); } diff --git a/src/parser_ksudp.rs b/src/parser_ksudp.rs new file mode 100644 index 0000000..4146b2c --- /dev/null +++ b/src/parser_ksudp.rs @@ -0,0 +1,241 @@ +//! Binary parser for KSUDP RTCarInfo packets (Assetto Corsa port 9996). +//! +//! After the 3-phase handshake, AC sends continuous RTCarInfo structs at ~20 Hz. +//! This parser converts the raw binary data into a `TelemetryFrame`. + +use crate::binary_utils::{read_bool, read_f32, read_f32_4, read_i32}; +use crate::telemetry_frame::{FeedSource, TelemetryFrame}; + +/// Minimum expected packet size (identifier + size field + all known fields). +/// The `size` field in the packet itself reports the true struct size. +const MIN_PACKET_SIZE: usize = 323; + +/// Parse error type for KSUDP packets. +#[derive(Debug)] +pub enum KsudpParseError { + /// Packet too short to contain a valid RTCarInfo. + TooShort { actual: usize, expected: usize }, + /// Invalid identifier byte (expected 0x61 = 'a'). + InvalidIdentifier { got: u8 }, +} + +impl std::fmt::Display for KsudpParseError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + KsudpParseError::TooShort { actual, expected } => { + write!(f, "packet too short: {} bytes (expected >= {})", actual, expected) + } + KsudpParseError::InvalidIdentifier { got } => { + write!(f, "invalid identifier: 0x{:02x} (expected 0x61 = 'a')", got) + } + } + } +} + +impl std::error::Error for KsudpParseError {} + +/// Parse a raw KSUDP RTCarInfo packet into a `TelemetryFrame`. +/// +/// # Errors +/// Returns `KsudpParseError` if the packet is too short or has an invalid identifier. +pub fn parse_rt_car_info(data: &[u8]) -> Result { + if data.len() < MIN_PACKET_SIZE { + return Err(KsudpParseError::TooShort { + actual: data.len(), + expected: MIN_PACKET_SIZE, + }); + } + + // Validate identifier + let identifier = data[0]; + if identifier != b'a' { + return Err(KsudpParseError::InvalidIdentifier { got: identifier }); + } + + let reported_size = read_i32(data, 1) as usize; + if data.len() < reported_size && reported_size > 0 { + // Log warning but don't fail — the struct might have fields we don't read yet + eprintln!( + "[WARN] KSUDP: reported struct size {} but packet is {} bytes", + reported_size, + data.len() + ); + } + + let frame = TelemetryFrame { + source: FeedSource::Ksudp, + raw_size: data.len(), + + // Speed + speed_kmh: read_f32(data, 5), + speed_mph: read_f32(data, 9), + speed_ms: read_f32(data, 13), + + // Flags + is_abs_enabled: read_bool(data, 17), + is_abs_in_action: read_bool(data, 18), + is_tc_in_action: read_bool(data, 19), + is_tc_enabled: read_bool(data, 20), + is_in_pit: read_bool(data, 21), + is_engine_limiter_on: read_bool(data, 22), + + // G-forces + acc_g_vertical: read_f32(data, 23), + acc_g_horizontal: read_f32(data, 27), + acc_g_frontal: read_f32(data, 31), + + // Lap timing + lap_time_ms: read_i32(data, 35), + last_lap_ms: read_i32(data, 39), + best_lap_ms: read_i32(data, 43), + lap_count: read_i32(data, 47), + + // Driver inputs + throttle: read_f32(data, 51), + brake: read_f32(data, 55), + clutch: read_f32(data, 59), + engine_rpm: read_f32(data, 63), + steer: read_f32(data, 67), + + // Gear & chassis + gear: read_i32(data, 71), + cg_height: read_f32(data, 75), + + // Per-wheel arrays + wheel_angular_speed: read_f32_4(data, 79), + slip_angle: read_f32_4(data, 95), + slip_angle_contact_patch: read_f32_4(data, 111), + slip_ratio: read_f32_4(data, 127), + tyre_slip: read_f32_4(data, 143), + nd_slip: read_f32_4(data, 159), + wheel_load: read_f32_4(data, 175), + wheel_dy: read_f32_4(data, 191), + wheel_mz: read_f32_4(data, 207), + tyre_dirty_level: read_f32_4(data, 223), + camber_rad: read_f32_4(data, 239), + tyre_radius: read_f32_4(data, 255), + tyre_loaded_radius: read_f32_4(data, 271), + suspension_height: read_f32_4(data, 287), + + // Position + // NOTE: carPosition (offset 303) is the normalized spline position (0.0–1.0). + // Both car_position_normalized and spline_position map to the same AC field. + car_position_normalized: read_f32(data, 303), + car_slope: read_f32(data, 307), + world_x: read_f32(data, 311), + world_y: read_f32(data, 315), + world_z: read_f32(data, 319), + spline_position: read_f32(data, 303), // same as car_position_normalized (AC SDK: carPosition) + + ..Default::default() + }; + + // KSUDP-specific: mark Telemetry Tool fields as None (already default) + // Keep KSUDP-specific fields populated above. + + Ok(frame) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_parse_minimal_valid_packet() { + let mut packet = vec![0u8; MIN_PACKET_SIZE]; + packet[0] = b'a'; // identifier + // size = MIN_PACKET_SIZE + packet[1..5].copy_from_slice(&(MIN_PACKET_SIZE as i32).to_le_bytes()); + // speed_Kmh = 123.4 + packet[5..9].copy_from_slice(&123.4f32.to_le_bytes()); + // gear = 3 (meaning gear 2) + packet[71..75].copy_from_slice(&3i32.to_le_bytes()); + // engineRPM = 7500.0 + packet[63..67].copy_from_slice(&7500.0f32.to_le_bytes()); + // lapTime = 78123 ms (1:18.123) + packet[35..39].copy_from_slice(&78123i32.to_le_bytes()); + + let frame = parse_rt_car_info(&packet).unwrap(); + assert_eq!(frame.source, FeedSource::Ksudp); + assert!((frame.speed_kmh - 123.4).abs() < 0.01); + assert_eq!(frame.gear, 3); + assert!((frame.engine_rpm - 7500.0).abs() < 0.01); + assert_eq!(frame.lap_time_ms, 78123); + } + + #[test] + fn test_too_short() { + let packet = vec![b'a', 0, 0, 0, 0]; + let result = parse_rt_car_info(&packet); + assert!(result.is_err()); + if let Err(KsudpParseError::TooShort { actual, .. }) = result { + assert_eq!(actual, 5); + } else { + panic!("Expected TooShort error"); + } + } + + #[test] + fn test_invalid_identifier() { + let mut packet = vec![0u8; MIN_PACKET_SIZE]; + packet[0] = b'x'; + let result = parse_rt_car_info(&packet); + assert!(result.is_err()); + if let Err(KsudpParseError::InvalidIdentifier { got }) = result { + assert_eq!(got, b'x'); + } else { + panic!("Expected InvalidIdentifier error"); + } + } + + #[test] + fn test_boolean_flags() { + let mut packet = vec![0u8; MIN_PACKET_SIZE]; + packet[0] = b'a'; + packet[17] = 1; // isAbsEnabled = true + packet[18] = 0; // isAbsInAction = false + packet[19] = 1; // isTcInAction = true + packet[20] = 1; // isTcEnabled = true + packet[21] = 0; // isInPit = false + packet[22] = 1; // isEngineLimiterOn = true + + let frame = parse_rt_car_info(&packet).unwrap(); + assert!(frame.is_abs_enabled); + assert!(!frame.is_abs_in_action); + assert!(frame.is_tc_in_action); + assert!(frame.is_tc_enabled); + assert!(!frame.is_in_pit); + assert!(frame.is_engine_limiter_on); + } + + #[test] + fn test_per_wheel_arrays() { + let mut packet = vec![0u8; MIN_PACKET_SIZE]; + packet[0] = b'a'; + // slipRatio at offset 127: [0.1, 0.2, 0.3, 0.4] + for i in 0..4 { + let val = (i as f32 + 1.0) * 0.1; + packet[127 + i * 4..127 + i * 4 + 4].copy_from_slice(&val.to_le_bytes()); + } + + let frame = parse_rt_car_info(&packet).unwrap(); + assert!((frame.slip_ratio[0] - 0.1).abs() < 0.001); + assert!((frame.slip_ratio[1] - 0.2).abs() < 0.001); + assert!((frame.slip_ratio[2] - 0.3).abs() < 0.001); + assert!((frame.slip_ratio[3] - 0.4).abs() < 0.001); + } + + #[test] + fn test_world_coordinates() { + let mut packet = vec![0u8; MIN_PACKET_SIZE]; + packet[0] = b'a'; + packet[311..315].copy_from_slice(&100.5f32.to_le_bytes()); // x + packet[315..319].copy_from_slice(&2.3f32.to_le_bytes()); // y + packet[319..323].copy_from_slice(&(-50.7f32).to_le_bytes()); // z + + let frame = parse_rt_car_info(&packet).unwrap(); + assert!((frame.world_x - 100.5).abs() < 0.01); + assert!((frame.world_y - 2.3).abs() < 0.01); + assert!((frame.world_z - (-50.7)).abs() < 0.01); + } +} diff --git a/src/parser_telemetry_tool.rs b/src/parser_telemetry_tool.rs new file mode 100644 index 0000000..272222b --- /dev/null +++ b/src/parser_telemetry_tool.rs @@ -0,0 +1,363 @@ +//! Binary parser for Telemetry Tool plugin packets (port 10101). +//! +//! The Telemetry Tool is a third-party AC Python plugin by IkoRein v1.3. +//! It sends binary telemetry for all cars in the session. +//! +//! Reference: `/apps/python/Telemetry_Tool_plugin/Telemetry_Tool_plugin.py` + +use crate::binary_utils::{ + read_bool, read_f32, read_f32_2, read_f32_3, read_f32_4, read_f64, read_i16, read_i32_3, + read_string, read_u32, +}; +use crate::telemetry_frame::{FeedSource, TelemetryFrame}; + +/// Packet type byte indicating telemetry data. +const PACKET_TYPE_TELEMETRY: u8 = 22; +/// Packet type byte indicating version info (sent every ~100 frames). +const PACKET_TYPE_VERSION: u8 = 23; + +/// Size of a single per-driver telemetry structure. +const DRIVER_STRUCT_SIZE: usize = 357; + +/// Validation constants embedded by the plugin. +const TEST_FLOAT: f32 = 666.666; +const TEST_INT: u32 = 666; + +/// Parse error type for Telemetry Tool packets. +#[derive(Debug)] +pub enum TelemetryToolParseError { + /// Packet too short for the declared content. + TooShort { actual: usize, needed: usize }, + /// Unknown packet type. + UnknownPacketType { got: u8 }, +} + +impl std::fmt::Display for TelemetryToolParseError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + TelemetryToolParseError::TooShort { actual, needed } => { + write!(f, "packet too short: {} bytes (needed {})", actual, needed) + } + TelemetryToolParseError::UnknownPacketType { got } => { + write!(f, "unknown packet type: {} (expected 22 or 23)", got) + } + } + } +} + +impl std::error::Error for TelemetryToolParseError {} + +/// Parsed result of a Telemetry Tool packet. +/// +/// Reserved for future multi-driver UI support. +#[allow(dead_code)] +pub enum TelemetryToolPacket { + /// Telemetry data for one or more drivers. + Telemetry(Vec), + /// Version info packet (typically ignored). + Version, +} + +/// Parse a complete Telemetry Tool UDP packet. +/// +/// Returns either a list of parsed driver frames or a version marker. +/// +/// Reserved for future multi-driver UI support. +#[allow(dead_code)] +pub fn parse_packet(data: &[u8]) -> Result { + if data.is_empty() { + return Err(TelemetryToolParseError::TooShort { actual: 0, needed: 1 }); + } + + let packet_type = data[0]; + + match packet_type { + PACKET_TYPE_TELEMETRY => { + let payload = &data[1..]; + if payload.len() < DRIVER_STRUCT_SIZE { + return Err(TelemetryToolParseError::TooShort { + actual: data.len(), + needed: 1 + DRIVER_STRUCT_SIZE, + }); + } + + let num_drivers = payload.len() / DRIVER_STRUCT_SIZE; + let mut drivers = Vec::with_capacity(num_drivers); + for i in 0..num_drivers { + let start = i * DRIVER_STRUCT_SIZE; + let end = start + DRIVER_STRUCT_SIZE; + if end > payload.len() { + break; + } + let driver_data = &payload[start..end]; + drivers.push(parse_driver(driver_data)?); + } + + Ok(TelemetryToolPacket::Telemetry(drivers)) + } + PACKET_TYPE_VERSION => Ok(TelemetryToolPacket::Version), + _ => Err(TelemetryToolParseError::UnknownPacketType { got: packet_type }), + } +} + +/// Parse only the first driver from a Telemetry Tool UDP packet. +/// +/// Avoids allocating a `Vec` for all drivers when only the local player's +/// telemetry is needed. +pub fn parse_first_driver(data: &[u8]) -> Result { + if data.is_empty() { + return Err(TelemetryToolParseError::TooShort { actual: 0, needed: 1 }); + } + + let packet_type = data[0]; + if packet_type != PACKET_TYPE_TELEMETRY { + return Err(TelemetryToolParseError::UnknownPacketType { got: packet_type }); + } + + let payload = &data[1..]; + if payload.len() < DRIVER_STRUCT_SIZE { + return Err(TelemetryToolParseError::TooShort { + actual: data.len(), + needed: 1 + DRIVER_STRUCT_SIZE, + }); + } + + parse_driver(&payload[..DRIVER_STRUCT_SIZE]) +} + +/// Parse a single 357-byte driver structure. +fn parse_driver(data: &[u8]) -> Result { + // Validation: check test_float at offset 93 + let test_float_val = read_f32(data, 93); + if (test_float_val - TEST_FLOAT).abs() > 1.0 { + eprintln!( + "[WARN] Telemetry Tool: test_float validation: expected {}, got {}", + TEST_FLOAT, test_float_val + ); + } + + // Validation: check test_int at offset 177 + let test_int_val = read_u32(data, 177); + if test_int_val != TEST_INT { + eprintln!( + "[WARN] Telemetry Tool: test_int validation: expected {}, got {}", + TEST_INT, test_int_val + ); + } + + let speed_kmh = read_f32(data, 63); + + let frame = TelemetryFrame { + source: FeedSource::TelemetryTool, + raw_size: data.len(), + + // Speed (convert km/h back) + speed_kmh, + speed_mph: speed_kmh * 0.621371, + speed_ms: speed_kmh / 3.6, + + // Engine & drivetrain + engine_rpm: read_f32(data, 59), + gear: read_i16(data, 57) as i32, + + // Driver inputs + steer: read_f32(data, 41), + throttle: read_f32(data, 45), + brake: read_f32(data, 49), + clutch: read_f32(data, 53), + + // Lap timing + lap_time_ms: read_u32(data, 4) as i32, + last_lap_ms: read_u32(data, 8) as i32, + best_lap_ms: read_u32(data, 12) as i32, + lap_count: read_i16(data, 71) as i32, + + // World position + world_x: read_f32(data, 73), + world_y: read_f32(data, 77), + world_z: read_f32(data, 81), + spline_position: read_f64(data, 85) as f32, + + // G-forces + acc_g_vertical: read_f32(data, 185), + acc_g_horizontal: read_f32(data, 189), + acc_g_frontal: read_f32(data, 193), + + // Boolean flags + is_in_pit: read_bool(data, 182), + is_engine_limiter_on: read_bool(data, 196), + + // Per-wheel data + tyre_dirty_level: read_f32_4(data, 213), + slip_ratio: read_f32_4(data, 257), + slip_angle: read_f32_4(data, 273), + nd_slip: read_f32_4(data, 289), + tyre_slip: read_f32_4(data, 305), + + // Telemetry Tool-specific fields + driver_id: Some(read_u32(data, 0)), + driver_name: Some(read_string(data, 97, 33)), + nationality: Some(read_string(data, 130, 4)), + car_name: Some(read_string(data, 134, 33)), + tyre_compound: Some(read_string(data, 167, 10)), + track_position: Some(read_i16(data, 67)), + leaderboard_position: Some(read_i16(data, 69)), + is_connected: Some(read_bool(data, 181)), + is_car_in_pitlane: Some(read_bool(data, 183)), + lap_invalidated: Some(read_bool(data, 184)), + is_drs_available: Some(read_bool(data, 197)), + is_drs_enabled: Some(read_bool(data, 198)), + is_race_finished: Some(read_bool(data, 199)), + turbo_boost: Some(read_f32(data, 229)), + ers_max_kj: Some(read_f32(data, 205)), + ers_current_kj: Some(read_f32(data, 209)), + ride_height_front_rear: Some(read_f32_2(data, 233)), + tyre_core_temp: Some(read_f32_4(data, 241)), + world_velocity: Some(read_f32_3(data, 321)), + local_velocity: Some(read_f32_3(data, 333)), + sector_times: Some(read_i32_3(data, 17)), + last_sector_times: Some(read_i32_3(data, 29)), + + // Remaining KSUDP-only fields stay at default (0.0 / [0.0; 4]) + ..Default::default() + }; + + Ok(frame) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn make_driver_struct() -> Vec { + let mut data = vec![0u8; DRIVER_STRUCT_SIZE]; + + // driver_id = 42 + data[0..4].copy_from_slice(&42u32.to_le_bytes()); + // lap_time = 65000 + data[4..8].copy_from_slice(&65000u32.to_le_bytes()); + // test_float at offset 93 + data[93..97].copy_from_slice(&TEST_FLOAT.to_le_bytes()); + // driver_name at offset 97 + let name = b"TestDriver"; + data[97..97 + name.len()].copy_from_slice(name); + // test_int at offset 177 + data[177..181].copy_from_slice(&TEST_INT.to_le_bytes()); + // speed at offset 63 = 180.5 + data[63..67].copy_from_slice(&180.5f32.to_le_bytes()); + // engineRPM at offset 59 = 7200.0 + data[59..63].copy_from_slice(&7200.0f32.to_le_bytes()); + // gear at offset 57 = 5 + data[57..59].copy_from_slice(&5i16.to_le_bytes()); + // throttle at offset 45 = 0.85 + data[45..49].copy_from_slice(&0.85f32.to_le_bytes()); + // spline_position at offset 85 = 0.5 + data[85..93].copy_from_slice(&0.5f64.to_le_bytes()); + // x at offset 73 + data[73..77].copy_from_slice(&100.0f32.to_le_bytes()); + // is_in_pit at offset 182 = false + data[182] = 0; + + data + } + + #[test] + fn test_parse_telemetry_packet_single_driver() { + let driver = make_driver_struct(); + let mut packet = vec![PACKET_TYPE_TELEMETRY]; + packet.extend_from_slice(&driver); + + let result = parse_packet(&packet).unwrap(); + match result { + TelemetryToolPacket::Telemetry(drivers) => { + assert_eq!(drivers.len(), 1); + let f = &drivers[0]; + assert_eq!(f.source, FeedSource::TelemetryTool); + assert_eq!(f.driver_id, Some(42)); + assert_eq!(f.lap_time_ms, 65000); + assert!((f.speed_kmh - 180.5).abs() < 0.01); + assert!((f.engine_rpm - 7200.0).abs() < 0.01); + assert_eq!(f.gear, 5); + assert!((f.throttle - 0.85).abs() < 0.01); + assert!((f.world_x - 100.0).abs() < 0.01); + assert_eq!(f.driver_name.as_deref(), Some("TestDriver")); + } + TelemetryToolPacket::Version => panic!("Expected Telemetry, got Version"), + } + } + + #[test] + fn test_parse_version_packet() { + let packet = vec![PACKET_TYPE_VERSION, 1, 2, 3]; + let result = parse_packet(&packet).unwrap(); + match result { + TelemetryToolPacket::Version => {} + _ => panic!("Expected Version"), + } + } + + #[test] + fn test_unknown_packet_type() { + let packet = vec![99u8]; + let result = parse_packet(&packet); + assert!(result.is_err()); + if let Err(TelemetryToolParseError::UnknownPacketType { got }) = result { + assert_eq!(got, 99); + } else { + panic!("Expected UnknownPacketType"); + } + } + + #[test] + fn test_empty_packet() { + let result = parse_packet(&[]); + assert!(result.is_err()); + } + + #[test] + fn test_multiple_drivers() { + let driver1 = make_driver_struct(); + let mut driver2 = make_driver_struct(); + // Change driver_id for second driver + driver2[0..4].copy_from_slice(&99u32.to_le_bytes()); + + let mut packet = vec![PACKET_TYPE_TELEMETRY]; + packet.extend_from_slice(&driver1); + packet.extend_from_slice(&driver2); + + let result = parse_packet(&packet).unwrap(); + match result { + TelemetryToolPacket::Telemetry(drivers) => { + assert_eq!(drivers.len(), 2); + assert_eq!(drivers[0].driver_id, Some(42)); + assert_eq!(drivers[1].driver_id, Some(99)); + } + TelemetryToolPacket::Version => panic!("Expected Telemetry"), + } + } + + #[test] + fn test_per_wheel_slip_ratio() { + let mut driver = make_driver_struct(); + // slip_ratio at offset 257: [0.1, 0.2, 0.3, 0.4] + for i in 0..4 { + let val = (i as f32 + 1.0) * 0.1; + driver[257 + i * 4..257 + i * 4 + 4].copy_from_slice(&val.to_le_bytes()); + } + + let mut packet = vec![PACKET_TYPE_TELEMETRY]; + packet.extend_from_slice(&driver); + + let result = parse_packet(&packet).unwrap(); + match result { + TelemetryToolPacket::Telemetry(drivers) => { + let f = &drivers[0]; + assert!((f.slip_ratio[0] - 0.1).abs() < 0.001); + assert!((f.slip_ratio[1] - 0.2).abs() < 0.001); + assert!((f.slip_ratio[2] - 0.3).abs() < 0.001); + assert!((f.slip_ratio[3] - 0.4).abs() < 0.001); + } + TelemetryToolPacket::Version => panic!("Expected Telemetry"), + } + } +} diff --git a/src/stats.rs b/src/stats.rs deleted file mode 100644 index eb53e21..0000000 --- a/src/stats.rs +++ /dev/null @@ -1,70 +0,0 @@ -use std::fmt; -use std::time::Instant; - -pub struct ReceiveStats { - pub total_packets: u64, - pub total_bytes: u64, - pub start_time: Option, - pub last_packet_time: Option, -} - -impl ReceiveStats { - pub fn new() -> Self { - Self { - total_packets: 0, - total_bytes: 0, - start_time: None, - last_packet_time: None, - } - } - - pub fn record(&mut self, bytes_received: usize) { - let now = Instant::now(); - if self.start_time.is_none() { - self.start_time = Some(now); - } - self.last_packet_time = Some(now); - self.total_packets += 1; - self.total_bytes += bytes_received as u64; - } - - pub fn packets_per_second(&self) -> f64 { - match (self.start_time, self.last_packet_time) { - (Some(start), Some(last)) => { - let elapsed = last.duration_since(start).as_secs_f64(); - if elapsed > 0.0 { - self.total_packets as f64 / elapsed - } else { - 0.0 - } - } - _ => 0.0, - } - } - - pub fn data_rate_kbps(&self) -> f64 { - match self.start_time { - Some(start) => { - let elapsed = start.elapsed().as_secs_f64(); - if elapsed > 0.0 { - (self.total_bytes as f64 / 1024.0) / elapsed - } else { - 0.0 - } - } - _ => 0.0, - } - } -} - -impl fmt::Display for ReceiveStats { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - write!( - f, - "Packets: {} | Rate: {:.1} pkt/s | Data: {:.1} KB/s", - self.total_packets, - self.packets_per_second(), - self.data_rate_kbps(), - ) - } -} diff --git a/src/telemetry.rs b/src/telemetry.rs deleted file mode 100644 index ec0cc57..0000000 --- a/src/telemetry.rs +++ /dev/null @@ -1,56 +0,0 @@ -use serde::Deserialize; -use std::fmt; - -#[derive(Debug, Deserialize)] -#[serde(deny_unknown_fields)] -pub struct TelemetryFrame { - pub timestamp: f64, - pub speed_kmh: f64, - pub rpm: f64, - pub gear: i32, - pub throttle: f64, - pub brake: f64, - pub steering: f64, - pub g_forces: GForces, - pub wheel_slip: [f64; 4], - pub suspension_travel: [f64; 4], - pub lap_time: f64, - pub lap_count: i32, -} - -#[derive(Debug, Deserialize)] -#[serde(deny_unknown_fields)] -pub struct GForces { - pub x: f64, - pub y: f64, - pub z: f64, -} - -impl fmt::Display for TelemetryFrame { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - write!( - f, - "[Frame @ {:.3}]\n Speed: {:.1} km/h | RPM: {:.0} | Gear: {}\n Throttle: {:.1}% | Brake: {:.1}% | Steering: {:.1}%\n G-Forces: X={:.2} Y={:.2} Z={:.2}\n Wheel Slip: [{:.2}, {:.2}, {:.2}, {:.2}]\n Suspension: [{:.2}, {:.2}, {:.2}, {:.2}]\n Lap: #{} | Lap Time: {:.3}s", - self.timestamp, - self.speed_kmh, - self.rpm, - self.gear, - self.throttle * 100.0, - self.brake * 100.0, - self.steering * 100.0, - self.g_forces.x, - self.g_forces.y, - self.g_forces.z, - self.wheel_slip[0], - self.wheel_slip[1], - self.wheel_slip[2], - self.wheel_slip[3], - self.suspension_travel[0], - self.suspension_travel[1], - self.suspension_travel[2], - self.suspension_travel[3], - self.lap_count, - self.lap_time, - ) - } -} diff --git a/src/telemetry_frame.rs b/src/telemetry_frame.rs new file mode 100644 index 0000000..1fab9e2 --- /dev/null +++ b/src/telemetry_frame.rs @@ -0,0 +1,218 @@ +//! Generic telemetry frame model — unified representation of parsed sim racing data. +//! +//! Both the KSUDP (port 9996) and Telemetry Tool (port 10101) parsers produce +//! `TelemetryFrame` instances from their raw binary data. + +use std::fmt; + +/// Identifies which feed source produced this frame. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum FeedSource { + Ksudp, + TelemetryTool, +} + +/// Per-wheel index convention: FL=0, FR=1, RL=2, RR=3. +pub const NUM_WHEELS: usize = 4; + +/// A single parsed telemetry snapshot from any supported feed. +/// +/// Fields present in both feeds are stored directly; fields unique to one feed +/// are wrapped in `Option`. +#[derive(Debug, Clone)] +pub struct TelemetryFrame { + // ── Speed ────────────────────────────────────────────────────── + pub speed_kmh: f32, + pub speed_mph: f32, + pub speed_ms: f32, + + // ── Engine & Drivetrain ──────────────────────────────────────── + pub engine_rpm: f32, + pub gear: i32, + + // ── Driver Inputs ────────────────────────────────────────────── + pub throttle: f32, + pub brake: f32, + pub clutch: f32, + pub steer: f32, + + // ── Lap Timing ───────────────────────────────────────────────── + pub lap_time_ms: i32, + pub last_lap_ms: i32, + pub best_lap_ms: i32, + pub lap_count: i32, + + // ── World Position ───────────────────────────────────────────── + pub world_x: f32, + pub world_y: f32, + pub world_z: f32, + pub spline_position: f32, + + // ── Acceleration (G-forces) ──────────────────────────────────── + pub acc_g_vertical: f32, + pub acc_g_horizontal: f32, + pub acc_g_frontal: f32, + + // ── Boolean Flags ────────────────────────────────────────────── + pub is_abs_enabled: bool, + pub is_abs_in_action: bool, + pub is_tc_enabled: bool, + pub is_tc_in_action: bool, + pub is_in_pit: bool, + pub is_engine_limiter_on: bool, + + // ── Per-Wheel Data (FL, FR, RL, RR) ──────────────────────────── + pub wheel_angular_speed: [f32; NUM_WHEELS], + pub slip_angle: [f32; NUM_WHEELS], + pub slip_ratio: [f32; NUM_WHEELS], + pub tyre_slip: [f32; NUM_WHEELS], + pub nd_slip: [f32; NUM_WHEELS], + pub suspension_height: [f32; NUM_WHEELS], + pub tyre_dirty_level: [f32; NUM_WHEELS], + + // ── KSUDP-specific fields ────────────────────────────────────── + pub slip_angle_contact_patch: [f32; NUM_WHEELS], + pub wheel_load: [f32; NUM_WHEELS], + pub wheel_dy: [f32; NUM_WHEELS], + pub wheel_mz: [f32; NUM_WHEELS], + pub camber_rad: [f32; NUM_WHEELS], + pub tyre_radius: [f32; NUM_WHEELS], + pub tyre_loaded_radius: [f32; NUM_WHEELS], + pub cg_height: f32, + pub car_slope: f32, + pub car_position_normalized: f32, + + // ── Telemetry Tool-specific fields ───────────────────────────── + pub driver_name: Option, + pub car_name: Option, + pub nationality: Option, + pub tyre_compound: Option, + pub track_position: Option, + pub leaderboard_position: Option, + pub is_connected: Option, + pub is_car_in_pitlane: Option, + pub lap_invalidated: Option, + pub is_drs_available: Option, + pub is_drs_enabled: Option, + pub is_race_finished: Option, + pub turbo_boost: Option, + pub ers_max_kj: Option, + pub ers_current_kj: Option, + pub ride_height_front_rear: Option<[f32; 2]>, + pub tyre_core_temp: Option<[f32; NUM_WHEELS]>, + pub world_velocity: Option<[f32; 3]>, + pub local_velocity: Option<[f32; 3]>, + pub sector_times: Option<[i32; 3]>, + pub last_sector_times: Option<[i32; 3]>, + pub driver_id: Option, + + // ── Meta ─────────────────────────────────────────────────────── + pub source: FeedSource, + pub raw_size: usize, +} + +impl Default for TelemetryFrame { + fn default() -> Self { + Self { + speed_kmh: 0.0, + speed_mph: 0.0, + speed_ms: 0.0, + engine_rpm: 0.0, + gear: 0, + throttle: 0.0, + brake: 0.0, + clutch: 0.0, + steer: 0.0, + lap_time_ms: 0, + last_lap_ms: 0, + best_lap_ms: 0, + lap_count: 0, + world_x: 0.0, + world_y: 0.0, + world_z: 0.0, + spline_position: 0.0, + acc_g_vertical: 0.0, + acc_g_horizontal: 0.0, + acc_g_frontal: 0.0, + is_abs_enabled: false, + is_abs_in_action: false, + is_tc_enabled: false, + is_tc_in_action: false, + is_in_pit: false, + is_engine_limiter_on: false, + wheel_angular_speed: [0.0; NUM_WHEELS], + slip_angle: [0.0; NUM_WHEELS], + slip_ratio: [0.0; NUM_WHEELS], + tyre_slip: [0.0; NUM_WHEELS], + nd_slip: [0.0; NUM_WHEELS], + suspension_height: [0.0; NUM_WHEELS], + tyre_dirty_level: [0.0; NUM_WHEELS], + slip_angle_contact_patch: [0.0; NUM_WHEELS], + wheel_load: [0.0; NUM_WHEELS], + wheel_dy: [0.0; NUM_WHEELS], + wheel_mz: [0.0; NUM_WHEELS], + camber_rad: [0.0; NUM_WHEELS], + tyre_radius: [0.0; NUM_WHEELS], + tyre_loaded_radius: [0.0; NUM_WHEELS], + cg_height: 0.0, + car_slope: 0.0, + car_position_normalized: 0.0, + driver_name: None, + car_name: None, + nationality: None, + tyre_compound: None, + track_position: None, + leaderboard_position: None, + is_connected: None, + is_car_in_pitlane: None, + lap_invalidated: None, + is_drs_available: None, + is_drs_enabled: None, + is_race_finished: None, + turbo_boost: None, + ers_max_kj: None, + ers_current_kj: None, + ride_height_front_rear: None, + tyre_core_temp: None, + world_velocity: None, + local_velocity: None, + sector_times: None, + last_sector_times: None, + driver_id: None, + source: FeedSource::Ksudp, + raw_size: 0, + } + } +} + +impl TelemetryFrame { + /// Format lap time from milliseconds as M:SS.mmm + pub fn format_lap_time(ms: i32) -> String { + if ms <= 0 { + return "-:--.---".to_string(); + } + let total_ms = ms as u32; + let minutes = total_ms / 60000; + let seconds = (total_ms % 60000) / 1000; + let millis = total_ms % 1000; + format!("{}:{:02}.{:03}", minutes, seconds, millis) + } + + /// Gear display: 0=Reverse, 1=Neutral, 2+=Gear N-1 + pub fn gear_display(&self) -> String { + match self.gear { + 0 => "R".to_string(), + 1 => "N".to_string(), + n => format!("{}", n - 1), + } + } +} + +impl fmt::Display for FeedSource { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + FeedSource::Ksudp => write!(f, "KSUDP"), + FeedSource::TelemetryTool => write!(f, "Telemetry Tool"), + } + } +} diff --git a/src/ui.rs b/src/ui.rs new file mode 100644 index 0000000..8893477 --- /dev/null +++ b/src/ui.rs @@ -0,0 +1,222 @@ +use ratatui::{ + Frame, + layout::{Constraint, Direction, Layout, Rect}, + style::{Color, Modifier, Style}, + text::{Line, Span}, + widgets::{Block, Borders, Paragraph, Wrap}, +}; + +use crate::feed_state::{FeedHealth, FeedMode, FeedState, HandshakePhase}; +use crate::telemetry_frame::TelemetryFrame; +use std::sync::{Arc, Mutex}; + +#[derive(Clone)] +pub struct FeedDisplayInfo { + pub label: &'static str, + pub port: u16, + pub mode: FeedMode, +} + +fn health_indicator(health: FeedHealth) -> Span<'static> { + match health { + FeedHealth::Alive => Span::styled( + "*".to_string(), + Style::default().fg(Color::Green).add_modifier(Modifier::BOLD), + ), + FeedHealth::Stale => Span::styled( + "~".to_string(), + Style::default().fg(Color::Yellow), + ), + FeedHealth::Waiting => Span::styled( + ".".to_string(), + Style::default().fg(Color::DarkGray), + ), + FeedHealth::Error => Span::styled( + "!".to_string(), + Style::default().fg(Color::Red).add_modifier(Modifier::BOLD), + ), + } +} + +fn freq_style(health: FeedHealth) -> Style { + match health { + FeedHealth::Alive => Style::default().fg(Color::Green), + FeedHealth::Stale => Style::default().fg(Color::Yellow), + _ => Style::default().fg(Color::DarkGray), + } +} + +fn format_ago(ago: Option) -> String { + match ago { + None => "never".to_string(), + Some(t) => format!("{:.2}s", t), + } +} + +fn feed_block_title(info: &FeedDisplayInfo, state: &FeedState) -> String { + match info.mode { + FeedMode::Handshake { + target_port, + recv_port, + } => { + let phase_str = match state.handshake_phase { + HandshakePhase::NotStarted => "not started", + HandshakePhase::WaitingForResponse => "waiting", + HandshakePhase::Subscribing => "subscribing", + HandshakePhase::Subscribed => "subscribed", + }; + format!( + " {} -> :{} recv :{} [{}] ", + info.label, target_port, recv_port, phase_str + ) + } + FeedMode::BindListen => { + format!(" {} Port :{} ", info.label, info.port) + } + FeedMode::Disabled => { + format!(" {} Port :{} [optional, not available] ", info.label, info.port) + } + } +} + +fn render_feed_panel(frame: &mut Frame, area: Rect, info: &FeedDisplayInfo, state: &FeedState) { + let title = feed_block_title(info, state); + let border_style = match state.health { + FeedHealth::Alive => Style::default().fg(Color::Green), + FeedHealth::Stale => Style::default().fg(Color::Yellow), + FeedHealth::Error => Style::default().fg(Color::Red), + FeedHealth::Waiting => { + if info.mode == FeedMode::Disabled { + Style::default().fg(Color::DarkGray).add_modifier(Modifier::DIM) + } else { + Style::default().fg(Color::DarkGray) + } + } + }; + + let block = Block::default() + .title(title) + .borders(Borders::ALL) + .border_style(border_style); + let inner = block.inner(area); + frame.render_widget(block, area); + + let indicator = health_indicator(state.health); + let status_text = Span::raw(format!(" {:<12}", state.status_text)); + + let line1 = Line::from(vec![indicator, status_text]); + + let freq_span = Span::styled( + format!("{:>7.1} Hz", state.frequency_hz), + freq_style(state.health), + ); + let packets_span = Span::raw(format!(" {:>8} packets", state.total_packets)); + let rate_span = Span::raw(format!(" {:>7.1} KB/s", state.data_rate_kb_s)); + let ago_span = Span::raw(format!(" last: {:>8}", format_ago(state.last_packet_ago_secs))); + + let line2 = Line::from(vec![freq_span, packets_span, rate_span, ago_span]); + + let mut lines = vec![line1, line2]; + + if let Some(ref frame) = state.last_frame { + let speed_span = Span::styled( + format!(" {:>6.0} km/h", frame.speed_kmh), + Style::default().fg(Color::White).add_modifier(Modifier::BOLD), + ); + let gear_span = Span::styled( + format!(" Gear:{}", frame.gear_display()), + Style::default().fg(Color::Yellow), + ); + let rpm_span = Span::raw(format!(" {:.0} RPM", frame.engine_rpm)); + let lap_span = Span::raw(format!(" Lap:{}", TelemetryFrame::format_lap_time(frame.lap_time_ms))); + let line3 = Line::from(vec![speed_span, gear_span, rpm_span, lap_span]); + lines.push(line3); + + let throttle_pct = (frame.throttle.clamp(0.0, 1.0) * 100.0) as u8; + let brake_pct = (frame.brake.clamp(0.0, 1.0) * 100.0) as u8; + let inputs_span = Span::raw(format!(" THR:{:>3}% BRK:{:>3}%", throttle_pct, brake_pct)); + let pos_span = Span::raw(format!(" Pos:{:.3}", frame.spline_position)); + let line4 = Line::from(vec![inputs_span, pos_span]); + lines.push(line4); + } + + let paragraph = Paragraph::new(lines); + frame.render_widget(paragraph, inner); +} + +pub fn render( + frame: &mut Frame, + feeds: &[(FeedDisplayInfo, Arc>)], + session_dir: &str, + version: &str, +) { + let total_feeds = feeds.len() as u16; + + let outer = Layout::default() + .direction(Direction::Vertical) + .constraints([ + Constraint::Length(3), + Constraint::Min(total_feeds * 4 + 2), + Constraint::Length(3), + ]) + .split(frame.area()); + + let title_block = Block::default() + .borders(Borders::ALL) + .border_style(Style::default().fg(Color::Cyan)); + let title_inner = title_block.inner(outer[0]); + frame.render_widget(title_block, outer[0]); + + let title_text = Line::from(Span::styled( + format!( + " rusty-telemetry v{} -- Telemetry Listener ", + version + ), + Style::default() + .fg(Color::Cyan) + .add_modifier(Modifier::BOLD), + )); + let title_para = Paragraph::new(title_text); + frame.render_widget(title_para, title_inner); + + let feed_constraints: Vec = feeds + .iter() + .map(|_| Constraint::Min(6)) + .collect(); + let feed_layout = Layout::default() + .direction(Direction::Vertical) + .constraints(feed_constraints) + .split(outer[1]); + + for (i, (info, state)) in feeds.iter().enumerate() { + // Clone inside the lock, then drop the guard before rendering + // to avoid blocking the UDP receive task during the entire draw. + let snapshot = state.lock().ok().map(|mut s| { + s.tick(); + s.clone() + }); + if let Some(snap) = snapshot { + render_feed_panel(frame, feed_layout[i], info, &snap); + } + } + + let footer_block = Block::default() + .borders(Borders::ALL) + .border_style(Style::default().fg(Color::DarkGray)); + let footer_inner = footer_block.inner(outer[2]); + frame.render_widget(footer_block, outer[2]); + + let footer_text = Line::from(vec![ + Span::styled( + format!(" Session: {}/ ", session_dir), + Style::default().fg(Color::White), + ), + Span::raw(" "), + Span::styled( + "Ctrl+C Quit".to_string(), + Style::default().fg(Color::DarkGray), + ), + ]); + let footer_para = Paragraph::new(footer_text).wrap(Wrap { trim: true }); + frame.render_widget(footer_para, footer_inner); +}