radomctld/daemon/src/main.rs

use anyhow::anyhow;
use anyhow::Result;
use axum::{extract::State, routing::get, Json, Router};
use clap::Parser;
use log::{debug, error, info};
use serde_json::{json, Value};
use tokio::{
    self,
    io::{AsyncBufReadExt, AsyncWriteExt, BufStream},
    net::{TcpListener, TcpStream},
    sync::{mpsc, watch},
    task::JoinSet,
};
use tokio_serial;
use tower_http::{services::ServeFile, trace::TraceLayer};

use radomctld::{
    logger::setup_logger,
    rotctlprotocol::{parse_command, Command},
    rotor::control_rotor,
};

async fn process_socket(
    socket: TcpStream,
    cmd_tx: mpsc::Sender<Command>,
    pos_rx: watch::Receiver<(f32, f32)>,
) {
    let mut stream = BufStream::new(socket);

    let mut line = String::new();

    loop {
        if let Ok(n) = stream.read_line(&mut line).await {
            if n == 0 {
                return;
            }

            debug!("Received: {}", line.strip_suffix("\n").unwrap());

            match parse_command(&line) {
                Ok(cmd) => match cmd {
                    Command::GetPos => {
                        let (az, el) = pos_rx.borrow().clone();

                        stream
                            .write_all(format!("{}\n{}\n", az, el).as_bytes())
                            .await
                            .unwrap();
                        stream.flush().await.unwrap();
                    }
                    Command::Exit => {
                        stream.write_all("RPRT 0\n".as_bytes()).await.unwrap();
                        stream.flush().await.unwrap();
                        return;
                    }
                    cmd => {
                        cmd_tx.send(cmd).await.unwrap();
                        stream.write_all("RPRT 0\n".as_bytes()).await.unwrap();
                        stream.flush().await.unwrap();
                    }
                },
                Err(msg) => {
                    error!("Unable to parse input:\n{}", msg);
                    stream.write_all("RPRT 6\n".as_bytes()).await.unwrap();
                    stream.flush().await.unwrap();
                }
            }

            line.clear();
        } else {
            return;
        }
    }
}

#[derive(Clone)]
struct AxumAppState {
    pos_rx: watch::Receiver<(f32, f32)>,
}

#[derive(Parser)]
struct Cli {
    /// The usb serial number of the radom-controller
    #[arg(short, long)]
    serialnumber: String,

    /// Listen address for the webserver
    #[arg(short, long, default_value = "0.0.0.0:8000")]
    web_listen_address: String,

    /// Listen address for rotctl
    #[arg(short, long, default_value = "0.0.0.0:1337")]
    rotctl_listen_address: String,
}

#[tokio::main]
async fn main() -> Result<()> {
    setup_logger()?;

    let args = Cli::parse();

    let ports = tokio_serial::available_ports().unwrap_or(Vec::<serialport::SerialPortInfo>::new());

    let mut radom_port: Option<String> = None;
    for port in ports {
        match port.port_type {
            serialport::SerialPortType::UsbPort(usb_port_info) => {
                match usb_port_info.serial_number {
                    Some(serial) => {
                        debug!("Found a serial port with: {}", serial);
                        if serial == args.serialnumber {
                            radom_port = Some(port.port_name.to_owned());
                            info!("Found radom-controller as {}", port.port_name)
                        }
                    }
                    None => continue,
                }
            }
            _ => continue,
        }
    }

    let radom_port = match radom_port {
        Some(port) => port,
        _ => {
            return Err(anyhow!("No matching port found."));
        }
    };

    let (cmd_tx, cmd_rx) = mpsc::channel::<Command>(16);
    let (pos_tx, pos_rx) = watch::channel::<(f32, f32)>((0.0, 0.0));

    let mut tasks = JoinSet::new();

    tasks.spawn(async move { control_rotor(cmd_rx, pos_tx, radom_port).await });

    let state = AxumAppState {
        pos_rx: pos_rx.clone(),
    };

    tasks.spawn(async move {
        let app = Router::new()
            .route_service("/", ServeFile::new("assets/index.html"))
            .route("/state", get(get_state))
            .with_state(state)
            .layer(TraceLayer::new_for_http());

        let listener = tokio::net::TcpListener::bind(args.web_listen_address).await?;
        axum::serve(listener, app).await?;

        Ok(())
    });

    tasks.spawn(async move {
        let listener = TcpListener::bind(args.rotctl_listen_address).await?;

        loop {
            let (socket, _) = listener.accept().await?;

            let cmd_tx = cmd_tx.clone();
            let pos_rx = pos_rx.clone();
            tokio::spawn(async move {
                process_socket(socket, cmd_tx, pos_rx).await;
            });
        }
    });

    while let Some(res) = tasks.join_next().await {
        res.unwrap();
    }

    Ok(())
}

async fn get_state(State(state): State<AxumAppState>) -> Json<Value> {
    let (az, el) = state.pos_rx.borrow().clone();

    Json(json!({
        "position" : {
            "az": az,
            "el": el
        }
    }))
}