humpback-dds/src/mqtt_mux.rs

use log::info;
use nom::IResult;
use nom::combinator::{value, map, map_res, not, opt, all_consuming};
use nom::sequence::{terminated, preceded, pair, delimited, tuple};
use nom::bytes::complete::{take, tag, tag_no_case, take_while};
use nom::character::complete::digit1;
use nom::character::is_space;
use nom::branch::alt;
use nom::number::complete::{float, double};

use uom::si::f64::Frequency;
use uom::si::frequency::{hertz, kilohertz, megahertz, gigahertz};

use embedded_hal::blocking::spi::Transfer;
use core::convert::TryInto;
use crate::ClockSource as UrukulClockSource;
use crate::ClockSource::*;
use crate::Urukul;
use crate::Error;

#[derive(Debug, Clone)]
pub enum MqttTopic {
    Switch(u8),
    Attenuation(u8),
    Clock,
    ClockSource,
    ClockFrequency,
    ClockDivision,
    SystemClock(u8),
    Singletone(u8, u8),
    SingletoneFrequency(u8, u8),
    SingletoneAmplitude(u8, u8),
    SingletonePhase(u8, u8),
    Profile,
}

// Prossible change: Make this enum public to all comm protocol (if any)
// Such that Urukul accepts the enum directly
#[derive(Debug, Clone)]
pub enum MqttCommand {
    Switch(u8, bool),
    Attenuation(u8, f32),
    Clock(UrukulClockSource, f64, u8),
    ClockSource(UrukulClockSource),
    ClockFrequency(f64),
    ClockDivision(u8),
    SystemClock(u8, f64),
    Singletone(u8, u8, f64, f64, f64),
    SingletoneFrequency(u8, u8, f64),
    SingletoneAmplitude(u8, u8, f64),
    SingletonePhase(u8, u8, f64),
    Profile(u8)
}

pub struct MqttMux<SPI> {
    urukul: Urukul<SPI>
}

impl<SPI, E> MqttMux<SPI> where SPI: Transfer<u8, Error = E> {
    pub fn new(urukul: Urukul<SPI>) -> Self {
        MqttMux {
            urukul
        }
    }

    pub fn process_mqtt(&mut self, topic: &str, message: &[u8]) -> Result<(), Error<E>> {
        let (_, header) = self.parse_header(topic)
            .map_err(|_| Error::MqttTopicError)?;
        info!("{:?}", header);
        let (_, command) = self.parse_message(header, message)
            .map_err(|_| Error::MqttCommandError)?;
        info!("{:?}", command);
        self.execute(command)
    }

    fn parse_header<'a>(&mut self, topic: &'a str) -> IResult<&'a str, MqttTopic> {
        preceded(
            alt((
                tag("Urukul/Control/"),
                tag("/Urukul/Control/")
            )),
            alt((
                switch,
                attenuation,
                singletone,
                singletone_frequency,
                singletone_amplitude,
                singletone_phase,
                profile
            ))
        )(topic)
    }

    fn parse_message<'a>(&mut self, topic: MqttTopic, message: &'a [u8]) -> IResult<&'a [u8], MqttCommand> {
        match topic {
            MqttTopic::Switch(ch) => switch_message(ch, message),
            MqttTopic::Attenuation(ch) => attenuation_message(ch, message),
            MqttTopic::Clock => clock_message(message),
            MqttTopic::ClockSource => clock_source_message(message),
            MqttTopic::ClockFrequency => clock_frequency_message(message),
            MqttTopic::ClockDivision => clock_division_message(message),
            MqttTopic::SystemClock(ch) => system_clock_message(ch, message),
            MqttTopic::Singletone(ch, prof) => singletone_message(ch, prof, message),
            MqttTopic::SingletoneFrequency(ch, prof) => singletone_frequency_message(ch, prof, message),
            MqttTopic::SingletoneAmplitude(ch, prof) => singletone_amplitude_message(ch, prof, message),
            MqttTopic::SingletonePhase(ch, prof) => singletone_phase_message(ch, prof, message),
            MqttTopic::Profile => profile_message(message),
        }
    }

    fn execute(&mut self, command: MqttCommand) -> Result<(), Error<E>> {
        match command {
            MqttCommand::Switch(ch, state) => self.urukul.set_channel_switch(ch.into(), state),
            MqttCommand::Attenuation(ch, ampl) => self.urukul.set_channel_attenuation(ch, ampl),
            MqttCommand::Clock(src, freq, div) => self.urukul.set_clock(src, freq, div),
            MqttCommand::ClockSource(src) => self.urukul.set_clock_source(src),
            MqttCommand::ClockFrequency(freq) => self.urukul.set_clock_frequency(freq),
            MqttCommand::ClockDivision(div) => self.urukul.set_clock_division(div),
            MqttCommand::SystemClock(ch, freq) => self.urukul.set_channel_sys_clk(ch, freq),
            MqttCommand::Singletone(ch, prof, freq, ampl, deg) => self.urukul.set_channel_single_tone_profile(ch, prof, freq, ampl, deg),
            MqttCommand::SingletoneFrequency(ch, prof, freq) => self.urukul.set_channel_single_tone_profile_frequency(ch, prof, freq),
            MqttCommand::SingletoneAmplitude(ch, prof, ampl) => self.urukul.set_channel_single_tone_profile_amplitude(ch, prof, ampl),
            MqttCommand::SingletonePhase(ch, prof, deg) => self.urukul.set_channel_single_tone_profile_phase(ch, prof, deg),
            MqttCommand::Profile(prof) => self.urukul.set_profile(prof),
        }
    }
}

