diff --git a/GLOSSARY.md b/GLOSSARY.md index 88316fa..a1dd2ac 100644 --- a/GLOSSARY.md +++ b/GLOSSARY.md @@ -34,6 +34,10 @@ _Avoid_: IPC layer, transport A stand-in for iMotions used when no real iMotions instance is available. _Avoid_: Fake iMotions +**Mirror**: +A second destination the **Relay** copies every packet to, so the Python side can read a flight while iMotions records it. The copy is best effort: packets are dropped if the mirror falls behind, and the iMotions stream is never held up for it (`docs/adr/0003`). +_Avoid_: Python stream, fan-out, secondary connection, sink + ### Performance measurement **Airspeed**: diff --git a/README.md b/README.md index 6dc1b75..af4cf17 100644 --- a/README.md +++ b/README.md @@ -119,6 +119,24 @@ Two more behaviors that catch people out: - The eight dataref toggles do not affect what is sent over TCP. All eight values are always transmitted. The toggles control only which `` entries appear in the generated `iMotions.xml`. - Losing X-Plane also tears down the TCP connection. After restarting the simulator, press `Connect TCP` again. +### Sending to iMotions and Python at the same time + +`Connect TCP` has one destination. To feed iMotions and the Python side from the same flight, use the mirror: + +1. Connect TCP to iMotions as above. +2. Start the logger, which listens on port 5001 by default: `uv run python -m pilot_inference.capture.data_logger` from `inference/`. +3. Type its address, for example `127.0.0.1:5001`, into the second box and press `Connect mirror`. + +Both then receive the same packets, byte for byte. The mirror's socket is written from its own thread, so the iMotions stream never waits on it: if the Python process is missing, slow or killed, copies are dropped or the mirror is closed, a line appears in the event log, and iMotions keeps recording at full rate. Reconnect it with `Connect mirror` once Python is listening again. + +What the mirror does not promise is every packet. A reader that falls behind loses copies, which suits statistics over a stream and does not suit anything needing a complete record. Whatever reads this socket should keep reading it, on its own thread if it also does slow work. + +Reconnecting opens a new connection, and the logger starts a new `flight_id` for each one unless you passed `--flight-id`. Pass it for a real session, or one flight will be split across two ids in the CSV. + +With no iMotions to record to, skip the mirror and point `Connect TCP` straight at the logger instead. The mirror lives inside the iMotions connection, so it needs that connection first. + +`Send Packet` goes to both destinations, which makes it a check of both links. It also means the logger writes a row of zeros, so avoid it mid-recording. + ## 3. iMotions With the relay running, generate the event-source definition that iMotions needs: diff --git a/docs/adr/0003-the-relay-mirrors-its-stream-rather-than-holding-two.md b/docs/adr/0003-the-relay-mirrors-its-stream-rather-than-holding-two.md new file mode 100644 index 0000000..f09f906 --- /dev/null +++ b/docs/adr/0003-the-relay-mirrors-its-stream-rather-than-holding-two.md @@ -0,0 +1,24 @@ +# The relay mirrors its iMotions stream rather than holding two equal connections + +A flight has to reach iMotions and the Python side at the same time (#245). The relay keeps one primary TCP connection, the one `Connect TCP` opens, and an optional **Mirror** that receives a copy of every packet. Both are written from one packet list, so they cannot drift apart. The mirror's socket is owned by its own writer thread, and the TCP thread reaches it only through a bounded queue. + +## Considered Options + +- **Mirror inside the existing TCP thread.** Simplest, and how this was first written. Rejected after review: the thread wrote each packet to iMotions and then to the mirror, so a mirror write that blocked delayed the next iMotions write, and a connect attempt stalled the stream for the length of its timeout. The recording is the thing that cannot be redone, so it must not wait on a copy. +- **Mirror on its own writer thread, fed by a bounded queue (chosen).** The TCP thread's only mirror cost is pushing into a queue that either accepts the packet or refuses it. Connecting and writing both happen off that thread. +- **Two equal connections, either optional.** Cleaner in principle and would let the Python side run with no iMotions at all. Rejected because it doubles the connection state machine (`tcp_connect`, `tcp_disconnect`, `reap_tcp_thread`, the status line and the saved-address list all assume one stream), and the case it buys is already covered: with no iMotions, point `Connect TCP` straight at the Python listener. +- **Let the Python side read the CSV the logger already writes.** Rejected. Live grouping needs samples as they arrive, and the logger only has them because it is the thing the relay is connected to. + +## Consequences + +What is guaranteed: the TCP thread never waits on the mirror. Connecting, writing and failing all happen on the writer thread, and the queue is pushed with a non-blocking try-send. What is not guaranteed is delivery of every copy. When the queue is full the packet is dropped, which is the right trade for a consumer doing statistics over a stream, and the wrong one for anything that needs a complete record. + +A mirror falling behind is reported once when it starts dropping and once when it catches up, not per packet, because the event log is capped at 200 entries and 160 packets a second would empty it. + +The Python side has to read its socket continuously, on a thread of its own if it also does slow work. A clustering refit that stops reading for long enough will first lose copies and then, once the write timeout expires, lose the mirror, which does not reconnect by itself. + +Nothing reconnects the mirror automatically. If Python restarts, press `Connect mirror` again. + +`Send Packet` reaches both destinations, so it checks both links at once, and it writes a row of zeros to the logger as well as to iMotions. + +If the Python side later needs to run without iMotions in the same session, revisit this and promote the mirror to an equal connection. diff --git a/inference/README.md b/inference/README.md index 49bfb0e..f37f8a4 100644 --- a/inference/README.md +++ b/inference/README.md @@ -80,6 +80,11 @@ Without it the rows are written as "unknown". `--flight-id` is optional, but pas session: without it each relay connection gets its own generated id, so a reconnect mid-flight splits one flight into two. The logger writes `data/session_flight_data.csv`. +Point the relay at this listener with `Connect TCP`, or, when iMotions is recording the same +flight, with `Connect mirror` so both receive the stream (see the root README). A mirror reader +must keep reading: if it stops while it does slow work it loses copies, and then the mirror +itself, which does not reconnect on its own. + ## Performance groups | Step | Module | Purpose | Input | Output | diff --git a/relay/src/message.rs b/relay/src/message.rs index 28882aa..372f17e 100644 --- a/relay/src/message.rs +++ b/relay/src/message.rs @@ -35,6 +35,11 @@ pub(crate) enum Message { TcpAddrFieldUpdate(String), SavedTcpAddrSelected(String), + // Second destination for the same packets, read by the Python side + MirrorAddrFieldUpdate(String), + ConnectMirror, + DisconnectMirror, + SendPacket, } @@ -55,6 +60,10 @@ pub(crate) enum ToTcpThreadMessage { // a fully built packet to write as-is, used by the Send Packet button SendRaw(String), + + /// Point the mirror at an address, or `None` to drop it. The mirror gets a + /// copy of every packet iMotions receives. + SetMirror(Option), } pub(crate) enum FromTcpThreadMessage { @@ -64,6 +73,13 @@ pub(crate) enum FromTcpThreadMessage { Disconnected(String), /// A packet was sent to remote (In bytes) Sent(usize), - /// Error reported + /// Error reported SendError(String), + /// Mirror destination accepted the connection + MirrorConnected(String), + /// Mirror was dropped, by the user or by a failed connect or write + MirrorDisconnected(String), + /// Mirror is not keeping up, so copies are being dropped. The iMotions + /// stream is unaffected. + MirrorLagging(String), } diff --git a/relay/src/state.rs b/relay/src/state.rs index 23b261f..2b3f6e4 100644 --- a/relay/src/state.rs +++ b/relay/src/state.rs @@ -12,6 +12,7 @@ use std::path::PathBuf; use std::thread::{JoinHandle, spawn}; use iced::time::Duration; use crate::bichannel; +use crate::bichannel::ChildBiChannel; use crate::bichannel::ParentBiChannel; use crate::message::FromIpcThreadMessage; use crate::message::FromTcpThreadMessage; @@ -41,6 +42,27 @@ pub(crate) const EVENT_LOG_CAPACITY: usize = 200; /// The placeholder text and the validation message both come from here. pub(crate) const IMOTIONS_ADDR_HINT: &str = ":5050"; +/// What to type into the mirror's address box. Unlike the iMotions hint this is +/// a working example, because the Python listener usually runs on the same +/// machine as the relay, and 5001 is the capture logger's default port. +pub(crate) const MIRROR_ADDR_HINT: &str = "127.0.0.1:5001"; + +/// Bounds on the mirror destination that feeds the Python side (#245). +/// +/// Connecting and writing both run on the mirror's own thread, so these never +/// cost the iMotions stream anything. They exist so a Python process that is +/// gone or wedged is given up on rather than held open forever. +const MIRROR_CONNECT_TIMEOUT: StdDuration = StdDuration::from_millis(500); +const MIRROR_WRITE_TIMEOUT: StdDuration = StdDuration::from_millis(100); + +/// How many packets may wait for the mirror's writer thread. +/// +/// The stream is 8 packets per sample at 20 Hz, so 160 a second. This holds +/// about a second and a half, enough to ride out a clustering refit without +/// losing anything. Past that the oldest behaviour is to drop new packets +/// rather than make iMotions wait. +const MIRROR_QUEUE_CAPACITY: usize = 256; + // --- State definition ------------------------------------------------------- #[allow(unused)] pub(crate) struct State { @@ -84,6 +106,10 @@ pub(crate) struct State { pub saved_tcp_addrs: Vec, pub selected_tcp_addr: Option, pub tcp_addr_validation_error: Option, + + // Mirror destination: the Python side reads the same packets iMotions gets + pub mirror_addr_field: String, + pub mirror_connected: bool, } impl Default for State { @@ -124,6 +150,9 @@ impl Default for State { sent_packet_times: std::collections::VecDeque::new(), sent_samples: std::collections::VecDeque::new(), + + mirror_addr_field: String::new(), + mirror_connected: false, } } } @@ -150,6 +179,65 @@ fn normalize_baton_payload(raw: &str) -> Vec { .collect() } +/// The iMotions packets one baton sample turns into, in send order. +/// +/// Pulled out of the TCP thread so the mapping can be tested without a socket, +/// and so both destinations are written the same bytes from one list rather +/// than each building its own. +fn packets_for_fields(fields: &[String]) -> Vec { + let pair = |value: &String| vec![value.clone(), value.clone()]; + + if fields.len() == 4 { + let (alt, air, head, vv) = (&fields[0], &fields[1], &fields[2], &fields[3]); + return vec![ + build_imotions_packet("AltitudeSync", &pair(alt)), + build_imotions_packet("AirspeedSync", &pair(air)), + build_imotions_packet("VerticalVelocitySync", &pair(vv)), + build_imotions_packet("HeadingSync", &pair(head)), + ]; + } + + if fields.len() == 8 { + let (alt, air, head, vv) = (&fields[0], &fields[1], &fields[2], &fields[3]); + let (roll, pitch, yaw, gforce) = (&fields[4], &fields[5], &fields[6], &fields[7]); + return vec![ + build_imotions_packet("AltitudeSync", &pair(alt)), + build_imotions_packet("AirspeedSync", &pair(air)), + build_imotions_packet("VerticalVelocitySync", &pair(vv)), + build_imotions_packet("HeadingSync", &pair(head)), + build_imotions_packet("RollSync", &pair(roll)), + build_imotions_packet("PitchSync", &pair(pitch)), + build_imotions_packet("YawSync", &pair(yaw)), + build_imotions_packet("GForceSync", &pair(gforce)), + ]; + } + + // Anything else: take the fields two at a time, in event order, and stop + // when they run out. "Event" here is one iMotions entry, not a Sample in + // the GLOSSARY sense, which is a whole set of telemetry values. + const IMOTIONS_EVENTS: [&str; 8] = [ + "AltitudeSync", + "AirspeedSync", + "VerticalVelocitySync", + "HeadingSync", + "RollSync", + "PitchSync", + "YawSync", + "GForceSync", + ]; + let mut packets = Vec::new(); + let mut idx = 0usize; + for name in IMOTIONS_EVENTS { + if idx + 1 >= fields.len() { + break; + } + let payload = vec![fields[idx].clone(), fields[idx + 1].clone()]; + packets.push(build_imotions_packet(name, &payload)); + idx += 2; + } + packets +} + fn build_imotions_packet(event_name: &str, fields: &[String]) -> String { let mut packet = String::from("E;1;PilotDataSync;;;;;"); packet.push_str(event_name); @@ -264,6 +352,148 @@ fn human_log(level: &str, msg: &str) { } } +/// What the mirror's writer thread reports back to the TCP thread. +enum MirrorEvent { + Connected(String), + Disconnected(String), +} + +/// The TCP thread's handle on the mirror: a queue into the writer thread, and +/// the writer thread's reports coming back. +struct Mirror { + packets: crossbeam::channel::Sender, + events: crossbeam::channel::Receiver, + dropped: usize, + /// Whether we are in a run of drops, so the log gets one line when the + /// mirror falls behind and one when it recovers, not one per packet. + lagging: bool, +} + +/// Start the mirror's writer thread and hand back a queue into it. +/// +/// Connecting and writing both happen on that thread, so neither can delay the +/// iMotions stream. The TCP thread only ever pushes into a bounded queue, which +/// either takes the packet immediately or refuses it. +fn spawn_mirror(address: String) -> Mirror { + let (packets_tx, packets_rx) = crossbeam::channel::bounded::(MIRROR_QUEUE_CAPACITY); + let (events_tx, events_rx) = crossbeam::channel::unbounded(); + + spawn(move || { + let mut stream = match connect_mirror(&address) { + Ok(stream) => { + human_log("TCP", &format!("Python mirror connected to {}", address)); + let _ = events_tx.send(MirrorEvent::Connected(address.clone())); + stream + } + Err(e) => { + let reason = format!("Python mirror connection failed: {}", e); + human_log("TCP", &reason); + let _ = events_tx.send(MirrorEvent::Disconnected(reason)); + return; + } + }; + + // Ends when the TCP thread drops its sender, which is how the mirror is + // closed and how it goes away with the connection it belongs to. + for packet in packets_rx { + let write = stream + .write_all(packet.as_bytes()) + .and_then(|()| stream.flush()); + if let Err(e) = write { + let reason = format!("Python mirror dropped: {}", e); + human_log("TCP", &reason); + let _ = events_tx.send(MirrorEvent::Disconnected(reason)); + return; + } + } + }); + + Mirror { + packets: packets_tx, + events: events_rx, + dropped: 0, + lagging: false, + } +} + +/// Open the mirror connection. Runs on the writer thread, so the timeout here +/// costs the iMotions stream nothing. +fn connect_mirror(address: &str) -> Result { + let addr: SocketAddr = address + .parse() + .map_err(|_| anyhow!("Mirror address is not a literal IP and port: {address}"))?; + let stream = TcpStream::connect_timeout(&addr, MIRROR_CONNECT_TIMEOUT)?; + stream.set_write_timeout(Some(MIRROR_WRITE_TIMEOUT))?; + stream.set_nodelay(true)?; + Ok(stream) +} + +/// Hand one packet to the mirror's writer thread, if there is one. +/// +/// Never waits. A full queue means the reader is behind, so the packet is +/// dropped and counted. Losing copies is the right trade: the mirror is for +/// analysis that can tolerate gaps, and the iMotions recording cannot be redone. +fn mirror_send( + mirror: &mut Option, + packet: &str, + child_bichannel: &mut ChildBiChannel, +) { + let Some(active) = mirror.as_mut() else { + return; + }; + match active.packets.try_send(packet.to_string()) { + Ok(()) => { + if active.lagging { + let note = format!( + "Python mirror caught up, {} copies dropped while it was behind", + active.dropped + ); + human_log("TCP", ¬e); + let _ = child_bichannel.send_to_parent(FromTcpThreadMessage::MirrorLagging(note)); + active.lagging = false; + active.dropped = 0; + } + } + Err(_) => { + active.dropped += 1; + // One line per episode. At 160 packets a second, reporting each + // drop would push everything else out of the capped event log. + if !active.lagging { + active.lagging = true; + let note = + "Python mirror is behind, dropping copies. iMotions is unaffected".to_string(); + human_log("TCP", ¬e); + let _ = child_bichannel.send_to_parent(FromTcpThreadMessage::MirrorLagging(note)); + } + } + } +} + +/// Forward anything the writer thread has reported, and clear the mirror if it +/// has gone away. +fn drain_mirror_events( + mirror: &mut Option, + child_bichannel: &mut ChildBiChannel, +) { + let events: Vec = match mirror.as_ref() { + Some(active) => active.events.try_iter().collect(), + None => return, + }; + for event in events { + match event { + MirrorEvent::Connected(address) => { + let _ = child_bichannel.send_to_parent(FromTcpThreadMessage::MirrorConnected(address)); + } + MirrorEvent::Disconnected(reason) => { + let _ = child_bichannel + .send_to_parent(FromTcpThreadMessage::MirrorDisconnected(reason)); + *mirror = None; + return; + } + } + } +} + fn send_packet_and_debug(stream: &mut TcpStream, packet: &str) -> Result<()> { human_log("TX", &format!("packet len={} text={:?}", packet.len(), packet)); let hex: String = packet @@ -322,7 +552,7 @@ impl State { }; for address in addresses { let candidate = address.trim(); - if Self::validate_tcp_addr(candidate).is_ok() { + if Self::validate_tcp_addr(candidate, IMOTIONS_ADDR_HINT).is_ok() { uniq.insert(candidate.to_string()); } } @@ -349,12 +579,14 @@ impl State { Ok(()) } - pub fn validate_tcp_addr(address: &str) -> Result<()> { + /// `example` is the address hint for whichever box is being validated, so + /// the mirror does not tell people to type an iMotions address. + pub fn validate_tcp_addr(address: &str, example: &str) -> Result<()> { let _parsed = address .trim() .parse::() .map_err(|_| { - anyhow!("Invalid address. Enter a literal IP and port, for example {IMOTIONS_ADDR_HINT}") + anyhow!("Invalid address. Enter a literal IP and port, for example {example}") })?; Ok(()) } @@ -602,13 +834,18 @@ impl State { bail!(reason); } }; - + + // Second destination for the same packets, set by SetMirror. Its + // socket is owned by a writer thread; this is only the queue into + // it, so nothing here can be slowed by the Python side. + let mut mirror: Option = None; + while !child_bichannel.is_killswitch_engaged() { + drain_mirror_events(&mut mirror, &mut child_bichannel); for message in child_bichannel.received_messages() { match message { ToTcpThreadMessage::Send(data) => { let fields = normalize_baton_payload(&data); - if fields.len() < 2 { human_log("TCP", &format!( @@ -618,70 +855,11 @@ impl State { continue; } - let mut try_send = |pkt: String| -> Result<()> { - let res = send_packet_and_debug(&mut stream, &pkt); - if res.is_err() { - let _ = child_bichannel.set_is_conn_to_endpoint(false); - } else { - let _ = child_bichannel.set_is_conn_to_endpoint(true); - } - res - }; - - if fields.len() == 4 { - let alt = fields[0].clone(); - let air = fields[1].clone(); - let head = fields[2].clone(); - let vv = fields[3].clone(); - - let _ = try_send(build_imotions_packet("AltitudeSync", &[alt.clone(), alt.clone()])); - let _ = try_send(build_imotions_packet("AirspeedSync", &[air.clone(), air.clone()])); - let _ = try_send(build_imotions_packet("VerticalVelocitySync", &[vv.clone(), vv.clone()])); - let _ = try_send(build_imotions_packet("HeadingSync", &[head.clone(), head.clone()])); - continue; - } - - if fields.len() == 8 { - let alt = fields[0].clone(); - let air = fields[1].clone(); - let head = fields[2].clone(); - let vv = fields[3].clone(); - let roll = fields[4].clone(); - let pitch = fields[5].clone(); - let yaw = fields[6].clone(); - let gforce = fields[7].clone(); - - let _ = try_send(build_imotions_packet("AltitudeSync", &[alt.clone(), alt.clone()])); - let _ = try_send(build_imotions_packet("AirspeedSync", &[air.clone(), air.clone()])); - let _ = try_send(build_imotions_packet("VerticalVelocitySync", &[vv.clone(), vv.clone()])); - let _ = try_send(build_imotions_packet("HeadingSync", &[head.clone(), head.clone()])); - let _ = try_send(build_imotions_packet("RollSync", &[roll.clone(), roll.clone()])); - let _ = try_send(build_imotions_packet("PitchSync", &[pitch.clone(), pitch.clone()])); - let _ = try_send(build_imotions_packet("YawSync", &[yaw.clone(), yaw.clone()])); - let _ = try_send(build_imotions_packet("GForceSync", &[gforce.clone(), gforce.clone()])); - continue; + for packet in packets_for_fields(&fields) { + let res = send_packet_and_debug(&mut stream, &packet); + let _ = child_bichannel.set_is_conn_to_endpoint(res.is_ok()); + mirror_send(&mut mirror, &packet, &mut child_bichannel); } - - // Paired mapping - let mut idx = 0usize; - let mut send_pair_if_present = |name: &str, idx: &mut usize| -> Result<()> { - if *idx + 1 < fields.len() { - let payload = vec![fields[*idx].clone(), fields[*idx + 1].clone()]; - let pkt = build_imotions_packet(name, &payload); - *idx += 2; - try_send(pkt)?; - } - Ok(()) - }; - - let _ = send_pair_if_present("AltitudeSync", &mut idx); - let _ = send_pair_if_present("AirspeedSync", &mut idx); - let _ = send_pair_if_present("VerticalVelocitySync", &mut idx); - let _ = send_pair_if_present("HeadingSync", &mut idx); - let _ = send_pair_if_present("RollSync", &mut idx); - let _ = send_pair_if_present("PitchSync", &mut idx); - let _ = send_pair_if_present("YawSync", &mut idx); - let _ = send_pair_if_present("GForceSync", &mut idx); } ToTcpThreadMessage::SendRaw(packet) => { let res = send_packet_and_debug(&mut stream, &packet); @@ -689,6 +867,27 @@ impl State { if let Err(e) = res { human_log("TCP", &format!("Test packet failed: {}", e)); } + // The mirror gets the test packet too, so the button + // checks both links at once. + mirror_send(&mut mirror, &packet, &mut child_bichannel); + } + ToTcpThreadMessage::SetMirror(target) => { + // Dropping the old handle closes its queue, which + // ends that writer thread and its socket with it. + mirror = None; + match target { + None => { + let _ = child_bichannel.send_to_parent( + FromTcpThreadMessage::MirrorDisconnected( + "closed".to_string(), + ), + ); + } + // Connecting happens on the writer thread, so + // this returns immediately and the result + // arrives as a MirrorEvent. + Some(address) => mirror = Some(spawn_mirror(address)), + } } } } @@ -704,6 +903,32 @@ impl State { Ok(()) } + /// Point the mirror at `address`, so the Python side receives a copy of + /// every packet (#245). + /// + /// The mirror lives inside the TCP thread, so iMotions has to be connected + /// first. With no iMotions to talk to, point Connect TCP straight at the + /// Python listener instead and leave the mirror unused. + pub fn mirror_connect(&mut self, address: String) -> Result<()> { + Self::validate_tcp_addr(&address, MIRROR_ADDR_HINT)?; + // validate_tcp_addr trims before parsing, so trim here too. Otherwise a + // pasted address with a stray space passes validation and then fails to + // parse inside the thread. + let address = address.trim().to_string(); + let Some(bichannel) = self.tcp_bichannel.as_mut() else { + bail!("Connect TCP first. The mirror copies that stream.") + }; + bichannel.send_to_child(ToTcpThreadMessage::SetMirror(Some(address))) + } + + pub fn mirror_disconnect(&mut self) -> Result<()> { + let Some(bichannel) = self.tcp_bichannel.as_mut() else { + self.mirror_connected = false; + return Ok(()); + }; + bichannel.send_to_child(ToTcpThreadMessage::SetMirror(None)) + } + pub fn tcp_disconnect(&mut self) -> Result<()> { if self.tcp_thread_handle.is_none() { bail!("TCP thread does not exist.") @@ -715,6 +940,9 @@ impl State { }; let was_connected = bichannel.is_conn_to_endpoint().unwrap_or(false); bichannel.killswitch_engage()?; + // The mirror lives in that thread, so it goes down with it. Without + // this the GUI keeps claiming the mirror is connected. + self.mirror_connected = false; if !was_connected { spawn(move || { @@ -738,6 +966,8 @@ impl State { } self.tcp_bichannel = None; self.tcp_connected = false; + // The mirror lived in that thread, so it is gone with it. + self.mirror_connected = false; } pub fn is_tcp_connected(&self) -> bool { @@ -774,6 +1004,53 @@ mod tests { assert!(!state.event_log.iter().any(|e| e.ends_with("event 0"))); } + fn fields(values: &[&str]) -> Vec { + values.iter().map(|v| v.to_string()).collect() + } + + #[test] + fn eight_fields_become_one_packet_per_event_in_order() { + let packets = packets_for_fields(&fields(&[ + "1000", "120", "90", "500", "2", "3", "91", "1.0", + ])); + + let names: Vec<&str> = packets + .iter() + .map(|p| p.split(';').nth(7).expect("no sample name")) + .collect(); + assert_eq!( + names, + [ + "AltitudeSync", + "AirspeedSync", + "VerticalVelocitySync", + "HeadingSync", + "RollSync", + "PitchSync", + "YawSync", + "GForceSync", + ] + ); + // Heading is field 2 and vertical velocity field 3, but they are sent + // the other way round. Pin that, it is easy to "tidy" and break. + assert_eq!(packets[2], "E;1;PilotDataSync;;;;;VerticalVelocitySync;500;500\r\n"); + assert_eq!(packets[3], "E;1;PilotDataSync;;;;;HeadingSync;90;90\r\n"); + } + + #[test] + fn four_fields_send_only_the_flight_dynamics_factors() { + let packets = packets_for_fields(&fields(&["1000", "120", "90", "500"])); + assert_eq!(packets.len(), 4); + assert_eq!(packets[0], "E;1;PilotDataSync;;;;;AltitudeSync;1000;1000\r\n"); + } + + #[test] + fn an_odd_field_count_pairs_what_it_can_and_drops_the_remainder() { + let packets = packets_for_fields(&fields(&["1000", "120", "90"])); + assert_eq!(packets.len(), 1); + assert_eq!(packets[0], "E;1;PilotDataSync;;;;;AltitudeSync;1000;120\r\n"); + } + #[test] fn test_packet_matches_imotions_format() { let state = State::default(); diff --git a/relay/src/update.rs b/relay/src/update.rs index b8acc74..b10ec37 100644 --- a/relay/src/update.rs +++ b/relay/src/update.rs @@ -7,12 +7,12 @@ use chrono::Local; use crate::{ message::FromTcpThreadMessage, message::FromIpcThreadMessage, message::ToTcpThreadMessage, - state::build_test_packet, Message, State, + state::build_test_packet, state::IMOTIONS_ADDR_HINT, Message, State, }; fn connect_tcp_with_validation_and_save(state: &mut State, address: String) { let trimmed = address.trim().to_string(); - match State::validate_tcp_addr(&trimmed) { + match State::validate_tcp_addr(&trimmed, IMOTIONS_ADDR_HINT) { Ok(()) => { state.tcp_addr_validation_error = None; state.tcp_addr_field = trimmed.clone(); @@ -92,6 +92,17 @@ pub(crate) fn update(state: &mut State, message: Message) -> Task { FromTcpThreadMessage::SendError(err) => { state.log_event(format!("TCP send error: {}", err)); } + FromTcpThreadMessage::MirrorConnected(address) => { + state.log_event(format!("Python mirror connected to {}", address)); + state.mirror_connected = true; + } + FromTcpThreadMessage::MirrorDisconnected(reason) => { + state.log_event(format!("Python mirror stopped: {}", reason)); + state.mirror_connected = false; + } + FromTcpThreadMessage::MirrorLagging(reason) => { + state.log_event(reason); + } } } @@ -189,6 +200,25 @@ pub(crate) fn update(state: &mut State, message: Message) -> Task { state.tcp_addr_validation_error = None; Task::none() } + M::MirrorAddrFieldUpdate(addr) => { + state.mirror_addr_field = addr; + Task::none() + } + M::ConnectMirror => { + let address = state.mirror_addr_field.clone(); + if let Err(e) = state.mirror_connect(address) { + let msg = format!("Python mirror: {}", e); + state.error_message = Some(msg.clone()); + state.log_event(msg); + } + Task::none() + } + M::DisconnectMirror => { + if let Err(e) = state.mirror_disconnect() { + state.log_event(format!("Python mirror disconnect failed: {}", e)); + } + Task::none() + } M::SendPacket => { let connected = state.is_tcp_connected(); let packet = build_test_packet(state); @@ -329,7 +359,7 @@ fn create_xml_file(state: &mut State) -> Task { mod tests { use super::*; use std::io::{BufRead, BufReader}; - use std::net::TcpListener; + use std::net::{TcpListener, TcpStream}; use std::thread::sleep; // Waits for a condition, polling every 10ms, so the test does not hang. @@ -380,6 +410,20 @@ mod tests { } } + // Pumps until the state satisfies `condition`, returning as soon as it + // does. A fixed tick count turns a busy machine into a failing test, since + // these are all waits on a background thread rather than on elapsed time. + fn pump_until(state: &mut State, mut condition: impl FnMut(&State) -> bool) -> bool { + for _ in 0..1000 { + let _ = update(state, Message::Update); + if condition(state) { + return true; + } + sleep(StdDuration::from_millis(10)); + } + false + } + // A port nothing listens on, obtained by binding and then dropping. fn closed_port_addr() -> String { let listener = TcpListener::bind("127.0.0.1:0").expect("bind failed"); @@ -395,7 +439,10 @@ mod tests { .tcp_connect(closed_port_addr()) .expect("tcp_connect should spawn a thread"); - pump_update(&mut state, 200); + assert!( + pump_until(&mut state, |s| s.error_message.is_some()), + "no error surfaced" + ); let error = state.error_message.clone().expect("no error surfaced"); assert!( @@ -426,6 +473,545 @@ mod tests { let _ = state.tcp_disconnect(); } + // Binds a listener and returns it with its address. + fn listener_on_free_port() -> (TcpListener, String) { + let listener = TcpListener::bind("127.0.0.1:0").expect("bind failed"); + let addr = listener.local_addr().expect("local_addr failed").to_string(); + (listener, addr) + } + + fn read_lines(stream: &TcpStream, count: usize) -> Vec { + stream + .set_read_timeout(Some(StdDuration::from_secs(5))) + .expect("set_read_timeout failed"); + let mut reader = BufReader::new(stream); + (0..count) + .map(|_| { + let mut line = String::new(); + reader.read_line(&mut line).expect("read failed"); + line + }) + .collect() + } + + // The point of #245: Python sees the stream without taking it away from + // iMotions. + #[test] + fn the_python_mirror_receives_the_same_packets_as_imotions() { + let (imotions, imotions_addr) = listener_on_free_port(); + let (python, python_addr) = listener_on_free_port(); + + let mut state = State::default(); + state.tcp_connect(imotions_addr).expect("tcp_connect failed"); + let (imotions_stream, _) = imotions.accept().expect("accept failed"); + assert!(wait_until(|| state.is_tcp_connected()), "TCP never connected"); + + state + .mirror_connect(python_addr) + .expect("mirror_connect failed"); + let (python_stream, _) = python.accept().expect("mirror accept failed"); + assert!( + pump_until(&mut state, |s| s.mirror_connected), + "mirror never reported connected" + ); + + // One eight-value sample, the shape the plugin sends. + state + .tcp_bichannel + .as_mut() + .expect("no bichannel") + .send_to_child(ToTcpThreadMessage::Send( + "1000;120;90;500;2;3;91;1.0".to_string(), + )) + .expect("send failed"); + + let to_imotions = read_lines(&imotions_stream, 8); + let to_python = read_lines(&python_stream, 8); + assert_eq!(to_imotions, to_python, "the two sinks disagree"); + assert_eq!( + to_imotions[0], + "E;1;PilotDataSync;;;;;AltitudeSync;1000;1000\r\n" + ); + + let _ = state.tcp_disconnect(); + } + + // A missing or dead Python process must not disturb the recording. + #[test] + fn imotions_keeps_streaming_when_the_mirror_cannot_connect() { + let (imotions, imotions_addr) = listener_on_free_port(); + + let mut state = State::default(); + state.tcp_connect(imotions_addr).expect("tcp_connect failed"); + let (imotions_stream, _) = imotions.accept().expect("accept failed"); + assert!(wait_until(|| state.is_tcp_connected()), "TCP never connected"); + + state + .mirror_connect(closed_port_addr()) + .expect("the request itself should be accepted"); + assert!( + pump_until(&mut state, |s| !s.mirror_connected), + "mirror claimed to be connected" + ); + + let _ = update(&mut state, Message::SendPacket); + let lines = read_lines(&imotions_stream, 1); + assert_eq!(lines[0], "E;1;PilotDataSync;;;;;AltitudeSync;0;0\r\n"); + assert!(state.is_tcp_connected(), "iMotions was dropped with the mirror"); + + let _ = state.tcp_disconnect(); + } + + // Sends one eight-value sample the way baton would. + fn send_sample(state: &mut State, altitude: &str) { + state + .tcp_bichannel + .as_mut() + .expect("no bichannel") + .send_to_child(ToTcpThreadMessage::Send(format!( + "{altitude};120;90;500;2;3;91;1.0" + ))) + .expect("send failed"); + } + + // The case that actually happens in a lab: Python is killed or crashes + // while the flight is still being recorded. + #[test] + fn imotions_survives_the_python_side_dying_mid_stream() { + let (imotions, imotions_addr) = listener_on_free_port(); + let (python, python_addr) = listener_on_free_port(); + + let mut state = State::default(); + state.tcp_connect(imotions_addr).expect("tcp_connect failed"); + let (imotions_stream, _) = imotions.accept().expect("accept failed"); + assert!(wait_until(|| state.is_tcp_connected()), "TCP never connected"); + + state.mirror_connect(python_addr).expect("mirror_connect failed"); + let (python_stream, _) = python.accept().expect("mirror accept failed"); + assert!( + pump_until(&mut state, |s| s.mirror_connected), + "mirror never reported connected" + ); + + send_sample(&mut state, "1000"); + assert_eq!(read_lines(&python_stream, 8).len(), 8); + + // Python goes away mid-flight. + drop(python_stream); + drop(python); + + // Keep streaming. The write fails once the peer is gone, which may take + // a sample or two to surface. + for i in 0..20 { + send_sample(&mut state, &format!("{}", 2000 + i)); + pump_update(&mut state, 2); + } + + assert!( + wait_until(|| { + let _ = update(&mut state, Message::Update); + !state.mirror_connected + }), + "mirror never noticed Python was gone" + ); + assert!(state.is_tcp_connected(), "iMotions went down with the mirror"); + assert!( + state.event_log.iter().any(|e| e.contains("Python mirror stopped")), + "nothing in the event log: {:?}", + state.event_log + ); + + // iMotions still receives everything sent after the mirror died. + send_sample(&mut state, "9999"); + let lines = read_lines(&imotions_stream, 8); + assert!( + lines[0].contains("AltitudeSync"), + "unexpected line: {}", + lines[0] + ); + + let _ = state.tcp_disconnect(); + } + + // Counts lines off a socket on its own thread, the way a real recorder + // would. The counter is readable while it runs, so a test can wait for + // delivery instead of closing the socket to find out. + fn drain_in_background(stream: TcpStream) -> std::sync::Arc { + let counter = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let seen = counter.clone(); + std::thread::spawn(move || { + let mut reader = BufReader::new(stream); + let mut line = String::new(); + while reader.read_line(&mut line).unwrap_or(0) > 0 { + line.clear(); + seen.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + } + }); + counter + } + + // Connecting the mirror used to happen on the TCP thread, where a 500ms + // connect timeout stalled the iMotions stream. It now runs on the mirror's + // own thread, so streaming continues while the attempt is in flight. + #[test] + fn connecting_the_mirror_does_not_pause_the_imotions_stream() { + // RFC 5737 TEST-NET-1: routed nowhere, so the connect attempt runs to + // its timeout rather than being refused. + const BLACKHOLE: &str = "192.0.2.1:5001"; + + let (imotions, imotions_addr) = listener_on_free_port(); + let mut state = State::default(); + state.tcp_connect(imotions_addr).expect("tcp_connect failed"); + let (imotions_stream, _) = imotions.accept().expect("accept failed"); + assert!(wait_until(|| state.is_tcp_connected()), "TCP never connected"); + let delivered = drain_in_background(imotions_stream); + + state + .mirror_connect(BLACKHOLE.to_string()) + .expect("mirror_connect failed"); + + // Immediately, while the connect is still being attempted. + let started = std::time::Instant::now(); + for i in 0..20 { + send_sample(&mut state, &format!("{}", 1000 + i)); + } + assert!( + wait_until(|| delivered.load(std::sync::atomic::Ordering::Relaxed) >= 160), + "iMotions got {} of 160 packets while the mirror was connecting", + delivered.load(std::sync::atomic::Ordering::Relaxed) + ); + let elapsed = started.elapsed(); + assert!( + elapsed < StdDuration::from_millis(400), + "iMotions waited {elapsed:?} for the mirror's connect attempt" + ); + + let _ = state.tcp_disconnect(); + } + + // A Python process that reads, but slower than the relay writes. iMotions + // must still receive every packet, and the relay must say the copy is + // falling behind rather than quietly coping. + #[test] + fn a_slow_python_reader_does_not_pace_imotions() { + // Enough to exceed the socket buffers, which absorb a surprising amount: + // at 500 samples the slow reader never creates any backpressure at all. + const SAMPLES: usize = 1500; + const PACKETS: usize = SAMPLES * 8; + + let (imotions, imotions_addr) = listener_on_free_port(); + let (python, python_addr) = listener_on_free_port(); + + let mut state = State::default(); + state.tcp_connect(imotions_addr).expect("tcp_connect failed"); + let (imotions_stream, _) = imotions.accept().expect("accept failed"); + assert!(wait_until(|| state.is_tcp_connected()), "TCP never connected"); + let imotions_count = drain_in_background(imotions_stream); + + state.mirror_connect(python_addr).expect("mirror_connect failed"); + let (slow, _) = python.accept().expect("accept failed"); + assert!( + pump_until(&mut state, |s| s.mirror_connected), + "mirror never reported connected" + ); + + // Reads steadily but slower than the relay writes: about 1000 lines a + // second against 12000 queued. Fast enough that writes keep succeeding, + // so the mirror stays up and the queue is what fills, which is the case + // that used to pace iMotions. + let slow_reader = std::thread::spawn(move || { + let mut reader = BufReader::new(slow); + let mut line = String::new(); + let mut count = 0usize; + while reader.read_line(&mut line).unwrap_or(0) > 0 { + line.clear(); + count += 1; + sleep(StdDuration::from_millis(1)); + } + count + }); + + let started = std::time::Instant::now(); + for i in 0..SAMPLES { + send_sample(&mut state, &format!("{}", 1000 + i % 100)); + } + let queued = started.elapsed(); + assert!( + queued < StdDuration::from_secs(3), + "queueing {SAMPLES} samples took {queued:?}, the slow reader is holding the thread" + ); + + // Everything reaches iMotions long before the slow reader could have + // consumed it: pacing to 20ms a line would take about 80 seconds. + let delivered = |count: &std::sync::atomic::AtomicUsize| { + count.load(std::sync::atomic::Ordering::Relaxed) + }; + assert!( + wait_until(|| delivered(&imotions_count) >= PACKETS), + "iMotions only got {} of {PACKETS} packets", + delivered(&imotions_count) + ); + let full_delivery = started.elapsed(); + assert!( + full_delivery < StdDuration::from_secs(10), + "iMotions took {full_delivery:?}, it is being paced by the mirror" + ); + + // The relay says something about it rather than quietly coping: either + // copies are being dropped, or the writer gave up on the write timeout. + // Which one depends on how much the socket buffers absorb. + assert!( + wait_until(|| { + let _ = update(&mut state, Message::Update); + !state.mirror_connected + || state.event_log.iter().any(|e| e.contains("behind")) + }), + "a slow mirror was never reported: {:?}", + state.event_log + ); + assert!( + state.is_tcp_connected(), + "iMotions was dropped along with the slow mirror" + ); + + drop(slow_reader); + let _ = state.tcp_disconnect(); + } + + // The ADR 0003 claim: a Python process that accepts the connection and then + // stops reading gets dropped on the write timeout instead of stalling the + // feed to iMotions. + #[test] + fn a_python_process_that_stops_reading_is_dropped_not_waited_on() { + let (imotions, imotions_addr) = listener_on_free_port(); + let (python, python_addr) = listener_on_free_port(); + + let mut state = State::default(); + state.tcp_connect(imotions_addr).expect("tcp_connect failed"); + let (imotions_stream, _) = imotions.accept().expect("accept failed"); + assert!(wait_until(|| state.is_tcp_connected()), "TCP never connected"); + + // iMotions keeps draining, as a real recorder would. + let drain = std::thread::spawn(move || { + let mut reader = BufReader::new(imotions_stream); + let mut sink = String::new(); + let mut lines = 0usize; + while reader.read_line(&mut sink).unwrap_or(0) > 0 { + sink.clear(); + lines += 1; + } + lines + }); + + // Python accepts, then never reads a byte. + state.mirror_connect(python_addr).expect("mirror_connect failed"); + let (stalled, _) = python.accept().expect("accept failed"); + assert!( + pump_until(&mut state, |s| s.mirror_connected), + "mirror never reported connected" + ); + + let started = std::time::Instant::now(); + for i in 0..3000 { + send_sample(&mut state, &format!("{}", 1000 + i % 100)); + } + assert!( + wait_until(|| { + let _ = update(&mut state, Message::Update); + !state.mirror_connected + }), + "a stalled reader was never dropped" + ); + let elapsed = started.elapsed(); + + assert!(state.is_tcp_connected(), "iMotions was taken down with it"); + // One write timeout, not one per packet. + assert!( + elapsed < StdDuration::from_secs(20), + "took {elapsed:?}, the stalled mirror is holding up the stream" + ); + + drop(stalled); + let _ = state.tcp_disconnect(); + let delivered = drain.join().unwrap_or(0); + assert!( + delivered > 1000, + "iMotions only got {delivered} lines while the mirror was stalled" + ); + } + + #[test] + fn the_mirror_can_be_reconnected_after_python_restarts() { + let (imotions, imotions_addr) = listener_on_free_port(); + + let mut state = State::default(); + state.tcp_connect(imotions_addr).expect("tcp_connect failed"); + let (_imotions_stream, _) = imotions.accept().expect("accept failed"); + assert!(wait_until(|| state.is_tcp_connected()), "TCP never connected"); + + // First Python process. + let (python, python_addr) = listener_on_free_port(); + state + .mirror_connect(python_addr.clone()) + .expect("mirror_connect failed"); + let (first, _) = python.accept().expect("accept failed"); + assert!( + pump_until(&mut state, |s| s.mirror_connected), + "mirror never reported connected" + ); + drop(first); + drop(python); + + // Second Python process on the same address. + let listener = TcpListener::bind(&python_addr).expect("rebind failed"); + state.mirror_connect(python_addr).expect("reconnect rejected"); + let (second, _) = listener.accept().expect("accept failed"); + assert!( + pump_until(&mut state, |s| s.mirror_connected), + "mirror did not come back" + ); + + send_sample(&mut state, "1234"); + let lines = read_lines(&second, 8); + assert_eq!(lines[0], "E;1;PilotDataSync;;;;;AltitudeSync;1234;1234\r\n"); + + let _ = state.tcp_disconnect(); + } + + #[test] + fn disconnecting_tcp_also_clears_the_mirror_status() { + let (imotions, imotions_addr) = listener_on_free_port(); + let (python, python_addr) = listener_on_free_port(); + + let mut state = State::default(); + state.tcp_connect(imotions_addr).expect("tcp_connect failed"); + let (_imotions_stream, _) = imotions.accept().expect("accept failed"); + assert!(wait_until(|| state.is_tcp_connected()), "TCP never connected"); + state.mirror_connect(python_addr).expect("mirror_connect failed"); + let (_python_stream, _) = python.accept().expect("accept failed"); + assert!( + pump_until(&mut state, |s| s.mirror_connected), + "mirror never reported connected" + ); + + let _ = state.tcp_disconnect(); + + assert!( + !state.mirror_connected, + "GUI still claims the mirror is up after Disconnect TCP" + ); + } + + // 100 samples is 800 packets to each sink. Checks nothing is dropped, + // reordered or interleaved between the two. + #[test] + fn a_long_run_reaches_both_sinks_in_order() { + let (imotions, imotions_addr) = listener_on_free_port(); + let (python, python_addr) = listener_on_free_port(); + + let mut state = State::default(); + state.tcp_connect(imotions_addr).expect("tcp_connect failed"); + let (imotions_stream, _) = imotions.accept().expect("accept failed"); + assert!(wait_until(|| state.is_tcp_connected()), "TCP never connected"); + state.mirror_connect(python_addr).expect("mirror_connect failed"); + let (python_stream, _) = python.accept().expect("accept failed"); + assert!( + pump_until(&mut state, |s| s.mirror_connected), + "mirror never reported connected" + ); + + const SAMPLES: usize = 100; + for i in 0..SAMPLES { + send_sample(&mut state, &format!("{}", 1000 + i)); + } + + let to_imotions = read_lines(&imotions_stream, SAMPLES * 8); + let to_python = read_lines(&python_stream, SAMPLES * 8); + assert_eq!(to_imotions, to_python, "the sinks diverged over a long run"); + + // Altitudes arrive in the order they were sent. + let altitudes: Vec<&String> = to_python + .iter() + .filter(|line| line.contains("AltitudeSync")) + .collect(); + assert_eq!(altitudes.len(), SAMPLES); + assert!(altitudes[0].contains(";1000;1000"), "{}", altitudes[0]); + assert!( + altitudes[SAMPLES - 1].contains(&format!(";{};{}", 1000 + SAMPLES - 1, 1000 + SAMPLES - 1)), + "{}", + altitudes[SAMPLES - 1] + ); + assert!(state.mirror_connected, "mirror dropped during a normal run"); + + let _ = state.tcp_disconnect(); + } + + #[test] + fn a_mirror_address_with_stray_whitespace_still_connects() { + let (imotions, imotions_addr) = listener_on_free_port(); + let (python, python_addr) = listener_on_free_port(); + + let mut state = State::default(); + state.tcp_connect(imotions_addr).expect("tcp_connect failed"); + let (_imotions_stream, _) = imotions.accept().expect("accept failed"); + assert!(wait_until(|| state.is_tcp_connected()), "TCP never connected"); + + state + .mirror_connect(format!(" {} ", python_addr)) + .expect("padded address was rejected"); + let (_python_stream, _) = python.accept().expect("accept failed"); + assert!( + pump_until(&mut state, |s| s.mirror_connected), + "padded address never connected" + ); + + let _ = state.tcp_disconnect(); + } + + #[test] + fn a_malformed_mirror_address_is_refused_before_the_thread_sees_it() { + let (imotions, imotions_addr) = listener_on_free_port(); + let mut state = State::default(); + state.tcp_connect(imotions_addr).expect("tcp_connect failed"); + let (_imotions_stream, _) = imotions.accept().expect("accept failed"); + + let error = state + .mirror_connect("localhost:5001".to_string()) + .expect_err("host names are not supported"); + let message = error.to_string(); + assert!( + message.contains("literal IP and port"), + "unexpected error: {message}" + ); + // The mirror box must offer its own example. Borrowing the iMotions one + // would tell people to point Python at the recorder. + assert!( + message.contains("127.0.0.1:5001"), + "mirror error should name the mirror's example: {message}" + ); + assert!( + !message.contains("iMotions"), + "mirror error leaked the iMotions hint: {message}" + ); + assert!(!state.mirror_connected); + + let _ = state.tcp_disconnect(); + } + + #[test] + fn the_mirror_needs_tcp_connected_first() { + let mut state = State::default(); + let (_python, python_addr) = listener_on_free_port(); + + let error = state + .mirror_connect(python_addr) + .expect_err("mirror without TCP should fail"); + assert!( + error.to_string().contains("Connect TCP first"), + "unexpected error: {error}" + ); + } + #[test] fn send_packet_with_every_toggle_off_reports_an_error() { let mut state = State::default(); diff --git a/relay/src/view.rs b/relay/src/view.rs index 3732a13..2b00c33 100644 --- a/relay/src/view.rs +++ b/relay/src/view.rs @@ -7,7 +7,7 @@ use iced::widget::{button, column, container, pick_list, row, text, text_input, use iced::{Element, Length}; use iced_aw::{helpers::card, style}; -use crate::state::IMOTIONS_ADDR_HINT; +use crate::state::{IMOTIONS_ADDR_HINT, MIRROR_ADDR_HINT}; use crate::{Message, State}; type UIElement<'a> = Element<'a, Message>; @@ -44,6 +44,9 @@ pub(crate) fn view(state: &State) -> UIElement { elements.push(tcp_connect_button(state)); elements.push(tcp_disconnect_button()); + // Second destination, read by the Python side + elements.push(mirror_row(state)); + // Metrics (if enabled) elements.push(metrics_block(state)); @@ -216,6 +219,48 @@ fn tcp_disconnect_button() -> UIElement<'static> { .into() } +/// Build the mirror address input, wired to `Message::MirrorAddrFieldUpdate`. +fn mirror_addr_input(state: &State) -> UIElement { + text_input(MIRROR_ADDR_HINT, &state.mirror_addr_field) + .on_input(Message::MirrorAddrFieldUpdate) + .into() +} + +fn mirror_status_element(state: &State) -> UIElement { + let status = if state.mirror_connected { + "Python mirror: connected" + } else { + "Python mirror: not connected" + }; + text(status).into() +} + +fn mirror_connect_button() -> UIElement<'static> { + button("Connect mirror").on_press(Message::ConnectMirror).into() +} + +fn mirror_disconnect_button() -> UIElement<'static> { + button("Disconnect mirror") + .on_press(Message::DisconnectMirror) + .into() +} + +/// Mirror controls: a second destination that receives a copy of every packet, +/// so the Python side can read the stream while iMotions keeps recording. +fn mirror_row(state: &State) -> UIElement { + column![ + mirror_status_element(state), + row![ + mirror_connect_button(), + mirror_addr_input(state), + mirror_disconnect_button(), + ] + .spacing(5), + ] + .spacing(5) + .into() +} + /// XML download card or opener button depending on `state.card_open` fn xml_download_popup(state: &State) -> UIElement { if state.card_open {