laatzen/Common/Hardware/Common.Hardware.SIRT/SIRTConnection.cs

namespace Common.Hardware.SIRT
{
    using Common.Hardware.Ports;

    using System;
    using System.Collections.Concurrent;
    using System.Collections.Generic;
    using System.Linq;
    using System.Threading;
    using System.Threading.Tasks;

    using Parity = System.IO.Ports.Parity;
    using StopBits = System.IO.Ports.StopBits;

    public class SIRTConnection
    {
        private readonly ConcurrentDictionary<uint, ConcurrentQueue<SIRTTask>> tasks;
        private readonly SerialPort serialPort;

        private CancellationTokenSource cancellationTokenSource;
        private CancellationToken cancellationToken;
        private Task readingTask;
        private Task processingTask;
        private Task activationTask;
        private AnswerTask sirtTask;

        public SIRTConnection(string comPort)
        {
            this.serialPort = new SerialPort(comPort, 115200, Parity.None, 8, StopBits.One);
            this.tasks = new ConcurrentDictionary<uint, ConcurrentQueue<SIRTTask>>();
        }

        public event Action Activated;
        public event Action Connected;
        public event Action Disconnected;
        public event Action<string> Log;
        public event Action<SIRTMessage> MessageReceived;

        public string COMPort => this.serialPort.COMPort;

        public string HexId { get; protected set; }

        public int Frequency { get; protected set; }

        public void Close()
        {
            this.serialPort.Close();

            this.HexId = null;
            this.Frequency = 0;
            this.serialPort.Connected -= this.SerialPort_Connected;
            this.serialPort.Disconnected -= this.SerialPort_Disconnected;
            this.serialPort.Log -= this.OnError;
            this.Activated -= this.StartProcessingTask;
            this.MessageReceived -= this.SIRTConnection_MessageReceived;
        }

        public void Open()
        {
            if (!this.serialPort.IsOpen)
            {
                this.serialPort.Connected += this.SerialPort_Connected;
                this.serialPort.Disconnected += this.SerialPort_Disconnected;
                this.serialPort.Log += this.OnError;
                this.Activated += this.StartProcessingTask;
                this.MessageReceived += this.SIRTConnection_MessageReceived;

                this.serialPort.Open();
            }
        }

        public void DequeueTasks(uint address)
        {
            if (this.tasks.TryRemove(address, out var tasksQueue))
            {
                while (tasksQueue.TryDequeue(out var _))
                {

                }
            }
        }

        public void EnqueueTask(SIRTTask task)
        {
            if (this.Frequency == task.Frequency)
            {
                task.Cancelled += this.Task_Dismiss;
                task.Completed += this.Task_Dismiss;
                task.Timeouted += this.Task_Timeouted;

                this.tasks
                    .GetOrAdd(task.Address, new ConcurrentQueue<SIRTTask>())
                    .Enqueue(task);
            }
        }

        public void Remove()
        {
            this.Close();
            this.serialPort.Remove();
        }

        private void ActivateSIRT()
        {
            var awaiter = new ManualResetEventSlim();

            bool ActivateSIRT()
            {
                this.sirtTask = SIRTTask.ActivateSIRT;
                this.sirtTask.Cancelled += _1 => awaiter.Set();
                this.sirtTask.Completed += _2 => awaiter.Set();

                this.sirtTask.SetRunning();
                this.serialPort.Write(this.sirtTask);
                awaiter.Wait(this.sirtTask.Timeout);
                awaiter.Reset();

                return this.sirtTask.State == SIRTTaskState.Completed;
            }

            bool IdentifySIRT(out byte[] data)
            {
                data = default(byte[]);

                this.sirtTask = SIRTTask.IdentifySIRT;
                this.sirtTask.Cancelled += _1 => awaiter.Set();
                this.sirtTask.Completed += _2 => awaiter.Set();

                this.sirtTask.SetRunning();
                this.serialPort.Write(this.sirtTask);
                awaiter.Wait(this.sirtTask.Timeout);
                awaiter.Reset();

                data = this.sirtTask.Last;

                return this.sirtTask.State == SIRTTaskState.Completed && data?.Length == 8;
            }

            void OnMessageReceived(SIRTMessage message)
            {
                if (message.Address == this.sirtTask?.Address)
                {
                    this.sirtTask.BeginReceiveMessage(message);
                }
            }

            this.MessageReceived += OnMessageReceived;

            if (ActivateSIRT())
            {
                if (IdentifySIRT(out var data))
                {
                    this.HexId = BitConverter.ToString(data);
                    this.Frequency = this.HexId[10] == '0' ? 868 : 433;

                    this.Activated?.Invoke();
                }
                else
                {
                    this.Log?.Invoke($"SIRT on {this.COMPort} not identified.");
                }
            }
            else
            {
                this.Log?.Invoke($"SIRT on {this.COMPort} not activated.");
            }

            this.MessageReceived -= OnMessageReceived;
        }

