netget/server/bacnet/
mod.rs

1pub mod actions;
2pub mod codec;
3use crate::{
4    llm::actions::protocol_trait::ActionResult,
5    protocol::{Event, SpawnContext},
6};
7use anyhow::{ensure, Result};
8use std::{net::SocketAddr, sync::Arc, time::Duration};
9pub async fn spawn(ctx: SpawnContext) -> Result<SocketAddr> {
10    let device = ctx
11        .startup_params
12        .as_ref()
13        .map(|p| p.get_optional_u32("device_id"))
14        .transpose()?
15        .flatten()
16        .unwrap_or(codec::DEFAULT_DEVICE_ID);
17    ensure!(device <= 4194303, "device id out of range");
18    let socket = Arc::new(tokio::net::UdpSocket::bind(ctx.legacy_listen_addr()).await?);
19    let addr = socket.local_addr()?;
20    let state = ctx.state.clone();
21    let sid = ctx.server_id;
22    state
23        .spawn_server_task(sid, async move {
24            let cap = Arc::new(tokio::sync::Semaphore::new(256));
25            let mut buf = [0; codec::MAX_FRAME + 1];
26            while let Ok((n, peer)) = socket.recv_from(&mut buf).await {
27                let Ok(a) = codec::unwrap(&buf[..n]) else {
28                    continue;
29                };
30                let Ok((automatic, request)) = codec::receive(a, device) else {
31                    continue;
32                };
33                if !automatic.is_empty() {
34                    let _ = socket.send_to(&automatic, peer).await;
35                }
36                let Some(request) = request else { continue };
37                let Ok(permit) = cap.clone().try_acquire_owned() else {
38                    continue;
39                };
40                let c = ctx.clone();
41                let socket = socket.clone();
42                ctx.state
43                    .spawn_server_task(sid, async move {
44                        let _permit = permit;
45                        let event = Event::new(&actions::EVENTS[0], request.clone());
46                        let result = tokio::time::timeout(
47                            Duration::from_secs(10),
48                            crate::llm::action_helper::call_llm(
49                                &c.llm_client,
50                                &c.state,
51                                sid,
52                                None,
53                                &event,
54                                &actions::BacnetProtocol,
55                            ),
56                        )
57                        .await;
58                        let tag = match &result {
59                            Err(_) | Ok(Err(_)) => "decision=fail_closed_llm_error",
60                            Ok(Ok(r)) if r.protocol_results.is_empty() => "decision=model_silent",
61                            _ => "decision=model_answer",
62                        };
63                        crate::logging::emit::Log::new(Some(&c.status_tx))
64                            .debug(format!("BACnet/IP {tag}"));
65                        let answer =
66                            result
67                                .as_ref()
68                                .ok()
69                                .and_then(|r| r.as_ref().ok())
70                                .and_then(|r| {
71                                    r.protocol_results.iter().find_map(|a| {
72                                        if let ActionResult::Custom { data, .. } = a {
73                                            Some(data)
74                                        } else {
75                                            None
76                                        }
77                                    })
78                                });
79                        if let Ok(b) = codec::answer(&request, answer) {
80                            let _ = socket.send_to(&b, peer).await;
81                        }
82                    })
83                    .await;
84            }
85        })
86        .await;
87    Ok(addr)
88}
netget/server/bacnet/mod.rs

netget/server/bacnet/
mod.rs
