e-bike-tracker-device/src/modem.rs

use crate::command::Command;
use crate::serial::SerialIO;

use anyhow;
use std::{
    error::Error,
    io::{Read, Write},
    thread,
    time::{Duration, Instant},
    sync::mpsc::Receiver,
};

use esp_idf_hal::prelude::*;
use esp_idf_hal::serial::{self, Rx, Tx};

use embedded_hal::digital::v2::OutputPin;

use mqtt::packet::{ConnectPacket, PublishPacket, QoSWithPacketIdentifier};
use mqtt::{Encodable, TopicName};

const MAX_TCP_MANUAL_REPLY_SIZE: usize = 300;

pub type Result<T> = std::result::Result<T, ModemError>;

pub struct Modem<UART: serial::Uart> {
    serial: SerialIO<UART>,
}

#[derive(Debug)]
pub enum ModemError {
    CommandError(String),
    SetupError(String),
    SendDataError,
    ReadError,
    TimeoutError,
}

impl Error for ModemError {}

impl std::fmt::Display for ModemError {
    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
        write!(f, "{:?}", self)
    }
}

impl<UART: serial::Uart> Modem<UART> {
    pub fn new(tx: Tx<UART>, rx: Rx<UART>) -> Self {
        Self {
            serial: SerialIO::new(tx, rx),
        }
    }

    /// Initialize the modem (sim800l in this case). The initialization process sets all pins in the
    /// required state so that the modem is turned on, then resets it a couple of times (beats me) and
    /// sleeps for 3 seconds, which is enough for the modem to come online.
    ///
    /// Below is an example for sim800l pins on a LilyGo TTGO T-Call.
    ///
    /// # Examples
    ///
    /// ```
    /// let modem_pwrkey = dp.pins.gpio4.into_output().unwrap();
    /// let modem_rst = dp.pins.gpio5.into_output().unwrap();
    /// let modem_power = dp.pins.gpio23.into_output().unwrap();
    ///
    /// modem::init(modem_pwrkey, modem_rst, modem_power);
    /// ```
    pub fn init(&mut self, mut pwrkey: impl OutputPin, mut rst: impl OutputPin, mut power: impl OutputPin) -> Result<()> {
        println!("Turning SIM800L on ...");
        power.set_high().map_err(|_| ModemError::SetupError("Error setting POWER to high.".to_string()))?;
        rst.set_high().map_err(|_| ModemError::SetupError("Error setting RST to high.".to_string()))?;
        // Pull down PWRKEY for more than 1 second according to manual requirements
        pwrkey.set_high().map_err(|_| ModemError::SetupError("Error setting PWRKEY to high.".to_string()))?;
        thread::sleep(Duration::from_millis(100));
        pwrkey.set_low().map_err(|_| ModemError::SetupError("Error setting PWRKEY to low.".to_string()))?;
        thread::sleep(Duration::from_millis(1000));
        pwrkey.set_high().map_err(|_| ModemError::SetupError("Error setting PWRKEY to high.".to_string()))?;
        println!("Waiting for sim module to come online ...");
        loop {
            match self.send_command(Command::probe()) {
                Ok(_) => break,
                _ => continue,
            }
        }
        Ok(())
    }

    /// Reads the serial RX until the `contains` string is encoutered if `contains` is Some(s), if
    /// None, then the first line is returned. If a timeout is reached. The timeout is provided on
    /// input via the `timeout` argument. The first argument `contains` is checked against every
    /// line in the response.
    fn read_response(&mut self, contains: Option<String>, timeout: Duration) -> Result<String> {
        let mut response = String::new();
        let start = Instant::now();
        let match_text: String = contains.unwrap_or("\n".to_string());

        loop {
            let timeout = timeout.saturating_sub(start.elapsed());
            let line = self.serial.read_line(timeout).map_err(|_| ModemError::ReadError)?;
            print!("Read {} bytes from serial: {}", line.len(), line);
            response.push_str(&line);
            if line.contains("ERROR") || line.contains(&match_text) {
                println!("Found match {} for line {} ; exiting response reader now ...", match_text, line);
                println!("-----------------------------------------------------------");
                break Ok(response.to_string())
            }
        }
    }

    fn send_command(&mut self, cmd: Command) -> Result<String> {
        println!("-----------------------------------------------------------");
        println!("Sending {} ...", cmd.text);
        let _ = self.serial
            .send_bytes(cmd.text.as_bytes(), Some('\r' as u8))
            .map_err(|_| ModemError::SendDataError)?;
        self.read_response(cmd.contains, cmd.timeout)
    }