// Topic separator parser
fn topic_separator<'a>(topic: &'a str) -> IResult<&'a str, ()> {
    value((), tag("/"))(topic)
}

// Message separator parser
fn message_separator(message: &[u8]) -> IResult<&[u8], ()> {
    value(
        (),
        preceded(
            whitespace,
            preceded(
                tag("/"),
                whitespace
            )
        )
    )(message)
}

// Selection parsers
fn select_channel<'a>(topic: &'a str) -> IResult<&'a str, u8> {
    terminated(
        map_res(
            preceded(
                tag("Channel"),
                digit1
            ),
            |num: &str| u8::from_str_radix(num, 10)
        ),
        topic_separator
    )(topic)
}

fn select_profile<'a>(topic: &'a str) -> IResult<&'a str, u8> {
    terminated(
        map_res(
            preceded(
                tag("Profile"),
                digit1
            ),
            |num: &str| u8::from_str_radix(num, 10)
        ),
        topic_separator
    )(topic)
}

fn select_clock<'a>(topic: &'a str) -> IResult<&'a str, ()> {
    value(
        (),
        preceded(
            tag("Clock"),
            topic_separator
        )
    )(topic)
}

// Selection parser for Singletone
// Note: This fucntion assumes singletone is not the most specific sub-topic
fn select_singletone<'a>(topic: &'a str) -> IResult<&'a str, ()> {
    preceded(
        tag("Singletone"),
        topic_separator
    )(topic)
}

// Read whitespace
fn whitespace(message: &[u8]) -> IResult<&[u8], ()> {
    value((), take_while(is_space))(message)
}

// Reader for uom instances
fn read_frequency(message: &[u8]) -> IResult<&[u8], f64> {
    map(
        pair(
            double,
            opt(
                preceded(
                    whitespace,
                    alt((
                        value(1.0, tag_no_case("hz")),
                        value(1_000.0, tag_no_case("khz")),
                        value(1_000_000.0, tag_no_case("mhz")),
                        value(1_000_000_000.0, tag_no_case("ghz"))
                    ))
                )
            )
        ),
        |(freq, unit): (f64, Option<f64>)| {
            freq * unit.map_or(1.0, |mul| mul)
        }
    )(message)
}

// Parser for Switch Command Topic
fn switch<'a>(topic: &'a str) -> IResult<&'a str, MqttTopic> {
    all_consuming(
        map(
            terminated(
                select_channel,
                tag("Switch")
            ),
            |channel: u8| MqttTopic::Switch(channel)
        )
    )(topic)
}

// Parser for Switch Command Message
fn switch_message(channel: u8, message: &[u8]) -> IResult<&[u8], MqttCommand> {
    all_consuming(
        map(
            alt((
                value(true, tag("on")),
                value(false, tag("off"))
            )),
            |switch| MqttCommand::Switch(channel, switch)
        )
    )(message)
}

// Parser for Attenuation Command Topic
fn attenuation<'a>(topic: &'a str) -> IResult<&'a str, MqttTopic> {
    all_consuming(
        map(
            terminated(
                select_channel,
                tag("Attenuation")
            ),
            |channel: u8| MqttTopic::Attenuation(channel)
        )
    )(topic)
}

// Parser for Attenuation Command Message
fn attenuation_message(channel: u8, message: &[u8]) -> IResult<&[u8], MqttCommand> {
    all_consuming(
        map(
            terminated(
                float,
                opt(
                    preceded(
                        whitespace,
                        tag_no_case("db")
                    )
                )
            ),
            |att: f32| MqttCommand::Attenuation(channel, att)
        )
    )(message)
}

