diff --git a/Cargo.lock b/Cargo.lock index c93c535..f832b4a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -17,12 +17,76 @@ dependencies = [ "libc", ] +[[package]] +name = "anyhow" +version = "1.0.102" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7f202df86484c868dbad7eaa557ef785d5c66295e41b460ef922eca0723b842c" + +[[package]] +name = "atomic-waker" +version = "1.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1505bd5d3d116872e7271a6d4e16d81d0c8570876c8de68093a09ac269d8aac0" + [[package]] name = "autocfg" version = "1.5.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f2032f911046de80f0a198e0901378627c33f59ea0ac00e363d481118bd70a53" +[[package]] +name = "axum" +version = "0.8.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "31b698c5f9a010f6573133b09e0de5408834d0c82f8d7475a89fc1867a71cd90" +dependencies = [ + "axum-core", + "bytes", + "form_urlencoded", + "futures-util", + "http", + "http-body", + "http-body-util", + "hyper", + "hyper-util", + "itoa", + "matchit", + "memchr", + "mime", + "percent-encoding", + "pin-project-lite", + "serde_core", + "serde_json", + "serde_path_to_error", + "serde_urlencoded", + "sync_wrapper", + "tokio", + "tower", + "tower-layer", + "tower-service", + "tracing", +] + +[[package]] +name = "axum-core" +version = "0.5.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "08c78f31d7b1291f7ee735c1c6780ccde7785daae9a9206026862dab7d8792d1" +dependencies = [ + "bytes", + "futures-core", + "http", + "http-body", + "http-body-util", + "mime", + "pin-project-lite", + "sync_wrapper", + "tower-layer", + "tower-service", + "tracing", +] + [[package]] name = "bitflags" version = "2.11.1" @@ -81,6 +145,7 @@ dependencies = [ "iana-time-zone", "js-sys", "num-traits", + "serde", "wasm-bindgen", "windows-link", ] @@ -198,6 +263,15 @@ version = "0.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d9c4f5dac5e15c24eb999c26181a6ca40b39fe946cbe4c263c7209467bc83af2" +[[package]] +name = "form_urlencoded" +version = "1.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cb4cb245038516f5f85277875cdaa4f7d2c9a0fa0468de06ed190163b1581fcf" +dependencies = [ + "percent-encoding", +] + [[package]] name = "futures" version = "0.3.32" @@ -286,6 +360,19 @@ dependencies = [ "slab", ] +[[package]] +name = "getrandom" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0de51e6874e94e7bf76d726fc5d13ba782deca734ff60d5bb2fb2607c7406555" +dependencies = [ + "cfg-if", + "libc", + "r-efi", + "wasip2", + "wasip3", +] + [[package]] name = "hashbrown" version = "0.15.5" @@ -297,12 +384,98 @@ dependencies = [ "foldhash", ] +[[package]] +name = "hashbrown" +version = "0.17.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a" + [[package]] name = "heck" version = "0.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea" +[[package]] +name = "http" +version = "1.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8be7462df143984c4598a256ef469b251d7d7f9e271135073e78fc535414f3d0" +dependencies = [ + "bytes", + "itoa", +] + +[[package]] +name = "http-body" +version = "1.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1efedce1fb8e6913f23e0c92de8e62cd5b772a67e7b3946df930a62566c93184" +dependencies = [ + "bytes", + "http", +] + +[[package]] +name = "http-body-util" +version = "0.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b021d93e26becf5dc7e1b75b1bed1fd93124b374ceb73f43d4d4eafec896a64a" +dependencies = [ + "bytes", + "futures-core", + "http", + "http-body", + "pin-project-lite", +] + +[[package]] +name = "httparse" +version = "1.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6dbf3de79e51f3d586ab4cb9d5c3e2c14aa28ed23d180cf89b4df0454a69cc87" + +[[package]] +name = "httpdate" +version = "1.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9" + +[[package]] +name = "hyper" +version = "1.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "55281c53a1894c864990125767da440a4e630446785086f52523b20033b74498" +dependencies = [ + "atomic-waker", + "bytes", + "futures-channel", + "futures-core", + "http", + "http-body", + "httparse", + "httpdate", + "itoa", + "pin-project-lite", + "smallvec", + "tokio", +] + +[[package]] +name = "hyper-util" +version = "0.1.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "96547c2556ec9d12fb1578c4eaf448b04993e7fb79cbaad930a656880a6bdfa0" +dependencies = [ + "bytes", + "http", + "http-body", + "hyper", + "pin-project-lite", + "tokio", + "tower-service", +] + [[package]] name = "iana-time-zone" version = "0.1.65" @@ -327,12 +500,30 @@ dependencies = [ "cc", ] +[[package]] +name = "id-arena" +version = "2.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3d3067d79b975e8844ca9eb072e16b31c3c1c36928edf9c6789548c524d0d954" + [[package]] name = "ident_case" version = "1.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b9e0384b61958566e926dc50660321d12159025e767c18e043daf26b70104c39" +[[package]] +name = "indexmap" +version = "2.14.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d466e9454f08e4a911e14806c24e16fba1b4c121d1ea474396f396069cf949d9" +dependencies = [ + "equivalent", + "hashbrown 0.17.1", + "serde", + "serde_core", +] + [[package]] name = "indoc" version = "2.0.7" @@ -382,6 +573,12 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "leb128fmt" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "09edd9e8b54e49e587e4f6295a7d29c3ea94d469cb40ab8ca70b288248a81db2" + [[package]] name = "libc" version = "0.2.186" @@ -415,15 +612,27 @@ version = "0.12.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "234cf4f4a04dc1f57e24b96cc0cd600cf2af460d4161ac5ecdd0af8e1f3b2a38" dependencies = [ - "hashbrown", + "hashbrown 0.15.5", ] +[[package]] +name = "matchit" +version = "0.8.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "47e1ffaa40ddd1f3ed91f717a33c8c0ee23fff369e3aa8772b9605cc1d22f4c3" + [[package]] name = "memchr" version = "2.8.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6b947ae49db0d222b1dbc6b113ce7248a3fc3a6ca21b696717bfc000ba4484d8" +[[package]] +name = "mime" +version = "0.3.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6877bb514081ee2a7ff5ef9de3281f14a4dd4bceac4c09388074a6b5df8a139a" + [[package]] name = "mio" version = "1.2.1" @@ -480,12 +689,28 @@ version = "1.0.15" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "57c0d7b74b563b49d38dae00a0c37d4d6de9b432382b2892f0574ddcae73fd0a" +[[package]] +name = "percent-encoding" +version = "2.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9b4f627cb1b25917193a259e49bdad08f671f8d9708acfd5fe0a8c1455d87220" + [[package]] name = "pin-project-lite" version = "0.2.17" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" +[[package]] +name = "prettyplease" +version = "0.2.37" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "479ca8adacdd7ce8f1fb39ce9ecccbfe93a3f1344b3d0d97f20bc0196208f62b" +dependencies = [ + "proc-macro2", + "syn", +] + [[package]] name = "proc-macro2" version = "1.0.106" @@ -504,6 +729,12 @@ dependencies = [ "proc-macro2", ] +[[package]] +name = "r-efi" +version = "6.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf" + [[package]] name = "ratatui" version = "0.29.0" @@ -555,13 +786,18 @@ checksum = "b39cdef0fa800fc44525c84ccb54a029961a8215f9619753635a9c0d2538d46d" [[package]] name = "rusty-telemetry" -version = "0.4.0" +version = "0.5.0" dependencies = [ + "axum", "chrono", "crossterm", "futures", "ratatui", + "serde", + "serde_json", "tokio", + "tower-http", + "uuid", ] [[package]] @@ -576,6 +812,78 @@ version = "1.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49" +[[package]] +name = "semver" +version = "1.0.28" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8a7852d02fc848982e0c167ef163aaff9cd91dc640ba85e263cb1ce46fae51cd" + +[[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 = "serde_path_to_error" +version = "0.1.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "10a9ff822e371bb5403e391ecd83e182e0e77ba7f6fe0160b795797109d1b457" +dependencies = [ + "itoa", + "serde", + "serde_core", +] + +[[package]] +name = "serde_urlencoded" +version = "0.7.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d3491c14715ca2294c4d6a88f15e84739788c1d030eed8c110436aafdaa2f3fd" +dependencies = [ + "form_urlencoded", + "itoa", + "ryu", + "serde", +] + [[package]] name = "shlex" version = "2.0.1" @@ -680,6 +988,12 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "sync_wrapper" +version = "1.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0bf256ce5efdfa370213c1dabab5935a12e49f2c58d15e9eac2870d3b4f27263" + [[package]] name = "tokio" version = "1.52.3" @@ -708,6 +1022,68 @@ dependencies = [ "syn", ] +[[package]] +name = "tower" +version = "0.5.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ebe5ef63511595f1344e2d5cfa636d973292adc0eec1f0ad45fae9f0851ab1d4" +dependencies = [ + "futures-core", + "futures-util", + "pin-project-lite", + "sync_wrapper", + "tokio", + "tower-layer", + "tower-service", + "tracing", +] + +[[package]] +name = "tower-http" +version = "0.6.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4cfcf7e2740e6fc6d4d688b4ef00650406bb94adf4731e43c096c3a19fe40840" +dependencies = [ + "bitflags", + "bytes", + "http", + "pin-project-lite", + "tower-layer", + "tower-service", +] + +[[package]] +name = "tower-layer" +version = "0.3.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "121c2a6cda46980bb0fcd1647ffaf6cd3fc79a013de288782836f6df9c48780e" + +[[package]] +name = "tower-service" +version = "0.3.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8df9b6e13f2d32c91b9bd719c00d1958837bc7dec474d94952798cc8e69eeec3" + +[[package]] +name = "tracing" +version = "0.1.44" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "63e71662fa4b2a2c3a26f570f037eb95bb1f85397f3cd8076caed2f026a6d100" +dependencies = [ + "log", + "pin-project-lite", + "tracing-core", +] + +[[package]] +name = "tracing-core" +version = "0.1.36" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "db97caf9d906fbde555dd62fa95ddba9eecfd14cb388e4f491a66d74cd5fb79a" +dependencies = [ + "once_cell", +] + [[package]] name = "unicode-ident" version = "1.0.24" @@ -743,12 +1119,48 @@ version = "0.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1fc81956842c57dac11422a97c3b8195a1ff727f06e85c84ed2e8aa277c9a0fd" +[[package]] +name = "unicode-xid" +version = "0.2.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ebc1c04c71510c7f702b52b7c350734c9ff1295c464a03335b00bb84fc54f853" + +[[package]] +name = "uuid" +version = "1.23.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d258b83ceec21034727ecee8c382cfa6c3e133699b0742c64571814fb420c9f7" +dependencies = [ + "getrandom", + "js-sys", + "serde_core", + "wasm-bindgen", +] + [[package]] name = "wasi" version = "0.11.1+wasi-snapshot-preview1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b" +[[package]] +name = "wasip2" +version = "1.0.3+wasi-0.2.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "20064672db26d7cdc89c7798c48a0fdfac8213434a1186e5ef29fd560ae223d6" +dependencies = [ + "wit-bindgen 0.57.1", +] + +[[package]] +name = "wasip3" +version = "0.4.0+wasi-0.3.0-rc-2026-01-06" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5428f8bf88ea5ddc08faddef2ac4a67e390b88186c703ce6dbd955e1c145aca5" +dependencies = [ + "wit-bindgen 0.51.0", +] + [[package]] name = "wasm-bindgen" version = "0.2.122" @@ -794,6 +1206,40 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "wasm-encoder" +version = "0.244.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "990065f2fe63003fe337b932cfb5e3b80e0b4d0f5ff650e6985b1048f62c8319" +dependencies = [ + "leb128fmt", + "wasmparser", +] + +[[package]] +name = "wasm-metadata" +version = "0.244.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bb0e353e6a2fbdc176932bbaab493762eb1255a7900fe0fea1a2f96c296cc909" +dependencies = [ + "anyhow", + "indexmap", + "wasm-encoder", + "wasmparser", +] + +[[package]] +name = "wasmparser" +version = "0.244.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "47b807c72e1bac69382b3a6fb3dbe8ea4c0ed87ff5629b8685ae6b9a611028fe" +dependencies = [ + "bitflags", + "hashbrown 0.15.5", + "indexmap", + "semver", +] + [[package]] name = "winapi" version = "0.3.9" @@ -956,3 +1402,103 @@ name = "windows_x86_64_msvc" version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec" + +[[package]] +name = "wit-bindgen" +version = "0.51.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d7249219f66ced02969388cf2bb044a09756a083d0fab1e566056b04d9fbcaa5" +dependencies = [ + "wit-bindgen-rust-macro", +] + +[[package]] +name = "wit-bindgen" +version = "0.57.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1ebf944e87a7c253233ad6766e082e3cd714b5d03812acc24c318f549614536e" + +[[package]] +name = "wit-bindgen-core" +version = "0.51.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ea61de684c3ea68cb082b7a88508a8b27fcc8b797d738bfc99a82facf1d752dc" +dependencies = [ + "anyhow", + "heck", + "wit-parser", +] + +[[package]] +name = "wit-bindgen-rust" +version = "0.51.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b7c566e0f4b284dd6561c786d9cb0142da491f46a9fbed79ea69cdad5db17f21" +dependencies = [ + "anyhow", + "heck", + "indexmap", + "prettyplease", + "syn", + "wasm-metadata", + "wit-bindgen-core", + "wit-component", +] + +[[package]] +name = "wit-bindgen-rust-macro" +version = "0.51.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0c0f9bfd77e6a48eccf51359e3ae77140a7f50b1e2ebfe62422d8afdaffab17a" +dependencies = [ + "anyhow", + "prettyplease", + "proc-macro2", + "quote", + "syn", + "wit-bindgen-core", + "wit-bindgen-rust", +] + +[[package]] +name = "wit-component" +version = "0.244.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9d66ea20e9553b30172b5e831994e35fbde2d165325bec84fc43dbf6f4eb9cb2" +dependencies = [ + "anyhow", + "bitflags", + "indexmap", + "log", + "serde", + "serde_derive", + "serde_json", + "wasm-encoder", + "wasm-metadata", + "wasmparser", + "wit-parser", +] + +[[package]] +name = "wit-parser" +version = "0.244.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ecc8ac4bc1dc3381b7f59c34f00b67e18f910c2c0f50015669dde7def656a736" +dependencies = [ + "anyhow", + "id-arena", + "indexmap", + "log", + "semver", + "serde", + "serde_derive", + "serde_json", + "unicode-xid", + "wasmparser", +] + +[[package]] +name = "zmij" +version = "1.0.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b8848ee67ecc8aedbaf3e4122217aff892639231befc6a1b58d29fff4c2cabaa" diff --git a/Cargo.toml b/Cargo.toml index b34e82c..62c801d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,12 +1,17 @@ [package] name = "rusty-telemetry" -version = "0.4.0" +version = "0.5.0" edition = "2021" description = "Sim racing telemetry receiver and analyzer" [dependencies] tokio = { version = "1", features = ["full"] } -chrono = "0.4" +chrono = { version = "0.4", features = ["serde"] } futures = "0.3" ratatui = "0.29" crossterm = "0.28" +axum = "0.8" +serde = { version = "1", features = ["derive"] } +serde_json = "1" +uuid = { version = "1", features = ["v4", "serde"] } +tower-http = { version = "0.6", features = ["cors"] } diff --git a/README.md b/README.md index 298086b..3a54935 100644 --- a/README.md +++ b/README.md @@ -32,6 +32,7 @@ Press **q** or **Ctrl+C** to quit. | `RUSTY_TELEMETRY_GAME_HOST` | IP of the machine running the sim game (used for KSUDP handshake target) | `127.0.0.1` | | `RUSTY_TELEMETRY_LISTEN` | Local address to bind for incoming UDP feeds (Telemetry Tool, PCARS) | `0.0.0.0` | | `RUSTY_TELEMETRY_BIND` | Fallback for both of the above (backward compatibility) | — | +| `RUSTY_TELEMETRY_API_PORT` | REST API server port | `8080` | ### Same Machine @@ -98,7 +99,7 @@ The terminal UI shows one panel per active feed: ``` ┌──────────────────────────────────────────────────────────┐ -│ rusty-telemetry v0.4.0 -- Telemetry Listener │ +│ rusty-telemetry v0.5.0 -- Telemetry Listener │ ├──────────────────────────────────────────────────────────┤ │ ╔ * Receiving ╗ │ │ ║ 20.0 Hz 1200 packets 15.2 KB/s last: 0.05s ║ @@ -116,7 +117,7 @@ The terminal UI shows one panel per active feed: │ ║ 0.0 Hz 0 packets 0.0 KB/s last: never ║ │ ╚ PCARS Port :5606 ╝ │ ├──────────────────────────────────────────────────────────┤ -│ Session: ./data/20260604_103000/ Ctrl+C Quit │ +│ Recording: 550e8400-… Ctrl+C Quit │ └──────────────────────────────────────────────────────────┘ ``` @@ -128,16 +129,76 @@ Each panel shows: ## Data Capture -All raw UDP packets are written to per-feed binary dump files: +A recording is auto-started on launch. All raw UDP packets are written to per-feed JSONL files under a UUID-based directory: ``` -./data// -├── feed_9996_ksudp.bin -├── feed_10101_telemetry_tool.bin -└── feed_5606_pcars.bin +./data// +├── manifest.json +├── feed_9996_ksudp.jsonl +├── feed_10101_telemetry_tool.jsonl +└── feed_5606_pcars.jsonl ``` -Each packet is prefixed with a 4-byte big-endian length header, allowing sequential replay of captured sessions. +Each recording contains a `manifest.json` with metadata (game, track, car, use case, status, timestamps) and JSONL files for each feed. Each line in a JSONL file is a serialized `TelemetryFrame` object, enabling sequential replay and analysis of captured sessions. + +## REST API + +A built-in HTTP API server runs on port 8080 for recording management and live telemetry access. + +### Endpoints + +| Method | Path | Description | +|--------|------|-------------| +| GET | `/api/live` | Live telemetry snapshot (all feeds) | +| GET | `/api/recordings` | List all recordings | +| POST | `/api/recordings` | Start new recording | +| GET | `/api/recordings/{id}` | Get recording details | +| POST | `/api/recordings/stop` | Stop active recording | +| POST | `/api/recordings/{id}/stop` | Stop specific recording | +| DELETE | `/api/recordings/{id}` | Delete recording | +| GET | `/api/recordings/{id}/data/{feed_name}` | Download raw feed JSONL | +| GET | `/api/use-cases` | List valid use cases | +| GET | `/api/games` | List supported games | + +### Start Recording Body + +```json +{ + "name": "My Session", + "game": "assetto_corsa", + "track": "nurburgring", + "car": "ferrari_488", + "use_case": "general", + "notes": "Testing session" +} +``` + +All fields are optional. + +### Live Telemetry Response + +```json +{ + "feeds": [ + { + "label": "KSUDP", + "port": 9996, + "health": "alive", + "frequency_hz": 20.0, + "total_packets": 1200, + "data_rate_kb_s": 15.2, + "speed_kmh": 245.0, + "gear": "5", + "engine_rpm": 8750.0, + "throttle": 0.85, + "brake": 0.0, + "lap_time": "1:23.456", + "spline_position": 0.453 + } + ], + "active_recording": "550e8400-e29b-41d4-a716-446655440000" +} +``` ## Testing @@ -156,7 +217,9 @@ cargo test ``` src/ ├── main.rs Multi-port UDP listener, TUI event loop, KSUDP handshake +├── api.rs REST API server (axum) for recording management and live telemetry ├── binary_utils.rs Shared safe binary read helpers (bounds-checked, zero fallback) +├── recording.rs Recording session manager (manifest, CRUD, auto-start) ├── telemetry_frame.rs Unified telemetry data model (TelemetryFrame) ├── feed_state.rs Per-feed health/frequency state machine with ring buffer ├── parser_ksudp.rs AC KSUDP binary parser (RTCarInfo, 6 tests) diff --git a/src/analysis.rs b/src/analysis.rs new file mode 100644 index 0000000..16755d2 --- /dev/null +++ b/src/analysis.rs @@ -0,0 +1,290 @@ +//! Telemetry analysis algorithms for sim racing data. +//! +//! Provides shift point analysis, track map reconstruction, and a dispatch +//! function that routes a [`UseCase`] to the appropriate analysis pipeline. + +use serde::Serialize; +use std::collections::HashMap; + +use crate::recording::UseCase; +use crate::telemetry_frame::{gear_display, TelemetryFrame}; + +// --------------------------------------------------------------------------- +// Shift Point Analysis +// --------------------------------------------------------------------------- + +#[derive(Debug, Clone, Serialize)] +pub struct GearRpmRange { + pub gear: i32, + pub gear_display: String, + pub min_rpm: f32, + pub max_rpm: f32, + pub avg_rpm: f32, + pub sample_count: usize, +} + +#[derive(Debug, Clone, Serialize)] +pub struct ShiftEvent { + pub from_gear: i32, + pub to_gear: i32, + pub rpm_before_shift: f32, + pub rpm_after_shift: f32, + pub rpm_drop: f32, + pub speed_kmh: f32, +} + +#[derive(Debug, Clone, Serialize)] +pub struct ShiftPointAnalysis { + pub gear_ranges: Vec, + pub shift_events: Vec, + pub recommended_shift_rpm: Option, + pub total_frames_analyzed: usize, +} + +pub fn analyze_shift_points(frames: &[TelemetryFrame]) -> ShiftPointAnalysis { + let valid: Vec<&TelemetryFrame> = frames + .iter() + .filter(|f| f.gear >= 2 && f.engine_rpm > 0.0) + .collect(); + + let mut buckets: HashMap> = HashMap::new(); + for f in &valid { + buckets.entry(f.gear).or_default().push(f.engine_rpm); + } + + let mut gear_ranges: Vec = buckets + .into_iter() + .map(|(gear, rpms)| { + let min_rpm = rpms.iter().cloned().fold(f32::INFINITY, f32::min); + let max_rpm = rpms.iter().cloned().fold(f32::NEG_INFINITY, f32::max); + let avg_rpm = rpms.iter().sum::() / rpms.len() as f32; + GearRpmRange { + gear, + gear_display: gear_display(gear), + min_rpm, + max_rpm, + avg_rpm, + sample_count: rpms.len(), + } + }) + .collect(); + gear_ranges.sort_by_key(|g| g.gear); + + let valid_indices: Vec = frames + .iter() + .enumerate() + .filter(|(_, f)| f.gear >= 2 && f.engine_rpm > 0.0) + .map(|(i, _)| i) + .collect(); + + let mut shift_events: Vec = Vec::new(); + for window in valid_indices.windows(2) { + let prev = &frames[window[0]]; + let curr = &frames[window[1]]; + if prev.gear + 1 == curr.gear { + shift_events.push(ShiftEvent { + from_gear: prev.gear, + to_gear: curr.gear, + rpm_before_shift: prev.engine_rpm, + rpm_after_shift: curr.engine_rpm, + rpm_drop: prev.engine_rpm - curr.engine_rpm, + speed_kmh: prev.speed_kmh, + }); + } + } + + let recommended_shift_rpm = { + let upshift_rpms: Vec = shift_events + .iter() + .filter(|e| e.from_gear >= 2 && e.to_gear >= 2) + .map(|e| e.rpm_before_shift) + .collect(); + if upshift_rpms.is_empty() { + None + } else { + Some(upshift_rpms.iter().sum::() / upshift_rpms.len() as f32) + } + }; + + ShiftPointAnalysis { + gear_ranges, + shift_events, + recommended_shift_rpm, + total_frames_analyzed: valid.len(), + } +} + +// --------------------------------------------------------------------------- +// Track Map Analysis +// --------------------------------------------------------------------------- + +#[derive(Debug, Clone, Serialize)] +pub struct TrackPoint { + pub x: f32, + pub y: f32, + pub z: f32, + pub spline: f32, +} + +#[derive(Debug, Clone, Serialize)] +pub struct TrackBoundary { + pub label: String, + pub point_count: usize, + pub points: Vec, +} + +#[derive(Debug, Clone, Serialize)] +pub struct TrackMapAnalysis { + pub boundaries: Vec, + pub center_line: Option, + pub total_points: usize, +} + +pub fn analyze_track_map( + runs: &[Vec], + labels: &[String], +) -> TrackMapAnalysis { + let mut boundaries: Vec = Vec::new(); + + for (run, label) in runs.iter().zip(labels.iter()) { + let mut points: Vec = run + .iter() + .filter(|f| f.spline_position != 0.0) + .map(|f| TrackPoint { + x: f.world_x, + y: f.world_y, + z: f.world_z, + spline: f.spline_position, + }) + .collect(); + points.sort_by(|a, b| a.spline.partial_cmp(&b.spline).unwrap_or(std::cmp::Ordering::Equal)); + let point_count = points.len(); + boundaries.push(TrackBoundary { + label: label.clone(), + point_count, + points, + }); + } + + let center_line = compute_center_line(&boundaries); + + let total_points: usize = boundaries.iter().map(|b| b.point_count).sum::() + + center_line.as_ref().map_or(0, |c| c.point_count); + + TrackMapAnalysis { + boundaries, + center_line, + total_points, + } +} + +fn compute_center_line(boundaries: &[TrackBoundary]) -> Option { + if boundaries.len() != 2 { + return None; + } + + let left_idx = boundaries + .iter() + .position(|b| b.label == "left")?; + let right_idx = boundaries + .iter() + .position(|b| b.label == "right")?; + + if left_idx == right_idx { + return None; + } + + let left = &boundaries[left_idx]; + let right = &boundaries[right_idx]; + + if left.points.is_empty() || right.points.is_empty() { + return None; + } + + let num_samples = left.point_count.max(right.point_count).min(2000); + let center_points: Vec = (0..num_samples) + .filter_map(|i| { + let spline = i as f32 / (num_samples - 1).max(1) as f32; + let lp = interpolate_boundary(left, spline)?; + let rp = interpolate_boundary(right, spline)?; + Some(TrackPoint { + x: (lp.x + rp.x) / 2.0, + y: (lp.y + rp.y) / 2.0, + z: (lp.z + rp.z) / 2.0, + spline, + }) + }) + .collect(); + + if center_points.is_empty() { + return None; + } + + Some(TrackBoundary { + label: "center_line".to_owned(), + point_count: center_points.len(), + points: center_points, + }) +} + +/// Linearly interpolate a TrackPoint at the given spline position from a boundary. +fn interpolate_boundary(boundary: &TrackBoundary, target_spline: f32) -> Option { + let points = &boundary.points; + if points.is_empty() { + return None; + } + + if target_spline <= points.first()?.spline { + return Some(points.first()?.clone()); + } + if target_spline >= points.last()?.spline { + return Some(points.last()?.clone()); + } + + let idx = points.partition_point(|p| p.spline < target_spline); + if idx == 0 || idx >= points.len() { + return points.first().cloned(); + } + + let a = &points[idx - 1]; + let b = &points[idx]; + let t = if (b.spline - a.spline).abs() < f32::EPSILON { + 0.0 + } else { + (target_spline - a.spline) / (b.spline - a.spline) + }; + + Some(TrackPoint { + x: a.x + t * (b.x - a.x), + y: a.y + t * (b.y - a.y), + z: a.z + t * (b.z - a.z), + spline: target_spline, + }) +} + +// --------------------------------------------------------------------------- +// Dispatch +// --------------------------------------------------------------------------- + +pub fn analyze( + use_case: &UseCase, + frames_by_run: &[Vec], + run_labels: &[String], +) -> serde_json::Value { + match use_case { + UseCase::ShiftPoints => { + if let Some(first_run) = frames_by_run.first() { + serde_json::to_value(analyze_shift_points(first_run)) + .unwrap_or_else(|_| serde_json::json!({"error": "serialization failed"})) + } else { + serde_json::json!({"error": "no frames provided"}) + } + } + UseCase::TrackMap => serde_json::to_value(analyze_track_map(frames_by_run, run_labels)) + .unwrap_or_else(|_| serde_json::json!({"error": "serialization failed"})), + _ => serde_json::json!({ + "status": "not_implemented", + "use_case": format!("{:?}", use_case), + }), + } +} diff --git a/src/api.rs b/src/api.rs new file mode 100644 index 0000000..85b5237 --- /dev/null +++ b/src/api.rs @@ -0,0 +1,526 @@ +//! REST API server for recording management and live telemetry access. +//! +//! Runs as a parallel tokio task alongside the UDP listeners. + +use axum::{ + Json, Router, + extract::{Path, State}, + http::StatusCode, + response::{IntoResponse, Response}, + routing::{get, post}, +}; +use serde::{Deserialize, Serialize}; +use std::sync::{Arc, Mutex, MutexGuard}; +use tower_http::cors::CorsLayer; + +use crate::analysis; +use crate::feed_state::FeedState; +use crate::recording::{Game, RecordingManager, RecordingStatus, StartRecordingRequest, UseCase}; +use crate::session::SessionManager; +use crate::ui::FeedDisplayInfo; + +pub struct AppState { + pub recording_manager: Arc, + pub feed_displays: Vec<(FeedDisplayInfo, Arc>)>, + pub session_manager: Arc>, +} + +#[derive(Serialize)] +struct ApiError { + error: String, +} + +fn ok_json(data: T) -> Response { + (StatusCode::OK, Json(data)).into_response() +} + +fn err(status: StatusCode, msg: &str) -> Response { + ( + status, + Json(ApiError { + error: msg.to_string(), + }), + ) + .into_response() +} + +fn lock_sessions(state: &Arc) -> MutexGuard<'_, SessionManager> { + state.session_manager.lock().unwrap_or_else(|e| { + eprintln!("[WARN] SessionManager mutex poisoned, recovering: {}", e); + e.into_inner() + }) +} + +#[derive(Serialize)] +struct LiveFeedSnapshot { + label: String, + port: u16, + health: String, + frequency_hz: f64, + total_packets: u64, + data_rate_kb_s: f64, + speed_kmh: Option, + gear: Option, + engine_rpm: Option, + throttle: Option, + brake: Option, + lap_time: Option, + spline_position: Option, +} + +#[derive(Serialize)] +struct LiveTelemetryResponse { + feeds: Vec, + active_recording: Option, +} + +#[derive(Deserialize)] +struct CreateSessionRequest { + use_case: UseCase, + label: Option, +} + +#[derive(Deserialize)] +struct StartRunRequest { + feed_name: String, + label: Option, +} + +async fn create_session( + State(state): State>, + Json(req): Json, +) -> Response { + let mut mgr = lock_sessions(&state); + match mgr.create_session(req.use_case, req.label) { + Ok(session) => (StatusCode::CREATED, Json(session)).into_response(), + Err(msg) => err(StatusCode::BAD_REQUEST, &msg), + } +} + +async fn list_sessions(State(state): State>) -> Response { + let mut mgr = lock_sessions(&state); + ok_json(mgr.list_sessions()) +} + +async fn get_session( + State(state): State>, + Path(id): Path, +) -> Response { + let mgr = lock_sessions(&state); + match mgr.get_session(&id) { + Some(session) => ok_json(session), + None => err(StatusCode::NOT_FOUND, "Session not found"), + } +} + +async fn delete_session( + State(state): State>, + Path(id): Path, +) -> Response { + let mut mgr = lock_sessions(&state); + match mgr.delete_session(&id) { + Ok(()) => StatusCode::NO_CONTENT.into_response(), + Err(msg) => err(StatusCode::NOT_FOUND, &msg), + } +} + +async fn start_run( + State(state): State>, + Path(id): Path, + Json(req): Json, +) -> Response { + let mut mgr = lock_sessions(&state); + match mgr.start_run(&id, &req.feed_name, req.label) { + Ok(run) => (StatusCode::CREATED, Json(run)).into_response(), + Err(msg) => err(StatusCode::BAD_REQUEST, &msg), + } +} + +async fn stop_run( + State(state): State>, + Path((id, run_id)): Path<(String, String)>, +) -> Response { + let mut mgr = lock_sessions(&state); + match mgr.stop_run(&id, &run_id) { + Ok(run) => ok_json(run), + Err(msg) => err(StatusCode::BAD_REQUEST, &msg), + } +} + +async fn complete_session( + State(state): State>, + Path(id): Path, +) -> Response { + let (use_case, frames_by_run, run_labels) = { + let mut mgr = lock_sessions(&state); + match mgr.complete_session(&id) { + Ok(_session) => {} + Err(msg) => return err(StatusCode::BAD_REQUEST, &msg), + } + match mgr.collect_run_frames(&id) { + Some((uc, frames, labels)) => (uc, frames, labels), + None => return err(StatusCode::NOT_FOUND, "Session not found after completion"), + } + }; + + let result = analysis::analyze(&use_case, &frames_by_run, &run_labels); + + { + let mut mgr = lock_sessions(&state); + let _ = mgr.set_analysis(&id, result); + } + + let mgr = lock_sessions(&state); + match mgr.get_session(&id) { + Some(session) => ok_json(session), + None => err(StatusCode::INTERNAL_SERVER_ERROR, "Session lost after analysis"), + } +} + +async fn get_analysis( + State(state): State>, + Path(id): Path, +) -> Response { + let mgr = lock_sessions(&state); + match mgr.get_session(&id) { + Some(session) => { + match session.analysis { + Some(ref analysis) => ok_json(analysis), + None => err(StatusCode::NOT_FOUND, "Analysis not available yet. Complete the session first."), + } + } + None => err(StatusCode::NOT_FOUND, "Session not found"), + } +} + +async fn run_analysis( + State(state): State>, + Path(id): Path, +) -> Response { + let (use_case, frames_by_run, run_labels) = { + let mgr = lock_sessions(&state); + match mgr.get_session(&id) { + Some(session) => { + if session.status != crate::session::SessionStatus::Completed { + return err(StatusCode::BAD_REQUEST, "Session must be completed before analysis"); + } + } + None => return err(StatusCode::NOT_FOUND, "Session not found"), + } + match mgr.collect_run_frames(&id) { + Some(data) => data, + None => return err(StatusCode::NOT_FOUND, "Session not found"), + } + }; + + let result = analysis::analyze(&use_case, &frames_by_run, &run_labels); + + { + let mut mgr = lock_sessions(&state); + let _ = mgr.set_analysis(&id, result.clone()); + } + + ok_json(result) +} + +async fn list_recordings(State(state): State>) -> Response { + match state.recording_manager.list_recordings().await { + Ok(recordings) => ok_json(recordings), + Err(e) => err( + StatusCode::INTERNAL_SERVER_ERROR, + &format!("Failed to list recordings: {}", e), + ), + } +} + +async fn get_recording( + State(state): State>, + Path(id): Path, +) -> Response { + match state.recording_manager.get_recording(&id).await { + Ok(manifest) => ok_json(manifest), + Err(e) if e.kind() == std::io::ErrorKind::NotFound => { + err(StatusCode::NOT_FOUND, "Recording not found") + } + Err(e) if e.kind() == std::io::ErrorKind::InvalidInput => { + err(StatusCode::BAD_REQUEST, &e.to_string()) + } + Err(e) => err( + StatusCode::INTERNAL_SERVER_ERROR, + &format!("Failed to get recording: {}", e), + ), + } +} + +async fn start_recording( + State(state): State>, + Json(req): Json, +) -> Response { + match state.recording_manager.start_recording(req).await { + Ok(manifest) => (StatusCode::CREATED, Json(manifest)).into_response(), + Err(e) => err( + StatusCode::INTERNAL_SERVER_ERROR, + &format!("Failed to start recording: {}", e), + ), + } +} + +async fn stop_recording(State(state): State>) -> Response { + match state.recording_manager.stop_recording().await { + Ok(Some(manifest)) => ok_json(manifest), + Ok(None) => err(StatusCode::BAD_REQUEST, "No active recording to stop"), + Err(e) => err( + StatusCode::INTERNAL_SERVER_ERROR, + &format!("Failed to stop recording: {}", e), + ), + } +} + +async fn stop_specific_recording( + State(state): State>, + Path(id): Path, +) -> Response { + let manifest = match state.recording_manager.get_recording(&id).await { + Ok(m) => m, + Err(e) if e.kind() == std::io::ErrorKind::NotFound => { + return err(StatusCode::NOT_FOUND, "Recording not found") + } + Err(e) if e.kind() == std::io::ErrorKind::InvalidInput => { + return err(StatusCode::BAD_REQUEST, &e.to_string()) + } + Err(e) => { + return err( + StatusCode::INTERNAL_SERVER_ERROR, + &format!("Failed to get recording: {}", e), + ) + } + }; + + if manifest.status != RecordingStatus::Recording { + return err(StatusCode::CONFLICT, "Recording is not active"); + } + + match state + .recording_manager + .stop_recording_internal(&id) + .await + { + Ok(stopped) => { + state + .recording_manager + .clear_active_if_matches(&id) + .await; + ok_json(stopped) + } + Err(e) => err( + StatusCode::INTERNAL_SERVER_ERROR, + &format!("Failed to stop recording: {}", e), + ), + } +} + +async fn delete_recording( + State(state): State>, + Path(id): Path, +) -> Response { + match state.recording_manager.delete_recording(&id).await { + Ok(()) => StatusCode::NO_CONTENT.into_response(), + Err(e) if e.kind() == std::io::ErrorKind::InvalidInput => { + err(StatusCode::CONFLICT, &e.to_string()) + } + Err(e) if e.kind() == std::io::ErrorKind::NotFound => { + err(StatusCode::NOT_FOUND, "Recording not found") + } + Err(e) => err( + StatusCode::INTERNAL_SERVER_ERROR, + &format!("Failed to delete recording: {}", e), + ), + } +} + +async fn get_feed_data( + State(state): State>, + Path((id, feed_name)): Path<(String, String)>, +) -> Response { + let active_id = state.recording_manager.active_recording_id().await; + if active_id.as_deref() == Some(id.as_str()) { + match state.recording_manager.read_frames(&feed_name) { + Some(frames) => return ok_json(frames), + None => { + return err( + StatusCode::NOT_FOUND, + &format!("Feed '{}' not found in active recording", feed_name), + ) + } + } + } + + let manifest = match state.recording_manager.get_recording(&id).await { + Ok(m) => m, + Err(e) if e.kind() == std::io::ErrorKind::NotFound => { + return err(StatusCode::NOT_FOUND, "Recording not found") + } + Err(e) if e.kind() == std::io::ErrorKind::InvalidInput => { + return err(StatusCode::BAD_REQUEST, &e.to_string()) + } + Err(e) => { + return err( + StatusCode::INTERNAL_SERVER_ERROR, + &format!("Failed to get recording: {}", e), + ) + } + }; + + let feed_stats = match manifest.feeds.iter().find(|f| f.name == feed_name) { + Some(f) => f, + None => { + return err( + StatusCode::NOT_FOUND, + &format!("Feed '{}' not found in recording", feed_name), + ) + } + }; + + let file_path = manifest.dir_path(state.recording_manager.data_dir()).join(&feed_stats.file); + if !file_path.exists() { + return err(StatusCode::NOT_FOUND, "Feed data file not found on disk (recording may still be in progress)"); + } + + match tokio::fs::read_to_string(&file_path).await { + Ok(content) => { + let frames: Vec = content + .lines() + .filter(|l| !l.is_empty()) + .filter_map(|l| serde_json::from_str(l).ok()) + .collect(); + ok_json(frames) + } + Err(e) => err( + StatusCode::INTERNAL_SERVER_ERROR, + &format!("Failed to read feed data: {}", e), + ), + } +} + +async fn get_live(State(state): State>) -> Response { + let mut feeds = Vec::new(); + + for (info, state_arc) in &state.feed_displays { + let snapshot = state_arc.lock().ok().map(|mut s| { + s.tick(); + s.clone() + }); + + let snap = match snapshot { + Some(s) => s, + None => continue, + }; + + let (speed_kmh, gear, engine_rpm, throttle, brake, lap_time, spline_position) = + if let Some(ref frame) = snap.last_frame { + ( + Some(frame.speed_kmh), + Some(frame.gear_display()), + Some(frame.engine_rpm), + Some(frame.throttle), + Some(frame.brake), + Some( + crate::telemetry_frame::TelemetryFrame::format_lap_time(frame.lap_time_ms), + ), + Some(frame.spline_position), + ) + } else { + (None, None, None, None, None, None, None) + }; + + feeds.push(LiveFeedSnapshot { + label: info.label.to_string(), + port: info.port, + health: format!("{:?}", snap.health).to_lowercase(), + frequency_hz: snap.frequency_hz, + total_packets: snap.total_packets, + data_rate_kb_s: snap.data_rate_kb_s, + speed_kmh, + gear, + engine_rpm, + throttle, + brake, + lap_time, + spline_position, + }); + } + + let active_recording = state.recording_manager.active_recording_id().await; + + ok_json(LiveTelemetryResponse { + feeds, + active_recording, + }) +} + +async fn list_use_cases() -> Response { + ok_json(UseCase::all()) +} + +async fn list_games() -> Response { + ok_json(Game::valid_choices()) +} + +pub fn build_router(state: Arc) -> Router { + Router::new() + .route("/api/sessions", get(list_sessions).post(create_session)) + .route("/api/sessions/{id}", get(get_session).delete(delete_session)) + .route("/api/sessions/{id}/runs", post(start_run)) + .route("/api/sessions/{id}/runs/{run_id}/stop", post(stop_run)) + .route("/api/sessions/{id}/complete", post(complete_session)) + .route("/api/sessions/{id}/analysis", get(get_analysis)) + .route("/api/sessions/{id}/analyze", post(run_analysis)) + .route("/api/recordings", get(list_recordings).post(start_recording)) + .route("/api/recordings/stop", post(stop_recording)) + .route( + "/api/recordings/{id}", + get(get_recording).delete(delete_recording), + ) + .route("/api/recordings/{id}/stop", post(stop_specific_recording)) + .route( + "/api/recordings/{id}/data/{feed_name}", + get(get_feed_data), + ) + .route( + "/api/recordings/{id}/frames/{feed_name}", + get(get_feed_data), + ) + .route("/api/live", get(get_live)) + .route("/api/use-cases", get(list_use_cases)) + .route("/api/games", get(list_games)) + .layer(CorsLayer::permissive()) + .with_state(state) +} + +pub async fn run_server(state: Arc, addr: String, mut shutdown_rx: tokio::sync::watch::Receiver) { + let router = build_router(state); + let listener = match tokio::net::TcpListener::bind(&addr).await { + Ok(l) => l, + Err(e) => { + eprintln!("[ERROR] Failed to bind API server on {}: {}", addr, e); + return; + } + }; + eprintln!("[INFO] REST API server listening on {}", addr); + + let shutdown_signal = async move { + while shutdown_rx.changed().await.is_ok() { + if *shutdown_rx.borrow() { + break; + } + } + }; + + if let Err(e) = axum::serve(listener, router) + .with_graceful_shutdown(shutdown_signal) + .await + { + eprintln!("[ERROR] API server error: {}", e); + } + eprintln!("[INFO] REST API server shut down."); +} diff --git a/src/main.rs b/src/main.rs index 03a7ef2..884862a 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,10 +1,14 @@ +mod api; mod binary_utils; mod feed_state; -mod ui; -mod telemetry_frame; mod parser_ksudp; -mod parser_telemetry_tool; mod parser_pcars; +mod parser_telemetry_tool; +mod recording; +mod telemetry_frame; +mod ui; +mod analysis; +mod session; use feed_state::*; use ui::*; @@ -21,10 +25,10 @@ use crossterm::{ }; use ratatui::backend::CrosstermBackend; use ratatui::Terminal; -use tokio::io::BufWriter; -use tokio::io::AsyncWriteExt; use crate::feed_state::HandshakePhase; +use crate::recording::FEED_DEFS; +use crate::session::SessionManager; use crate::telemetry_frame::TelemetryFrame; const VERSION: &str = env!("CARGO_PKG_VERSION"); @@ -35,6 +39,20 @@ 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]; +const API_PORT: u16 = 8080; + +fn api_port() -> u16 { + std::env::var("RUSTY_TELEMETRY_API_PORT") + .ok() + .and_then(|s| s.parse().ok()) + .unwrap_or(API_PORT) +} + +fn api_bind_addr() -> String { + let host = std::env::var("RUSTY_TELEMETRY_API_HOST") + .unwrap_or_else(|_| "127.0.0.1".to_string()); + format!("{}:{}", host, api_port()) +} fn decode_utf16le(bytes: &[u8]) -> String { let chars: Vec = bytes @@ -74,20 +92,20 @@ struct FeedConfig { static FEEDS: &[FeedConfig] = &[ FeedConfig { - port: 9996, - name: "ksudp", + port: FEED_DEFS[0].0, + name: FEED_DEFS[0].1, label: "KSUDP", needs_handshake: true, }, FeedConfig { - port: 10101, - name: "telemetry_tool", + port: FEED_DEFS[1].0, + name: FEED_DEFS[1].1, label: "Telemetry Tool", needs_handshake: false, }, FeedConfig { - port: 5606, - name: "pcars", + port: FEED_DEFS[2].0, + name: FEED_DEFS[2].1, label: "PCARS", needs_handshake: false, }, @@ -143,7 +161,7 @@ impl CachedStrings { async fn run_feed( feed: FeedConfig, socket: tokio::net::UdpSocket, - session_dir: String, + recording_manager: Arc, state: Arc>, mut shutdown_rx: tokio::sync::watch::Receiver, ) { @@ -264,29 +282,6 @@ async fn run_feed( None }; - let file_path = format!( - "{}/feed_{}_{}.bin", - session_dir, feed.port, feed.name - ); - let file = match tokio::fs::OpenOptions::new() - .create(true) - .append(true) - .open(&file_path) - .await - { - Ok(f) => f, - Err(e) => { - eprintln!( - "[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 string_cache = CachedStrings::new(); @@ -340,23 +335,13 @@ async fn run_feed( // 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); + if let Some(ref frame) = parsed { + s.update_frame(frame.clone()); } } - 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 Some(ref frame) = parsed { + recording_manager.push_frame(feed.name, frame.clone()); } } Err(e) => { @@ -388,20 +373,8 @@ async fn run_feed( } } - if let Err(e) = file.flush().await { - eprintln!( - "[ERROR] Flush error on :{} ({}): {}", - feed.port, feed.name, e - ); - } - if let Err(e) = file.get_ref().sync_all().await { - eprintln!( - "[ERROR] Sync error on :{} ({}): {}", - feed.port, feed.name, e - ); - } eprintln!( - "[INFO] Feed :{} ({}) flushed and closed", + "[INFO] Feed :{} ({}) stopped", feed.port, feed.name ); } @@ -411,14 +384,43 @@ async fn main() { install_panic_hook(); let listen = listen_addr(); - let session_id = chrono::Local::now().format("%Y%m%d_%H%M%S").to_string(); - let session_dir = format!("./data/{}", session_id); + let data_dir = std::path::PathBuf::from("./data"); - if let Err(e) = tokio::fs::create_dir_all(&session_dir).await { - eprintln!("[ERROR] Cannot create data dir {}: {}", session_dir, e); + if let Err(e) = tokio::fs::create_dir_all(&data_dir).await { + eprintln!("[ERROR] Cannot create data dir: {}", e); std::process::exit(1); } + let recording_manager = Arc::new(recording::RecordingManager::new(&data_dir)); + let session_manager = Arc::new(std::sync::Mutex::new( + SessionManager::new(recording_manager.clone()) + )); + + let auto_start = recording::StartRecordingRequest { + name: Some( + chrono::Local::now() + .format("Session %Y-%m-%d %H:%M") + .to_string(), + ), + game: Some(recording::Game::Unknown), + track: None, + car: None, + use_case: Some(recording::UseCase::General), + notes: None, + }; + match recording_manager.start_recording(auto_start).await { + Ok(m) => { + eprintln!( + "[INFO] Auto-started recording: {} ({})", + m.name, m.id + ); + } + Err(e) => { + eprintln!("[ERROR] Failed to start recording: {}", e); + std::process::exit(1); + } + } + let mut terminal = match setup_terminal() { Ok(t) => t, Err(e) => { @@ -483,14 +485,23 @@ async fn main() { if let Some(socket) = socket_opt { let feed_clone = feed.clone(); - let dir = session_dir.clone(); + let rm = recording_manager.clone(); let sh = shutdown_rx.clone(); feed_tasks.push(tokio::spawn(async move { - run_feed(feed_clone, socket, dir, state, sh).await; + run_feed(feed_clone, socket, rm, state, sh).await; })); } } + let api_state = Arc::new(api::AppState { + recording_manager: recording_manager.clone(), + feed_displays: feed_displays.clone(), + session_manager: session_manager.clone(), + }); + let api_addr = api_bind_addr(); + eprintln!("[INFO] REST API server will listen on {}", api_addr); + tokio::spawn(api::run_server(api_state, api_addr, shutdown_rx.clone())); + 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(); @@ -508,6 +519,12 @@ async fn main() { let mut render_interval = tokio::time::interval(Duration::from_millis(RENDER_INTERVAL_MS)); let running = Arc::new(AtomicBool::new(true)); + let recording_rx_display = recording_manager.subscribe(); + let get_session_display = || { + recording_rx_display.borrow().clone() + .map(|id| format!("./data/{}/", id)) + .unwrap_or_else(|| "No active recording".to_string()) + }; while running.load(Ordering::Relaxed) { tokio::select! { @@ -528,8 +545,9 @@ async fn main() { } } Event::Resize(_, _) => { + let session_display = get_session_display(); let _ = terminal.draw(|f| { - render(f, &feed_displays, &session_dir, VERSION); + render(f, &feed_displays, &session_display, VERSION); }); } _ => {} @@ -537,8 +555,9 @@ async fn main() { } } _ = render_interval.tick() => { + let session_display = get_session_display(); let _ = terminal.draw(|f| { - render(f, &feed_displays, &session_dir, VERSION); + render(f, &feed_displays, &session_display, VERSION); }); } } @@ -547,6 +566,11 @@ async fn main() { event_running.store(false, Ordering::Relaxed); let _ = shutdown_tx.send(true); + // Flush in-memory ring buffers to disk before exit. + if let Err(e) = recording_manager.stop_recording().await { + eprintln!("[WARN] Failed to flush recording on shutdown: {}", e); + } + match tokio::time::timeout( Duration::from_secs(3), futures::future::join_all(feed_tasks), @@ -561,7 +585,7 @@ async fn main() { } } Err(_) => { - eprintln!("[WARN] Timeout waiting for feeds to flush; forcing exit"); + eprintln!("[WARN] Timeout waiting for feeds to stop; forcing exit"); } } diff --git a/src/recording.rs b/src/recording.rs new file mode 100644 index 0000000..e21d2fd --- /dev/null +++ b/src/recording.rs @@ -0,0 +1,508 @@ +//! Recording session management with always-on ring buffers. +//! +//! Each feed has a rolling ring buffer holding the last 5 minutes of telemetry frames. +//! Explicit recordings can flush the ring buffer to disk as JSONL files. +//! Use-case sessions extract frames from the ring buffer for analysis. + +use serde::{Deserialize, Serialize}; +use std::collections::{HashMap, VecDeque}; +use std::path::{Path, PathBuf}; +use std::sync::Arc; +use tokio::fs; +use tokio::sync::{Mutex, watch}; +use crate::telemetry_frame::TelemetryFrame; + +/// Feed port/name definitions. Must be kept in sync with FEEDS in main.rs. +pub const FEED_DEFS: &[(u16, &str)] = &[ + (9996, "ksudp"), + (10101, "telemetry_tool"), + (5606, "pcars"), +]; + +/// Validate that a recording ID is a well-formed UUID. +/// Prevents path traversal attacks via the API. +fn validate_recording_id(id: &str) -> std::io::Result<()> { + if uuid::Uuid::parse_str(id).is_err() { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidInput, + "Invalid recording ID format", + )); + } + Ok(()) +} + +/// Generate the JSONL filename for a feed. +pub fn feed_filename(port: u16, name: &str) -> String { + format!("feed_{}_{}.jsonl", port, name) +} + +/// Ring buffer duration in seconds. +const RING_BUFFER_SECS: u64 = 300; // 5 minutes + +/// Maximum sample rate across all feeds (Hz). +const MAX_FEED_RATE_HZ: u64 = 60; + +/// Ring buffer capacity = duration × max rate. +pub const RING_BUFFER_CAPACITY: usize = (RING_BUFFER_SECS * MAX_FEED_RATE_HZ) as usize; + +/// In-memory ring buffer of parsed telemetry frames for a single feed. +/// Always active — captures the last 5 minutes of data regardless of recording state. +pub struct RingBuffer { + frames: VecDeque, + total_pushed: u64, + dropped_count: u64, +} + +impl RingBuffer { + pub fn new() -> Self { + Self { + frames: VecDeque::with_capacity(RING_BUFFER_CAPACITY), + total_pushed: 0, + dropped_count: 0, + } + } + + /// Push a frame. Oldest frame is evicted when at capacity (true ring behavior). + pub fn push(&mut self, frame: TelemetryFrame) { + if self.frames.len() >= RING_BUFFER_CAPACITY { + self.frames.pop_front(); + self.dropped_count += 1; + } + self.frames.push_back(frame); + self.total_pushed += 1; + } + + /// Returns all frames in chronological order. + pub fn frames(&self) -> &VecDeque { + &self.frames + } + + /// Number of frames currently stored. + pub fn len(&self) -> usize { + self.frames.len() + } + + /// Whether the buffer is empty. + pub fn is_empty(&self) -> bool { + self.frames.is_empty() + } + + /// Total frames ever pushed (monotonically increasing). + pub fn total_pushed(&self) -> u64 { + self.total_pushed + } + + /// Number of frames evicted due to ring overflow. + pub fn dropped_count(&self) -> u64 { + self.dropped_count + } + + /// Returns a marker representing the current position (for extract_since). + pub fn marker(&self) -> u64 { + self.total_pushed + } + + /// Extract frames pushed since the given marker. + /// Returns all frames if the marker predates the buffer contents. + /// Returns empty vec if marker is at or past the current position. + pub fn extract_since(&self, marker: u64) -> Vec { + if marker >= self.total_pushed { + return Vec::new(); + } + let buffer_start = self.total_pushed.saturating_sub(self.frames.len() as u64); + let start_offset = if marker < buffer_start { + 0 + } else { + (marker - buffer_start) as usize + }; + self.frames.iter().skip(start_offset).cloned().collect() + } + + /// Snapshot of all current frames. + pub fn snapshot(&self) -> Vec { + self.frames.iter().cloned().collect() + } +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +#[serde(rename_all = "snake_case")] +pub enum UseCase { + ShiftPoints, + RacingLine, + BrakingAnalysis, + LapTimes, + SectorAnalysis, + TrackMap, + General, +} + +impl UseCase { + pub fn all() -> Vec { + vec![ + UseCase::ShiftPoints, + UseCase::RacingLine, + UseCase::BrakingAnalysis, + UseCase::LapTimes, + UseCase::SectorAnalysis, + UseCase::TrackMap, + UseCase::General, + ] + } +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +#[serde(rename_all = "snake_case")] +pub enum Game { + AssettoCorsa, + AssettoCorsaCompetizione, + ProjectCars1, + ProjectCars2, + Unknown, +} + +impl Game { + /// Returns the list of valid game choices for API consumers (excludes Unknown). + pub fn valid_choices() -> Vec { + vec![ + Game::AssettoCorsa, + Game::AssettoCorsaCompetizione, + Game::ProjectCars1, + Game::ProjectCars2, + ] + } +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct FeedStats { + pub port: u16, + pub name: String, + pub bytes: u64, + pub file: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +#[serde(rename_all = "snake_case")] +pub enum RecordingStatus { + Recording, + Stopped, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct RecordingManifest { + pub id: String, + pub name: String, + pub game: Game, + pub track: String, + pub car: String, + pub use_case: UseCase, + pub status: RecordingStatus, + pub created_at: String, + pub stopped_at: Option, + pub feeds: Vec, + pub notes: String, +} + +impl RecordingManifest { + pub async fn save(&self, data_dir: &Path) -> std::io::Result<()> { + let dir = data_dir.join(&self.id); + fs::create_dir_all(&dir).await?; + let json = serde_json::to_string_pretty(self)?; + fs::write(dir.join("manifest.json"), json).await?; + Ok(()) + } + + pub async fn load(dir: &Path) -> std::io::Result { + let json = fs::read_to_string(dir.join("manifest.json")).await?; + let manifest: RecordingManifest = serde_json::from_str(&json)?; + Ok(manifest) + } + + pub fn dir_path(&self, data_dir: &Path) -> PathBuf { + data_dir.join(&self.id) + } +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct StartRecordingRequest { + pub name: Option, + pub game: Option, + pub track: Option, + pub car: Option, + pub use_case: Option, + pub notes: Option, +} + +/// Manages always-on ring buffers and optional disk-based recordings. +pub struct RecordingManager { + data_dir: PathBuf, + active: Mutex>, + recording_change_tx: watch::Sender>, + /// Per-feed ring buffers — always active, created at startup. + ring_buffers: HashMap>>, +} + +impl RecordingManager { + pub fn new(data_dir: &Path) -> Self { + let (tx, _) = watch::channel(None); + let mut ring_buffers = HashMap::new(); + for &(_port, name) in FEED_DEFS { + ring_buffers.insert(name.to_string(), Arc::new(std::sync::Mutex::new(RingBuffer::new()))); + } + Self { + data_dir: data_dir.to_path_buf(), + active: Mutex::new(None), + recording_change_tx: tx, + ring_buffers, + } + } + + pub fn data_dir(&self) -> &Path { + &self.data_dir + } + + pub fn subscribe(&self) -> watch::Receiver> { + self.recording_change_tx.subscribe() + } + + /// Push a telemetry frame into the feed's ring buffer. Always succeeds. + pub fn push_frame(&self, feed_name: &str, frame: TelemetryFrame) { + if let Some(buf) = self.ring_buffers.get(feed_name) { + if let Ok(mut b) = buf.lock() { + b.push(frame); + } + } + } + + /// Get the current ring buffer marker for a feed (for extract_since). + pub fn ring_marker(&self, feed_name: &str) -> u64 { + self.ring_buffers.get(feed_name) + .and_then(|b| b.lock().ok()) + .map(|b| b.marker()) + .unwrap_or(0) + } + + /// Extract frames from a feed's ring buffer since the given marker. + pub fn extract_since(&self, feed_name: &str, marker: u64) -> Vec { + self.ring_buffers.get(feed_name) + .and_then(|b| b.lock().ok()) + .map(|b| b.extract_since(marker)) + .unwrap_or_default() + } + + /// Read all current frames from a feed's ring buffer. + pub fn read_frames(&self, feed_name: &str) -> Option> { + self.ring_buffers.get(feed_name) + .and_then(|b| b.lock().ok()) + .map(|b| b.snapshot()) + } + + /// Get ring buffer stats for a feed. + pub fn ring_stats(&self, feed_name: &str) -> Option<(usize, u64, u64)> { + self.ring_buffers.get(feed_name) + .and_then(|b| b.lock().ok()) + .map(|b| (b.len(), b.total_pushed(), b.dropped_count())) + } + + pub async fn start_recording( + &self, + req: StartRecordingRequest, + ) -> std::io::Result { + // NOTE: Hold the active lock across I/O to prevent concurrent start_recording calls + // from creating orphaned recordings. For a single-user local tool, the contention + // cost is negligible compared to the correctness benefit. + let mut active = self.active.lock().await; + if let Some(ref id) = *active { + self.stop_recording_internal(id).await?; + } + + let id = uuid::Uuid::new_v4().to_string(); + let now = chrono::Utc::now().to_rfc3339(); + + let feed_name = |port: u16, name: &str| -> FeedStats { + FeedStats { + port, + name: name.to_string(), + bytes: 0, + file: feed_filename(port, name), + } + }; + + let manifest = RecordingManifest { + id: id.clone(), + name: req.name.unwrap_or_else(|| { + chrono::Local::now() + .format("Recording %Y-%m-%d %H:%M") + .to_string() + }), + game: req.game.unwrap_or(Game::Unknown), + track: req.track.unwrap_or_default(), + car: req.car.unwrap_or_default(), + use_case: req.use_case.unwrap_or(UseCase::General), + status: RecordingStatus::Recording, + created_at: now, + stopped_at: None, + feeds: FEED_DEFS.iter().map(|&(port, name)| feed_name(port, name)).collect(), + notes: req.notes.unwrap_or_default(), + }; + + let dir = manifest.dir_path(&self.data_dir); + fs::create_dir_all(&dir).await?; + manifest.save(&self.data_dir).await?; + + *active = Some(id.clone()); + let _ = self.recording_change_tx.send(Some(id)); + Ok(manifest) + } + + pub async fn stop_recording(&self) -> std::io::Result> { + let mut active = self.active.lock().await; + let id = match active.clone() { + Some(id) => id, + None => return Ok(None), + }; + let result = self.stop_recording_internal(&id).await; + if result.is_ok() { + active.take(); + let _ = self.recording_change_tx.send(None); + } + result.map(Some) + } + + pub async fn stop_recording_internal(&self, id: &str) -> std::io::Result { + validate_recording_id(id)?; + let dir = self.data_dir.join(id); + if !dir.exists() { + return Err(std::io::Error::new( + std::io::ErrorKind::NotFound, + "Recording not found", + )); + } + let mut manifest = RecordingManifest::load(&dir).await?; + + // Flush ring buffers to disk as JSONL files. + // Step 1: Clone frame data while locked (fast memcpy). + let frame_snapshots: Vec<(PathBuf, Vec)> = { + let mut snapshots = Vec::new(); + for &(port, name) in FEED_DEFS { + if let Some(buf_arc) = self.ring_buffers.get(name) { + if let Ok(buf) = buf_arc.lock() { + if !buf.is_empty() { + let file_path = dir.join(feed_filename(port, name)); + let frames: Vec = buf.frames().iter().cloned().collect(); + snapshots.push((file_path, frames)); + } + } + } + } + snapshots + }; + // Step 2: Serialize to JSONL outside the lock. + let mut flush_errors = Vec::new(); + for (path, frames) in &frame_snapshots { + let mut content = String::with_capacity(frames.len() * 400); + for frame in frames { + if let Ok(line) = serde_json::to_string(frame) { + content.push_str(&line); + content.push('\n'); + } + } + if let Err(e) = fs::write(path, content).await { + eprintln!("[WARN] Failed to flush frames to {}: {}", path.display(), e); + flush_errors.push(format!("{}: {}", path.display(), e)); + } + } + + if !flush_errors.is_empty() { + return Err(std::io::Error::new( + std::io::ErrorKind::Other, + format!("Flush errors: {}", flush_errors.join(", ")), + )); + } + + // Update feed stats from actual file sizes AFTER flush. + for feed in &mut manifest.feeds { + let feed_path = dir.join(&feed.file); + if feed_path.exists() { + if let Ok(metadata) = fs::metadata(&feed_path).await { + feed.bytes = metadata.len(); + } + } + } + + manifest.status = RecordingStatus::Stopped; + manifest.stopped_at = Some(chrono::Utc::now().to_rfc3339()); + manifest.save(&self.data_dir).await?; + Ok(manifest) + } + + pub async fn list_recordings(&self) -> std::io::Result> { + let mut recordings = Vec::new(); + let mut entries = fs::read_dir(&self.data_dir).await?; + + while let Some(entry) = entries.next_entry().await? { + let path = entry.path(); + if path.is_dir() && path.join("manifest.json").exists() { + match RecordingManifest::load(&path).await { + Ok(m) => recordings.push(m), + Err(e) => { + eprintln!( + "[WARN] Failed to load manifest from {}: {}", + path.display(), + e + ); + } + } + } + } + + recordings.sort_by(|a, b| b.created_at.cmp(&a.created_at)); + Ok(recordings) + } + + pub async fn get_recording(&self, id: &str) -> std::io::Result { + validate_recording_id(id)?; + let dir = self.data_dir.join(id); + if !dir.exists() { + return Err(std::io::Error::new( + std::io::ErrorKind::NotFound, + "Recording not found", + )); + } + RecordingManifest::load(&dir).await + } + + pub async fn delete_recording(&self, id: &str) -> std::io::Result<()> { + validate_recording_id(id)?; + { + let active = self.active.lock().await; + if active.as_deref() == Some(id) { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidInput, + "Cannot delete an active recording. Stop it first.", + )); + } + } + let dir = self.data_dir.join(id); + if !dir.exists() { + return Err(std::io::Error::new( + std::io::ErrorKind::NotFound, + "Recording not found", + )); + } + fs::remove_dir_all(&dir).await + } + + pub async fn active_recording_id(&self) -> Option { + self.active.lock().await.clone() + } + + pub async fn clear_active_if_matches(&self, id: &str) { + let mut active = self.active.lock().await; + if let Some(ref active_id) = *active { + if active_id == id { + *active = None; + let _ = self.recording_change_tx.send(None); + } + } + } +} diff --git a/src/session.rs b/src/session.rs new file mode 100644 index 0000000..ccff077 --- /dev/null +++ b/src/session.rs @@ -0,0 +1,273 @@ +//! Use-case driven session management for telemetry analysis. +//! +//! Users create sessions tied to specific analysis goals (shift point analysis, +//! track mapping, braking analysis, etc.). Each session contains multiple "runs" +//! that capture telemetry frames from the always-on ring buffer between start/stop +//! markers. Frames are stored in-memory and can be retrieved for analysis. + +use serde::{Deserialize, Serialize}; +use std::collections::HashMap; +use std::sync::Arc; + +use crate::recording::{RecordingManager, UseCase}; +use crate::telemetry_frame::TelemetryFrame; + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +#[serde(rename_all = "snake_case")] +pub enum SessionStatus { + Active, + Completed, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +#[serde(rename_all = "snake_case")] +pub enum RunStatus { + Recording, + Stopped, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct SessionRun { + pub id: String, + pub label: String, + pub feed_name: String, + pub status: RunStatus, + pub start_marker: u64, + pub end_marker: Option, + pub frame_count: usize, + pub data_truncated: bool, + #[serde(skip)] + pub frames: Vec, +} + +impl SessionRun { + fn summary_clone(&self) -> Self { + Self { + id: self.id.clone(), + label: self.label.clone(), + feed_name: self.feed_name.clone(), + status: self.status.clone(), + start_marker: self.start_marker, + end_marker: self.end_marker, + frame_count: self.frame_count, + data_truncated: self.data_truncated, + frames: Vec::new(), + } + } +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct UseCaseSession { + pub id: String, + pub use_case: UseCase, + pub label: String, + pub status: SessionStatus, + pub created_at: String, + pub completed_at: Option, + pub runs: Vec, + pub analysis: Option, +} + +impl UseCaseSession { + fn summary_clone(&self) -> Self { + Self { + id: self.id.clone(), + use_case: self.use_case.clone(), + label: self.label.clone(), + status: self.status.clone(), + created_at: self.created_at.clone(), + completed_at: self.completed_at.clone(), + runs: self.runs.iter().map(|r| r.summary_clone()).collect(), + analysis: self.analysis.clone(), + } + } +} + +pub struct SessionManager { + sessions: HashMap, + recording_manager: Arc, +} + +impl SessionManager { + pub fn new(recording_manager: Arc) -> Self { + Self { + sessions: HashMap::new(), + recording_manager, + } + } + + pub fn create_session( + &mut self, + use_case: UseCase, + label: Option, + ) -> Result { + let id = uuid::Uuid::new_v4().to_string(); + let now = chrono::Utc::now().to_rfc3339(); + let session = UseCaseSession { + id: id.clone(), + label: label.unwrap_or_else(|| format!("{:?} Session", use_case)), + use_case, + status: SessionStatus::Active, + created_at: now, + completed_at: None, + runs: Vec::new(), + analysis: None, + }; + self.sessions.insert(id, session.clone()); + Ok(session) + } + + pub fn list_sessions(&mut self) -> Vec { + let mut sessions: Vec = self + .sessions + .values() + .map(|s| s.summary_clone()) + .collect(); + sessions.sort_by(|a, b| b.created_at.cmp(&a.created_at)); + sessions + } + + pub fn get_session(&self, id: &str) -> Option { + self.sessions.get(id).map(|s| s.summary_clone()) + } + + pub fn start_run( + &mut self, + session_id: &str, + feed_name: &str, + label: Option, + ) -> Result { + let session = self + .sessions + .get_mut(session_id) + .ok_or_else(|| format!("Session '{}' not found", session_id))?; + + if session.status != SessionStatus::Active { + return Err(format!("Session '{}' is not active", session_id)); + } + + let marker = self.recording_manager.ring_marker(feed_name); + let run_id = uuid::Uuid::new_v4().to_string(); + let run = SessionRun { + id: run_id, + label: label.unwrap_or_else(|| format!("Run {}", session.runs.len() + 1)), + feed_name: feed_name.to_string(), + status: RunStatus::Recording, + start_marker: marker, + end_marker: None, + frame_count: 0, + data_truncated: false, + frames: Vec::new(), + }; + let summary = run.summary_clone(); + session.runs.push(run); + Ok(summary) + } + + pub fn stop_run( + &mut self, + session_id: &str, + run_id: &str, + ) -> Result { + let session = self + .sessions + .get_mut(session_id) + .ok_or_else(|| format!("Session '{}' not found", session_id))?; + + let run = session + .runs + .iter_mut() + .find(|r| r.id == run_id) + .ok_or_else(|| format!("Run '{}' not found in session '{}'", run_id, session_id))?; + + if run.status != RunStatus::Recording { + return Err(format!("Run '{}' is not recording", run_id)); + } + + let end_marker = self.recording_manager.ring_marker(&run.feed_name); + let frames = self + .recording_manager + .extract_since(&run.feed_name, run.start_marker); + + let expected = end_marker.saturating_sub(run.start_marker); + let data_truncated = (frames.len() as u64) < expected && expected > 0; + + run.end_marker = Some(end_marker); + run.frame_count = frames.len(); + run.frames = frames; + run.status = RunStatus::Stopped; + run.data_truncated = data_truncated; + + Ok(run.summary_clone()) + } + + pub fn complete_session(&mut self, id: &str) -> Result { + let session = self + .sessions + .get_mut(id) + .ok_or_else(|| format!("Session '{}' not found", id))?; + + if session.status != SessionStatus::Active { + return Err(format!("Session '{}' is not active", id)); + } + + for run in &mut session.runs { + if run.status == RunStatus::Recording { + let end_marker = self.recording_manager.ring_marker(&run.feed_name); + let frames = self + .recording_manager + .extract_since(&run.feed_name, run.start_marker); + + let expected = end_marker.saturating_sub(run.start_marker); + let data_truncated = (frames.len() as u64) < expected && expected > 0; + + run.end_marker = Some(end_marker); + run.frame_count = frames.len(); + run.frames = frames; + run.status = RunStatus::Stopped; + run.data_truncated = data_truncated; + } + } + + session.status = SessionStatus::Completed; + session.completed_at = Some(chrono::Utc::now().to_rfc3339()); + + Ok(session.summary_clone()) + } + + pub fn set_analysis( + &mut self, + session_id: &str, + result: serde_json::Value, + ) -> Result<(), String> { + let session = self + .sessions + .get_mut(session_id) + .ok_or_else(|| format!("Session '{}' not found", session_id))?; + session.analysis = Some(result); + Ok(()) + } + + /// Collect frames and labels from all runs for analysis. + /// + /// Clones frames to allow the caller to release the SessionManager lock + /// before running CPU-bound analysis. Frames must remain in runs for + /// potential re-analysis via `POST /api/sessions/{id}/analyze`. + pub fn collect_run_frames(&self, session_id: &str) -> Option<(UseCase, Vec>, Vec)> { + let session = self.sessions.get(session_id)?; + let mut frames_by_run = Vec::new(); + let mut run_labels = Vec::new(); + for run in &session.runs { + frames_by_run.push(run.frames.clone()); + run_labels.push(run.label.clone()); + } + Some((session.use_case.clone(), frames_by_run, run_labels)) + } + + pub fn delete_session(&mut self, id: &str) -> Result<(), String> { + self.sessions + .remove(id) + .ok_or_else(|| format!("Session '{}' not found", id))?; + Ok(()) + } +} diff --git a/src/telemetry_frame.rs b/src/telemetry_frame.rs index 9de4c08..49d9b4e 100644 --- a/src/telemetry_frame.rs +++ b/src/telemetry_frame.rs @@ -3,8 +3,10 @@ //! Both the KSUDP (port 9996) and Telemetry Tool (port 10101) parsers produce //! `TelemetryFrame` instances from their raw binary data. +use serde::Serialize; + /// Identifies which feed source produced this frame. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] pub enum FeedSource { Ksudp, TelemetryTool, @@ -18,7 +20,7 @@ pub const NUM_WHEELS: usize = 4; /// /// Fields present in both feeds are stored directly; fields unique to one feed /// are wrapped in `Option`. -#[derive(Debug, Clone)] +#[derive(Debug, Clone, Serialize)] pub struct TelemetryFrame { // ── Speed ────────────────────────────────────────────────────── pub speed_kmh: f32, @@ -233,10 +235,15 @@ impl TelemetryFrame { /// 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), - } + gear_display(self.gear) + } +} + +/// AC gear encoding: 0=Reverse, 1=Neutral, 2+=Gear N-1. +pub fn gear_display(gear: i32) -> String { + match gear { + 0 => "R".to_string(), + 1 => "N".to_string(), + n => format!("{}", n - 1), } } diff --git a/src/ui.rs b/src/ui.rs index 8893477..3492853 100644 --- a/src/ui.rs +++ b/src/ui.rs @@ -208,7 +208,7 @@ pub fn render( let footer_text = Line::from(vec![ Span::styled( - format!(" Session: {}/ ", session_dir), + format!(" Recording: {}/ ", session_dir), Style::default().fg(Color::White), ), Span::raw(" "),