    fn tcp_send_data(&mut self, buf: &[u8]) -> Result<String> {
        let _ = self.serial
            .write("AT+CIPSEND\r".as_bytes())
            .map_err(|_| ModemError::SendDataError)?;

        let send_prompt: String = self.serial.rx.reset(Duration::from_millis(3000))
            .map(char::from)
            .take_while(|c| *c != '>').collect();

        if send_prompt != "\r\n" {
            println!("{:?}", send_prompt.as_bytes());
            return Err(ModemError::SendDataError);
        }
        self.serial
            .send_bytes(buf, Some(26))  // 26_u8 = Ctrl+z - to end sending data
            .map_err(|_| ModemError::SendDataError)?;
        self.read_response(Some("SEND OK".to_string()), Duration::from_millis(3000))
    }

    pub fn gprs_status(&mut self) -> Result<String> {
        self.send_command(Command::gprs_bearer_status())
    }

    pub fn gprs_attach_ap(&mut self, apn: &str, username: &str, password: &str)-> Result<()> {
        println!("init gprs ...");
        let _ = self.send_command(Command::gprs_init())?;

        println!("setting up gprs credentials for apn {}, {}:{})", apn, username, password);

        let _ = self.send_command(Command::gprs_set_apn(apn))?;
        let _ = self.send_command(Command::gprs_set_user(username))?;
        let _ = self.send_command(Command::gprs_set_pwd(password))?;

        Ok(())
    }

    pub fn gprs_connect(&mut self)-> Result<()> {
        println!("open gprs ...");
        let _ = self.send_command(Command::gprs_bearer_open())?;

        Ok(())
    }

    pub fn is_gprs_attached(&mut self)-> Result<bool> {
        let res = self.send_command(Command::is_gprs_attached())?;
        Ok(res.contains("+CGATT: 1"))
    }

    pub fn tcp_is_ssl_enabled(&mut self) -> Result<bool> {
        let res = self.send_command(Command::tcp_ssl_check())?;
        Ok(res.contains("+CIPSSL: (1)"))
    }

    pub fn tcp_ssl_disable(&mut self) -> Result<()> {
        let _ = self.send_command(Command::tcp_ssl_set(false))?;
        Ok(())
    }

    pub fn tcp_ssl_enable(&mut self) -> Result<()> {
        let _ = self.send_command(Command::tcp_ssl_set(true))?;
        Ok(())
    }

    pub fn tcp_connect(&mut self, addr: &str, port: u16) -> Result<()> {
        self.send_command(Command::tcp_connect(addr, port))?;
        Ok(())
    }

    pub fn tcp_set_quick_mode(&mut self, mode: bool) -> Result<()> {
        self.send_command(Command::tcp_set_quick_mode(mode))?;
        Ok(())
    }

    pub fn tcp_set_manual_receive(&mut self, is_manual: bool) -> Result<()> {
        self.send_command(Command::tcp_set_manual_receive(is_manual))?;
        Ok(())
    }

    pub fn tcp_send(&mut self, buf: &[u8]) -> Result<()> {
        self.tcp_send_data(buf)?;
        Ok(())
    }

    fn tcp_parse_response_size(&mut self, reply_line: &str) -> Result<usize> {
        reply_line.split(',')
            .into_iter()
            .last()
            .unwrap_or("x")
            .parse::<usize>()
            .map_err(|_| ModemError::CommandError(format!("response size should be a number, got {}", reply_line)))
    }

    pub fn tcp_receive_reply_len(&mut self) -> Result<usize> {
        let reply = self.send_command(Command::tcp_receive_reply_len())?;
        reply.lines()
            .filter(|line| line.contains("+CIPRXGET: 4"))
            .next()
            .ok_or(ModemError::CommandError("reply not found :/".to_string()))
            .map(|line| self.tcp_parse_response_size(line))
            .unwrap_or(Err(ModemError::CommandError(format!("received 0 elements from parsing"))))
    }

    pub fn tcp_receive(&mut self, buf: &mut [u8]) -> Result<usize> {
        let mut size = 0;
        loop {
            let reply = self.send_command(Command::tcp_receive(MAX_TCP_MANUAL_REPLY_SIZE))
                .map(|reply| {
                    reply
                        .split("\r\n")
                        .filter(|line| line.len() > 2 && !line.contains("+CIPRXGET: 2,"))
                        .next()
                        .map(|line| line.chars().enumerate().map(|(idx, c)| buf[size + idx] = c as u8).count())
                })?;
            match reply {
                Some(0) | None => {
                    break Ok(size)
                },
                Some(x) => {
                    size += x;
                    continue
                },
            }
        }
    }

    pub fn tcp_close_connection(&mut self) -> Result<String> {
        self.send_command(Command::tcp_close())
    }