        private Boolean CanSend(uint address)
        {
            var awaiter = new ManualResetEventSlim();
            var task = SIRTTask.ReadPamPool;

            void messageReceived(SIRTMessage message)
            {
                if (message.Address == task.Address)
                {
                    task.BeginReceiveMessage(message);
                }
            }

            this.MessageReceived += messageReceived;
            task.Cancelled += _1 => awaiter.Set();
            task.Completed += _2 => awaiter.Set();

            task.SetRunning();
            this.serialPort.Write(task);
            awaiter.Wait(task.Timeout);
            awaiter.Reset();
            this.MessageReceived -= messageReceived;

            var payload = task.Last;

            if (task.State != SIRTTaskState.Completed || payload?.Length != 20)
            {
                return false;
            }

            var canSend = true;
            var _address = 0U;

            for (var i = 0; i < 20; i += 4)
            {
                _address = payload.ToUInt32BE(i);

                if (_address == address)
                {
                    canSend = false;

                    break;
                }
            }

            return canSend;
        }

        private void OnError(string e)
            => this.Log?.Invoke(e);

        private void SerialPort_Connected()
        {
            this.cancellationTokenSource = new CancellationTokenSource();
            this.cancellationToken = this.cancellationTokenSource.Token;

            this.readingTask = Task.Run((Action)this.StartReadingMessages, this.cancellationToken);
            this.activationTask = Task.Run((Action)this.ActivateSIRT, this.cancellationToken);

            this.Connected?.Invoke();
        }

        private void SerialPort_Disconnected()
        {
            try
            {
                this.cancellationTokenSource?.Cancel();
            }
            catch (Exception e)
            {
                this.OnError(e.ToString());
            }

            this.Disconnected?.Invoke();
        }

        private void SIRTConnection_MessageReceived(SIRTMessage message)
        {
            if (this.tasks.TryGetValue(message.Address, out var tasksQueue))
            {
                if (tasksQueue.TryPeek(out var task))
                {
                    task.BeginReceiveMessage(message);
                }
            }

            this.Log?.Invoke(message.ToString());
        }

        private void StartProcessingTask()
        {
            this.processingTask = Task.Factory.StartNew(async state =>
            {
                var connection = state as SIRTConnection;

                while (!connection.cancellationToken.IsCancellationRequested)
                {
                    foreach (var tasksQueue in connection.tasks.Values.ToArray())
                    {
                        if (tasksQueue.TryPeek(out var task) && task.State == SIRTTaskState.Pending)
                        {
                            // if (this.CanSend(task.Address))
                            {
                                task.SetRunning();
                                connection.serialPort.Write(task);
                            }
                        }

                        await Task.Delay(100);
                    }
                }
            
            }, this, this.cancellationToken);
        }

        private void StartReadingMessages()
        {
            var buffer = new List<byte>();
            int bufferLength;
            int startIX;
            int lengthIX;
            int dataLength;
            int stopIX;
            int messageLength;
            byte[] message;

            foreach (var bytes in this.serialPort.ReadBytes())
            {
                if (bytes?.Length > 0)
                {
                    buffer.AddRange(bytes);
                }

                bufferLength = buffer.Count;

                while (bufferLength >= 13)
                {
                    startIX = buffer.IndexOf(SIRTMessage.SIRT2PC);

                    if (startIX < 0)
                    {
                        buffer.Clear();

                        break;
                    }

                    lengthIX = startIX + 8;

                    if (lengthIX >= bufferLength)
                    {
                        break;
                    }

                    dataLength = buffer[lengthIX];
                    stopIX = lengthIX + dataLength + 3;

                    if (stopIX >= bufferLength)
                    {
                        break;
                    }
                    else if (buffer[stopIX] != SIRTMessage.STOP)
                    {
                        buffer.RemoveRange(0, stopIX + 1);
                    }
                    else
                    {
                        messageLength = stopIX - startIX + 1;
                        message = new byte[messageLength];

                        buffer.CopyTo(startIX, message, 0, messageLength);
                        buffer.RemoveRange(0, startIX + messageLength);

                        this.MessageReceived?.Invoke(new SIRTMessage(message));
                    }

                    bufferLength = buffer.Count;
                }
            }
        }

        private void Task_Dismiss(SIRTTask task)
        {
            task.Cancelled -= this.Task_Dismiss;
            task.Completed -= this.Task_Dismiss;
            task.Timeouted -= this.Task_Timeouted;

            this.tasks
                .GetOrAdd(task.Address, new ConcurrentQueue<SIRTTask>())
                .TryDequeue(out task);
        }

        private void Task_Timeouted(SIRTTask task)
        {
            //var awaiter = new ManualResetEventSlim();
            //var _task = SIRTTask.DeleteFromPamPool(task.Address);

            //void messageReceived(SIRTMessage message)
            //{
            //    if (message.Address == task.Address)
            //    {
            //        _task.BeginReceiveMessage(message);
            //    }
            //}

            //this.MessageReceived += messageReceived;
            //_task.Cancelled += _1 => awaiter.Set();
            //_task.Completed += _2 => awaiter.Set();

            //_task.SetRunning();
            //this.serialPort.Write(_task);
            //awaiter.Wait(_task.Timeout);
            //awaiter.Reset();
            //this.MessageReceived -= messageReceived;
        }
    }
}