use awc::{Client, BoxedSocket, ws::{Frame, Message, Codec}}; use actix_codec::Framed; use futures_util::{SinkExt, StreamExt}; use serde_json::{Value, from_str, json}; use tokio::sync::mpsc; pub async fn connect_to_ha( ip: String, token: String ) -> Result, Box> { let client = Client::new(); let (_resp, mut connection) = client .ws(format!("ws://{}:8123/api/websocket", ip)) .connect() .await?; while let Some(Ok(msg)) = connection.next().await { if let Frame::Text(bytes) = msg { let text = String::from_utf8(bytes.to_vec())?; let parsed: Value = from_str(&text)?; if parsed["type"] == "auth_required" { let auth_msg = json!({ "type": "auth", "access_token": token }); connection.send(Message::Text(auth_msg.to_string().into())).await?; } if parsed["type"] == "auth_ok" { /*let areas_msg = json!({"type": "config/area_registry/list", "id": message_id}); connection.send(Message::Text(areas_msg.to_string().into())).await?; message_id += 1;*/ /*let devices_msg = json!({"type": "config/device_registry/list", "id": message_id}); connection.send(Message::Text(devices_msg.to_string().into())).await?; message_id += 1;*/ /*let entities_msg = json!({"type": "config/entity_registry/list", "id": message_id}); connection.send(Message::Text(entities_msg.to_string().into())).await?; message_id += 1;*/ return Ok(connection); } if parsed["type"] == "auth_invalid" { return Err("HA authentication.failed".into()); } } } Err("Websocket closed before authentication completed".into()) } pub async fn listen( ip: String, token: String, mut outbound_rx: mpsc::UnboundedReceiver ) { loop { let mut connection = match connect_to_ha(ip.clone(), token.clone()).await { Ok(c) => c, Err(e) => { eprintln!("Failed to connect to HA: {e}. Retrying in 10s..."); tokio::time::sleep(std::time::Duration::from_secs(10)).await; continue; } }; loop { tokio::select! { msg = connection.next() => { match msg { Some(Ok(Frame::Text(bytes))) => { let text = match String::from_utf8(bytes.to_vec()) { Ok(t) => t, Err(_) => continue }; let parsed: Value = match from_str(&text) { Ok(v) => v, Err(_) => continue }; if parsed["type"] == "result" { println!("result"); println!("{:#?}", parsed); } } Some(Ok(Frame::Ping(bytes))) => { let _ = connection.send(Message::Pong(bytes)).await; } Some(Ok(Frame::Close(reason))) => { println!("HA closed connection: {:?}", reason); break; } Some(Err(e)) => { eprintln!("Websocket error: {e}"); break; } None => break, _ => {} } } Some(out_msg) = outbound_rx.recv() => { if let Err(e) = connection.send(out_msg).await { eprintln!("Failed to send outbound message: {e}"); } } } } println!("HA connection closed, reconnecting in 10s..."); tokio::time::sleep(std::time::Duration::from_secs(10)).await; } }