    pub fn http_post(&mut self, url: &str, token: &str, content: &[u8]) -> Result<String> {
        let _ = self.send_command(Command::http_init());
        let _ = self.send_command(Command::http_set_cid());
        let _ = self.send_command(Command::http_set_url(url));
        let _ = self.send_command(Command::http_set_header("X-Secret", token));
        let _ = self.send_command(Command::http_set_header("X-Topic", "device-dev"));
        let _ = self.send_command(Command::http_set_content("application/json"));
        let _ = self.send_command(Command::http_set_ssl(true));
        let _ = self.send_command(Command::http_post_len(content.len(), 100000));
        let _ = self.serial.send_bytes(content, Some(26));
        let _ = self.send_command(Command::http_post());
        self.send_command(Command::http_read_response())
    }

    pub fn http_get(&mut self, url: &str) -> Result<String> {
        let _ = self.send_command(Command::http_init());
        let _ = self.send_command(Command::http_set_cid());
        let _ = self.send_command(Command::http_set_url(url));
        let _ = self.send_command(Command::http_set_redirect(true));
        let _ = self.send_command(Command::http_set_ssl(true));
        let _ = self.send_command(Command::http_get());
        self.send_command(Command::http_read_response())
    }

    pub fn http_close(&mut self) -> Result<()> {
        let _ = self.send_command(Command::http_close())?;
        Ok(())
    }

    pub fn chip_info(&mut self) -> Result<()> {
        let _ = self.send_command(Command::manufacturer_id())?;
        thread::sleep(Duration::from_millis(1000));
        let _ = self.send_command(Command::model_id())?;
        thread::sleep(Duration::from_millis(1000));
        let _ = self.send_command(Command::release_id())?;
        Ok(())
    }

    pub fn location(&mut self) -> Result<()> {
        let _ = self.send_command(Command::get_location())?;
        Ok(())
    }

    pub fn ssl_opt(&mut self) -> Result<()> {
        let _ = self.send_command(Command::ssl_opt())?;
        Ok(())
    }

    fn file_write(&mut self, buf: &[u8], path: &str, append: bool, input_time_sec: usize) -> Result<()> {
        let cmd = Command::fs_file_write(path, append, buf.len(), input_time_sec);
        let _ = self.serial
            .send_bytes(cmd.text.as_bytes(), Some('\r' as u8))
            .map_err(|_| ModemError::SendDataError)?;
        let send_prompt: String = self.serial.rx.reset(Duration::from_millis(3000))
            .map(char::from)
            .take_while(|c| *c != '>').collect();

        if send_prompt == "" {
            return Err(ModemError::SendDataError);
        }

        self.serial
            .send_bytes(buf, None)
            .map_err(|_| ModemError::SendDataError)?;
        let _ = self.read_response(Some("OK".to_string()), Duration::from_millis(3000));

        Ok(())
    }

    pub fn upload_cert(&mut self, path: &str, cert: &[u8]) -> Result<()> {
        let _ = self.send_command(Command::fs_file_create(path))?;
        let _ = self.file_write(cert, path, false, 20000)?;
        Ok(())
    }

    pub fn fs_list(&mut self, path: &str) -> Result<()> {
        let _ = self.send_command(Command::fs_list(path))?;
        Ok(())
    }

    pub fn fs_free_space(&mut self) -> Result<()> {
        let _ = self.send_command(Command::fs_free_size())?;
        Ok(())
    }

    pub fn ssl_set_client_cert(&mut self, path: &str, password: &str) -> Result<()> {
        let _ = self.send_command(Command::ssl_set_client_cert(path, password))?;
        Ok(())
    }

    pub fn ssl_set_root_cert(&mut self, path: &str, filesize: usize) -> Result<()> {
        let _ = self.send_command(Command::ssl_set_root_cert(path, filesize))?;
        Ok(())
    }

    fn mqtt_receive_reply(&mut self)  -> std::result::Result<(), anyhow::Error> {
        println!("entered receiving modem reply ...");
        let size = self.tcp_receive_reply_len()?;
        println!("receiving reply len({}) ...", size);
        let mut reply = vec![0 as u8; size];
        println!("receiving tcp reply ...");
        let _ = self.tcp_receive(&mut reply);
        println!("received tcp reply ...");
        Ok(())
    }

    fn mqtt_connect(&mut self, device_id: &str) -> std::result::Result<(), anyhow::Error> {
        let mut buf = Vec::new();
        let mut conn = ConnectPacket::new(device_id);
        conn.set_clean_session(true);
        conn.set_keep_alive(0);
        let _ = conn.encode(&mut buf)?;
        let _ = self.tcp_send(&mut buf)?;

        thread::sleep(Duration::from_millis(2000));
        drop(buf);

        let _ = self.mqtt_receive_reply()?;
        Ok(())
    }