// Parser for Clock Source Command Topic
fn clock_source<'a>(topic: &'a str) -> IResult<&'a str, MqttTopic> {
    all_consuming(
        value(
            MqttTopic::ClockSource,
            preceded(
                clock_source,
                tag("Source")
            )
        )
    )(topic)
}

// Parser for Clock Source Command Message
fn clock_source_message(message: &[u8]) -> IResult<&[u8], MqttCommand> {
    all_consuming(
        alt((
            value(MqttCommand::ClockSource(UrukulClockSource::OSC), tag_no_case("OSC")),
            value(MqttCommand::ClockSource(UrukulClockSource::MMCX), tag_no_case("MMCX")),
            value(MqttCommand::ClockSource(UrukulClockSource::SMA), tag_no_case("SMA"))
        ))
    )(message)
}

// Parser for Clock Frequency Command Topic
fn clock_frequency<'a>(topic: &'a str) -> IResult<&'a str, MqttTopic> {
    all_consuming(
        value(
            MqttTopic::ClockFrequency,
            preceded(
                clock_source,
                tag("Frequency")
            )
        )
    )(topic)
}

// Parser for Clock Frequency Command Message
fn clock_frequency_message(message: &[u8]) -> IResult<&[u8], MqttCommand> {
    all_consuming(
        map(
            read_frequency,
            |freq: f64| MqttCommand::ClockFrequency(freq)
        )
    )(message)
}

// Parser for Clock Division Command Topic
fn clock_division<'a>(topic: &'a str) -> IResult<&'a str, MqttTopic> {
    all_consuming(
        value(
            MqttTopic::ClockDivision,
            preceded(
                clock_source,
                tag("Division")
            )
        )
    )(topic)
}

// Parser for Clock Division Command Message
fn clock_division_message(message: &[u8]) -> IResult<&[u8], MqttCommand> {
    all_consuming(
        map(
            digit1,
            |div: &[u8]| MqttCommand::ClockDivision(
                u8::from_str_radix(
                    core::str::from_utf8(div).unwrap(),
                    10
                ).unwrap()
            )
        )
    )(message)
}

// Parser for one-command master clock setup topic
fn clock<'a>(topic: &'a str) -> IResult<&'a str, MqttTopic> {
    all_consuming(
        value(
            MqttTopic::Clock,
            tag("Clock")
        )
    )(topic)
}

// Parser for one-command master clock setup message
fn clock_message(message: &[u8]) -> IResult<&[u8], MqttCommand> {
    all_consuming(
        map(
            delimited(
                tag("{"),
                tuple((
                    preceded(
                        whitespace,
                        preceded(
                            tag("\"source\":"),
                            preceded(
                                whitespace,
                                terminated(
                                    alt((
                                        value(UrukulClockSource::OSC, tag_no_case("OSC")),
                                        value(UrukulClockSource::MMCX, tag_no_case("MMCX")),
                                        value(UrukulClockSource::SMA, tag_no_case("SMA"))
                                    )),
                                    tag(",")
                                )
                            )
                        )
                    ),
                    preceded(
                        whitespace,
                        preceded(
                            tag("\"frequency\":"),
                            preceded(
                                whitespace,
                                terminated(
                                    read_frequency,
                                    tag(",")
                                )
                            )
                        )
                    ),
                    preceded(
                        whitespace,
                        preceded(
                            tag("\"division\":"),
                            preceded(
                                whitespace,
                                terminated(
                                    map_res(
                                        digit1,
                                        |div: &[u8]| u8::from_str_radix(core::str::from_utf8(div).unwrap(), 10)
                                    ),
                                    whitespace
                                )
                            )
                        )
                    )
                )),
                tag("}")
            ),
            |(src, freq, div): (UrukulClockSource, f64, u8)| MqttCommand::Clock(src, freq, div)
        )
    )(message)
}

// Topic parser for f_sys_clk of any channels
fn system_clock<'a>(topic: &'a str) -> IResult<&'a str, MqttTopic> {
    all_consuming(
        map(
            terminated(
                select_channel,
                tag("SystemClock")
            ),
            |channel: u8| MqttTopic::SystemClock(channel)
        )
    )(topic)
}

// Message parser for f_sys_clk of any channels
fn system_clock_message(channel: u8, message: &[u8]) -> IResult<&[u8], MqttCommand> {
    all_consuming(
        map(
            read_frequency,
            |freq: f64| MqttCommand::SystemClock(channel, freq)
        )
    )(message)
}

