c# - recibir - puertos seriales arduino mega

¿Cómo usar las extensiones reactivas para analizar una secuencia de caracteres desde un puerto serie?

Question

Necesito analizar un flujo de datos en serie procedentes de un instrumento de prueba, y esta parece ser una excelente aplicación para las extensiones reactivas.

El protocolo es muy simple ... cada "paquete" es una sola letra seguida de dígitos numéricos. El número de dígitos numéricos se fija para cada tipo de paquete, pero puede variar entre los tipos de paquetes. p.ej

... A1234B123456C12 ...

Estoy tratando de dividir esto en un Observable of Strings, por ejemplo

"A1234" "B123456" "C12" ...

Pensé que esto sería simple, pero no veo la manera obvia de abordar esto (tengo algo de experiencia con LINQ, pero soy nuevo en Rx).

Aquí está el código que tengo hasta ahora, que produce Observable de caracteres desde el evento SerialDataReceived del puerto serie.

        var serialData = Observable
                            .FromEventPattern<SerialDataReceivedEventArgs>(SerialPortMain, "DataReceived")
                            .SelectMany(_ => 
                            {
                                int dataLength = SerialPortMain.BytesToRead;
                                byte[] data = new byte[dataLength];
                                int nbrDataRead = SerialPortMain.Read(data, 0, dataLength);

                                if (nbrDataRead == 0)
                                    return  new char[0]; 

                                 var chars = Encoding.ASCII.GetChars(data);

                                 return chars; 
                            });

¿Cómo puedo transformar serialData en Observable of String, donde cada cadena es un paquete?

Answer 1

Aquí está mi opinión sobre el problema.

static IObservable<string> Packets(IObservable<char> source)
{
    return Observable.Create<string>(observer =>
    {
        var packet = new List<char>();
        Action emitPacket = () =>
        {
            if (packet.Count > 0)
            {
                observer.OnNext(new string(packet.ToArray()));
                packet.Clear();
            }
        };
        return source.Subscribe(
            c =>
            {
                if (char.IsLetter(c))
                {
                    emitPacket();
                }
                packet.Add(c);
            },
            observer.OnError,
            () =>
            {
                emitPacket();
                observer.OnCompleted();
            });
    });
}

Si la entrada son los caracteres A425B90C2DX812 la salida es A425 , B90 , C2 , D , X812 .

Tenga en cuenta que no necesita especificar longitudes de paquetes o tipos de paquetes (letras de inicio) por adelantado.

Puede implementar el mismo método utilizando un método de extensión de propósito más general:

static IObservable<IList<T>> GroupSequential<T>(
    this IObservable<T> source, Predicate<T> isFirst)
{
    return Observable.Create<T>(observer =>
    {
        var group = new List<T>();
        Action emitGroup = () =>
        {
            if (group.Count > 0)
            {
                observer.OnNext(group.ToList());
                group.Clear();
            }
        };
        return source.Subscribe(
            item =>
            {
                if (isFirst(item))
                {
                    emitGroup();
                }
                group.Add(item);
            },
            observer.OnError,
            () =>
            {
                emitGroup();
                observer.OnCompleted();
            });
    });
}

La implementación de Packets es simplemente:

static IObservable<string> Packets(IObservable<char> source)
{
    return source
        .GroupSequential(char.IsLetter)
        .Select(x => new string(x.ToArray()));
}

Answer 2

Aquí hay un método un poco más corto, en el mismo estilo que la primera solución de James , con un método de ayuda similar:

public static bool IsCompletePacket(string s)
{
    switch (s[0])
    {
        case ''A'':
            return s.Length == 5;
        case ''B'':
            return s.Length == 6;
        case ''C'':
            return s.Length == 7;
        default:
            throw new ArgumentException("Packet must begin with a letter");
    }
}

El código es entonces:

var packets = chars
    .Scan(string.Empty, (prev, cur) => char.IsLetter(cur) ? cur.ToString() : prev + cur)
    .Where(IsCompletePacket);

La parte de Scan construye cadenas que terminan en una letra, por ejemplo:

A
A1
A12
A123
...

The Where then simplemente selecciona aquellos que tienen la longitud correcta. Básicamente, simplemente elimina la Tupla de James y utiliza la longitud de la cadena en su lugar.

Answer 3

Sorprendentemente complicado! He resuelto esto de varias maneras:

// helper method to get the packet length
public int GetPacketLength(char c)
{
    switch(c)
    {
        case ''A'':
            return 5;
        case ''B'':
            return 6;
        case ''C'':
            return 7;
        default:
            throw new Exception("Unknown packet code");
    }
}

Entonces podemos hacer esto:

// chars is a test IObservable<char> 
string[] messages = { "A1234", "B12345", "C123456" };
var serialPort = Enumerable.Range(1, 10).ToObservable();
var chars = serialPort.SelectMany((_, i) => messages[i % 3]);

var packets = chars.Scan(
    Tuple.Create(string.Empty, -1),
    (acc, c) =>
        Char.IsLetter(c)
            ? Tuple.Create(c.ToString(), GetPacketLength(c) - 1)
            : Tuple.Create(acc.Item1 + c, acc.Item2 - 1))
    .Where(acc => acc.Item2 == 0)
    .Select(acc => acc.Item1)
    .Subscribe(Console.WriteLine);

Lo que hace es esto:

• El Scan construye cada paquete y lo empareja con el número de caracteres restantes en el paquete: ej. ("A", 4) ("A1", 3) ("A12", 2) ("A123", 1) (" A1234 ", 0) (" B ", 5) ...
• Entonces sabemos que los pares con 0 caracteres restantes son los que necesitamos, entonces usamos Where filtrar el resto y Select el resultado del par

Alternativa

Aquí hay otro enfoque, en un estilo menos funcional. Me gusta el código anterior desde un punto de vista de estilo, pero el código siguiente es un poco más eficiente con la memoria, en el improbable caso de que vaya a hacer una diferencia.

public static class ObservableExtensions
{
    private const int MaxPacketLength = 7;
    private static Dictionary<char, int> PacketLengthTable =
        new Dictionary<char, int> { {''A'', 5}, {''B'', 6}, {''C'', 7 } };

    public static IObservable<string> GetPackets(this IObservable<char> source)
    {
        return Observable.Create<string>(o =>
        {
            var currentPacketLength = 0;
            var buffer = new char[MaxPacketLength];
            var index = -1;
            return source.Subscribe(
                c => {
                    if (Char.IsLetter(c))
                    {
                        currentPacketLength = PacketLengthTable[c];
                        buffer[0] = c;
                        index = 0;
                    }
                    else if(index >= 0)
                    {
                        index++;
                        buffer[index] = c;
                    }
                    if (index == currentPacketLength - 1)
                    {
                        o.OnNext(new string(buffer,0, currentPacketLength));
                        index = -1;
                    }
                },
                o.OnError,
                o.OnCompleted);
        });
    }
}

Y se puede usar así:

// chars is a test IObservable<char> 
string[] messages = { "A1234", "B12345", "C123456" };
var serialPort = Enumerable.Range(1, 10).ToObservable();
var chars = serialPort.SelectMany((_, i) => messages[i % 3]);

var packets = chars.GetPackets().Subscribe(Console.WriteLine);