    fn mqtt_publish(&mut self, _device_id: &str, message: &str) -> std::result::Result<(), anyhow::Error> {
        println!("entered mqtt publish ...");
        let mut buf = Vec::new();
        let packet = PublishPacket::new(
            TopicName::new(format!("bajsevi/location")).unwrap(),
            QoSWithPacketIdentifier::Level0,
            message.as_bytes(),
        );
        println!("created mqtt publish packet ...");
        let _ = packet.encode(&mut buf)?;
        println!("modem tcp send publish pakage ...");
        let _ = self.tcp_send(&mut buf)?;

        thread::sleep(Duration::from_millis(2000));
        drop(buf);

        println!("receiving modem publish reply ...");
        let _ = self.mqtt_receive_reply()?;
        Ok(())
    }
}

impl<UART: serial::Uart> std::io::Read for Modem<UART> {
    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
        self.tcp_receive(buf).map_err(|_| std::io::Error::from(std::io::ErrorKind::ConnectionAborted))
    }
}

impl<UART: serial::Uart> std::io::Write for Modem<UART> {
    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
        self.tcp_send(buf)
            .map_err(|_| std::io::Error::from(std::io::ErrorKind::ConnectionAborted))?;
        Ok(buf.len())
    }

    fn flush(&mut self) -> std::io::Result<()> {
        Ok(())
    }
}

pub enum Msg {
    Location(String),
    Movement(String),
}

pub fn main<T: Sync + Send>(
    rx: esp_idf_hal::gpio::Gpio26<T>,
    tx: esp_idf_hal::gpio::Gpio27<T>,
    uart: serial::UART1,
    pwrkey: esp_idf_hal::gpio::Gpio4<esp_idf_hal::gpio::Output>,
    rst: esp_idf_hal::gpio::Gpio5<esp_idf_hal::gpio::Output>,
    power: esp_idf_hal::gpio::Gpio23<esp_idf_hal::gpio::Output>,
    receiver: Receiver<Msg>,
) -> std::result::Result<(), anyhow::Error> {
    let serial_pins = serial::Pins {
        tx,
        rx,
        cts: None,
        rts: None,
    };

    let serial: serial::Serial<serial::UART1, _, _> = serial::Serial::new(
        uart,
        serial_pins,
        serial::config::Config::default().baudrate(Hertz(115200)),
    )?;

    let (tx, rx) = serial.split();
    let mut mdm = Modem::new(tx, rx);

    mdm.init(pwrkey, rst, power)?;

    if !mdm.is_gprs_attached()? {
        let _ = mdm.gprs_attach_ap(
            crate::config::A1_GPRS_AP.apn,
            crate::config::A1_GPRS_AP.username,
            crate::config::A1_GPRS_AP.password,
        )?;
    }

    thread::sleep(Duration::from_millis(1000));

    //println!("setting up client TLS cert");
    //let client_cert = include_bytes!("../certs/full-bin.p12");
    //let client_cert_path = "C:\\USER\\fullchain.pem";

    //let _ = mdm.upload_cert(client_cert_path, client_cert)?;
    //let _ = mdm.ssl_set_client_cert(client_cert_path, "t")?;
    //let _ = mdm.fs_list("C:\\USER\\")?;

    let mut retries = 0;
    let is_connected: bool = loop {
        if mdm.is_gprs_attached()? {
            let _ = mdm.gprs_connect()?;
            thread::sleep(Duration::from_millis(1000));
            let ip_addr = mdm.gprs_status()?;
            if ip_addr.contains("0.0.0.0") && retries < 5 {
                thread::sleep(Duration::from_millis(1000));
                retries += 1;
                continue
            } else if retries < 5 {
                break true
            } else {
                break false
            }
        }
    };

    if is_connected {
        let device_id = "c36a72df-5bd6-4f9b-995d-059433bc3267";

        let _ = mdm.tcp_set_quick_mode(false);
        let _ = mdm.tcp_set_manual_receive(true);
        let _ = mdm.tcp_connect("51.158.66.64", 1883)?;

        let _ = mdm.mqtt_connect(device_id)?;

        println!("entering queue receive loop ...");
        while let Ok(Msg::Location(msg)) = receiver.recv() {
            println!("received message {} | sending to mqtt ...", msg);
            let _ = mdm.mqtt_publish(device_id, &msg)?;
        }

        let _ = mdm.tcp_close_connection()?;
    }

    Ok(())
}