// Parser for Singletone frequenct Command Topic
fn singletone_frequency<'a>(topic: &'a str) -> IResult<&'a str, MqttTopic> {
    all_consuming(
        map(
            pair(
                select_channel,
                terminated(
                    select_profile,
                    preceded(
                        select_singletone,
                        tag("Frequency"),
                    )
                )
            ),
            |(channel, profile): (u8, u8)| MqttTopic::SingletoneFrequency(channel, profile)
        )
    )(topic)
}

// Parser for Singletone frequency Command Message
fn singletone_frequency_message(channel: u8, profile: u8, message: &[u8]) -> IResult<&[u8], MqttCommand> {
    all_consuming(
        map(
            read_frequency,
            |freq: f64| MqttCommand::SingletoneFrequency(channel, profile, freq)
        )
    )(message)
}

// Parser for Singletone Amplitude Command Topic
fn singletone_amplitude<'a>(topic: &'a str) -> IResult<&'a str, MqttTopic> {
    all_consuming(
        map(
            pair(
                select_channel,
                terminated(
                    select_profile,
                    preceded(
                        select_singletone,
                        tag("Amplitude"),
                    )
                )
            ),
            |(channel, profile): (u8, u8)| MqttTopic::SingletoneAmplitude(channel, profile)
        )
    )(topic)
}

// Parser for Singletone AMplitude Command Message
fn singletone_amplitude_message(channel: u8, profile: u8, message: &[u8]) -> IResult<&[u8], MqttCommand> {
    all_consuming(
        map(
            double,
            |ampl: f64| MqttCommand::SingletoneAmplitude(channel, profile, ampl)
        )
    )(message)
}

// Parser for Phase Command Topic
fn singletone_phase<'a>(topic: &'a str) -> IResult<&'a str, MqttTopic> {
    all_consuming(
        map(
            pair(
                select_channel,
                terminated(
                    select_profile,
                    preceded(
                        select_singletone,
                        tag("Phase"),
                    )
                )
            ),
            |(channel, profile): (u8, u8)| MqttTopic::SingletonePhase(channel, profile)
        )
    )(topic)
}

// Parser for Phase Command Message
fn singletone_phase_message(channel: u8, profile: u8, message: &[u8]) -> IResult<&[u8], MqttCommand> {
    all_consuming(
        map(
            terminated(
                double,
                opt(
                    preceded(
                        whitespace,
                        tag_no_case("deg")
                    )
                )
            ),
            |deg: f64| MqttCommand::SingletonePhase(channel, profile, deg)
        )
    )(message)
}

// Parser for one-command singletone profile Topic
fn singletone<'a>(topic: &'a str) -> IResult<&'a str, MqttTopic> {
    all_consuming(
        map(
            pair(
                select_channel,
                terminated(
                    select_profile,
                    tag("Singletone")
                )
            ),
            |(channel, profile): (u8, u8)| MqttTopic::Singletone(channel, profile)
        )
    )(topic)
}

// Parser for one-command singletone profile Command
// Using JSON like command structure
// Possible enhancement: further modularize parsing of all separate fields
fn singletone_message(channel: u8, profile: u8, message: &[u8]) -> IResult<&[u8], MqttCommand> {
    all_consuming(
        map(
            tuple((
                preceded(
                    tag("{"),
                    preceded(
                        whitespace,
                        preceded(
                            tag("\"frequency\":"),
                            preceded(
                                whitespace,
                                read_frequency
                            )
                        )
                    )
                ),
                preceded(
                    tag(","),
                    preceded(
                        whitespace,
                        preceded(
                            tag("\"amplitude\":"),
                            preceded(
                                whitespace,
                                double
                            )
                        )
                    )
                ),
                preceded(
                    tag(","),
                    preceded(
                        whitespace,
                        preceded(
                            tag("\"phase\":"),
                            preceded(
                                whitespace,
                                terminated(
                                    double,
                                    preceded(
                                        opt(
                                            preceded(
                                                whitespace,
                                                tag_no_case("deg")
                                            )
                                        ),
                                        preceded(
                                            whitespace,
                                            tag("}")
                                        )
                                    )
                                )
                            )
                        )
                    )
                )
            )),
            |(freq, ampl, phase): (f64, f64, f64)| MqttCommand::Singletone(channel, profile, freq, ampl, phase)
        )
    )(message)
}

fn profile<'a>(topic: &'a str) -> IResult<&'a str, MqttTopic> {
    all_consuming(
        value(
            MqttTopic::Profile,
            tag("Profile")
        )
    )(topic)
}

fn profile_message(message: &[u8]) -> IResult<&[u8], MqttCommand> {
    all_consuming(
        map(
            digit1,
            |num: &[u8]| {
                MqttCommand::Profile(
                    u8::from_str_radix(core::str::from_utf8(num).unwrap(), 10).unwrap()
                )
            }
        )
    )(message)
}