Ncap.cs

using System;
using System.Collections.Generic;
using System.Text;
using System.Threading.Tasks;
using Waher.Networking.MQTT;
using Waher.Runtime.Inventory;
using Waher.Security;
using Waher.Things.Ieee1451.Ieee1451_0;
using Waher.Things.Ieee1451.Ieee1451_0.Messages;
using Waher.Things.Metering;
using Waher.Things.Mqtt;
using Waher.Things.Mqtt.Model;
using Waher.Things.Mqtt.Model.Encapsulations;
using Waher.Things.SensorData;
 
namespace Waher.Things.Ieee1451.Ieee1451_1_6
{
    public abstract class Ncap : MqttData
    {
        private byte[] value;
        private bool firstMessage = true;
 
        public Ncap()
            : base()
        {
        }
 
        public Ncap(MqttTopic Topic, byte[] Value)
            : base(Topic)
        {
            this.value = Value;
        }
 
        public async Task<DataProcessingResult> DataReported(MqttTopic Topic, MqttContent Content, byte[] Data)
        {
            Message Message = await Ieee1451Parser.TryParseMessage(Data, Topic is null ? null : Content.CommunicationLayer);
            if (Message is null)
                return this.firstMessage ? DataProcessingResult.Incompatible : DataProcessingResult.Processed;
 
            this.firstMessage = false;
            this.value = Data;
            this.Timestamp = DateTime.UtcNow;
            this.QoS = Content.Header.QualityOfService;
            this.Retain = Content.Header.Retain;
 
            if (Topic is null)
                return DataProcessingResult.Processed;
 
            return await MessageReceived(this, Topic, Message);
        }
 
        public static async Task<DataProcessingResult> MessageReceived(MqttData This, MqttTopic Topic, Message Message)
        {
            DataProcessingResult Result = DataProcessingResult.Processed;
 
            try
            {
                switch (Message.MessageType)
                {
                    case MessageType.Reply:
                        Result = await ProcessReply(This, Topic, Message);
                        break;
 
                    case MessageType.Command:
                        await ProcessRequest(This, Message);
                        break;
 
                    case MessageType.Announcement:
                    case MessageType.Notification:
                    case MessageType.Callback:
                    default:
                        break;
                }
 
                await Message.LogInformationToSniffer();
            }
            catch (Exception ex)
            {
                await LogErrorAsync(This, string.Empty, ex.Message);
            }
 
            return Result;
        }
 
        internal static async Task<DataProcessingResult> ProcessReply(MqttData This, MqttTopic Topic, Message Message)
        {
            MqttTopic SubTopic;
            bool ContainsMomentary = false;
 
            if (Message is TransducerAccessMessage TransducerAccessMessage)
            {
                ThingReference Ref = new ThingReference(This.Topic.Node);
                if (TransducerAccessMessage.TryParseTransducerData(Ref, null, null, out ushort ErrorCode, out TransducerData Data))
                {
                    await RemoveErrorAsync(This, "TransducerResponseError");
 
                    SubTopic = await This.Topic.Broker.GetTopic(Data.ChannelInfo.GetTopic(This.Topic.FullTopic), true, false);
 
                    if (ErrorCode == 0)
                    {
                        await SubTopic.Node.RemoveErrorAsync("TranducerError");
                        ContainsMomentary = true;
                    }
                    else
                        await SubTopic.Node.LogErrorAsync("TranducerError", "Transducer error: " + ErrorCode.ToString("X4"));
                }
                else
                {
                    await LogErrorAsync(This, "TransducerResponseError", "Unable to parse Transducer response.");
                    return DataProcessingResult.Processed;
                }
            }
            else if (Message is TedsAccessMessage TedsAccessMessage)
            {
                (ushort ErrorCode, Teds Teds) = await TedsAccessMessage.TryParseTeds(true);
                if (!(Teds is null))
                {
                    await RemoveErrorAsync(This, "TedsResponseError");
 
                    SubTopic = await This.Topic.Broker.GetTopic(Teds.ChannelInfo.GetTopic(This.Topic.FullTopic), true, false);
 
                    if (ErrorCode == 0)
                        await SubTopic.Node.RemoveErrorAsync("TedsError");
                    else
                        await SubTopic.Node.LogErrorAsync("TedsError", "TEDS error: " + ErrorCode.ToString("X4"));
                }
                else
                {
                    await LogErrorAsync(This, "TedsResponseError", "Unable to parse TEDS response.");
                    return DataProcessingResult.Processed;
                }
            }
            else if (Message is DiscoveryMessage DiscoveryMessage)
            {
                if (DiscoveryMessage.TryParseMessage(out ushort ErrorCode, out DiscoveryData Data))
                {
                    await RemoveErrorAsync(This, "DiscoveryResponseError");
 
                    string TopicString;
                    bool Created;
 
                    switch (DiscoveryMessage.DiscoveryService)
                    {
                        case DiscoveryService.NCAPDiscovery:
                            if (Data is DiscoveryDataEntity NcapEntity)
                            {
                                TopicString = NcapEntity.Channel.GetTopic(This.Topic.FullTopic);
 
                                SubTopic = await This.Topic.Broker.GetTopic(TopicString, false, false);
                                if (Created = SubTopic is null)
                                    SubTopic = await This.Topic.Broker.GetTopic(TopicString, true, false);
 
                                if (SubTopic.Node is MqttNcapTopicNode TopicNode)
                                    await TopicNode.NameReceived(NcapEntity.Name);
 
                                if (ErrorCode == 0)
                                    await SubTopic.Node.RemoveErrorAsync("DiscoveryError");
                                else
                                    await SubTopic.Node.LogErrorAsync("DiscoveryError", "Discovery error: " + ErrorCode.ToString("X4"));
 
                                if (Created)
                                {
                                    MqttBroker Broker = Topic.Broker;
                                    MqttBrokerNode BrokerNode = Broker.Node;
                                    StringBuilder ToSniffer = BrokerNode.HasSniffers ? new StringBuilder() : null;
                                    byte[] Request = DiscoveryMessage.SerializeRequest(NcapEntity.Channel.NcapId, ToSniffer);
 
                                    if (!(ToSniffer is null))
                                        await BrokerNode.Information(ToSniffer.ToString());
 
                                    await Broker.Publish(This.Topic.FullTopic, MqttQualityOfService.AtLeastOnce, false, Request);
                                }
                            }
                            break;
 
                        case DiscoveryService.NCAPTIMDiscovery:
                            if (Data is DiscoveryDataEntities TimEntities)
                            {
                                int i, c = Math.Min(TimEntities.Names.Length, TimEntities.Identities.Length);
 
                                for (i = 0; i < c; i++)
                                {
                                    ChannelAddress TimAddress = new ChannelAddress()
                                    {
                                        ApplicationId = TimEntities.Channel.ApplicationId,
                                        NcapId = TimEntities.Channel.NcapId,
                                        TimId = TimEntities.Identities[i],
                                        ChannelId = 0
                                    };
 
                                    TopicString = TimAddress.GetTopic(This.Topic.FullTopic);
                                    SubTopic = await This.Topic.Broker.GetTopic(TopicString, false, false);
                                    if (Created = SubTopic is null)
                                        SubTopic = await This.Topic.Broker.GetTopic(TopicString, true, false);
 
                                    if (SubTopic.Node is MqttNcapTopicNode TopicNode)
                                        await TopicNode.NameReceived(TimEntities.Names[i]);
 
                                    if (ErrorCode == 0)
                                        await SubTopic.Node.RemoveErrorAsync("DiscoveryError");
                                    else
                                        await SubTopic.Node.LogErrorAsync("DiscoveryError", "Discovery error: " + ErrorCode.ToString("X4"));
 
                                    if (Created)
                                    {
                                        MqttBroker Broker = Topic.Broker;
                                        MqttBrokerNode BrokerNode = Broker.Node;
                                        StringBuilder ToSniffer = BrokerNode.HasSniffers ? new StringBuilder() : null;
                                        byte[] Request = DiscoveryMessage.SerializeRequest(TimEntities.Channel.NcapId, TimEntities.Identities[i], ToSniffer);
 
                                        if (!(ToSniffer is null))
                                            await BrokerNode.Information(ToSniffer.ToString());
 
                                        await Broker.Publish(This.Topic.FullTopic, MqttQualityOfService.AtLeastOnce, false, Request);
                                    }
                                }
                            }
                            break;
 
                        case DiscoveryService.NCAPTIMTransducerDiscovery:
                            if (Data is DiscoveryDataChannels Channels)
                            {
                                int i, c = Math.Min(Channels.Names.Length, Channels.Channels.Length);
 
                                for (i = 0; i < c; i++)
                                {
                                    ChannelAddress TimAddress = new ChannelAddress()
                                    {
                                        ApplicationId = Channels.Channel.ApplicationId,
                                        NcapId = Channels.Channel.NcapId,
                                        TimId = Channels.Channel.TimId,
                                        ChannelId = Channels.Channels[i],
                                    };
 
                                    TopicString = TimAddress.GetTopic(This.Topic.FullTopic);
                                    SubTopic = await This.Topic.Broker.GetTopic(TopicString, true, false);
                                    if (SubTopic.Node is MqttNcapTopicNode TopicNode)
                                        await TopicNode.NameReceived(Channels.Names[i]);
 
                                    if (ErrorCode == 0)
                                        await SubTopic.Node.RemoveErrorAsync("DiscoveryError");
                                    else
                                        await SubTopic.Node.LogErrorAsync("DiscoveryError", "Discovery error: " + ErrorCode.ToString("X4"));
                                }
                            }
                            break;
                    }
                }
                else
                    await LogErrorAsync(This, "DiscoveryResponseError", "Unable to parse Discovery response.");
 
                return DataProcessingResult.Processed;
            }
            else
                return DataProcessingResult.Processed;
 
            if (!(SubTopic?.Node is MqttNcapTopicNode NcapTopicNode))
                return DataProcessingResult.Processed;
 
            if (!NcapTopicNode.ResponseReceived(Topic, Message) && ContainsMomentary)
            {
                // TODO: Report new momentary values on node.
            }
 
            return DataProcessingResult.Processed;
        }
 
        private static async Task ProcessRequest(MqttData This, Message Message)
        {
            MqttTopic SubTopic;
 
            if (Message is TransducerAccessMessage TransducerAccessMessage)
            {
                switch (TransducerAccessMessage.TransducerAccessService)
                {
                    case TransducerAccessService.SyncReadTransducerSampleDataFromAChannelOfATIM:
                        if (TransducerAccessMessage.TryParseRequest(out ChannelAddress Address,
                            out SamplingMode SamplingMode, out double TimeoutSeconds))
                        {
                            await RemoveErrorAsync(This, "TransducerRequestError");
 
                            StringBuilder sb = new StringBuilder();
 
                            sb.Append(This.Topic.FullTopic);
                            sb.Append('/');
                            sb.Append(Hashes.BinaryToString(Address.NcapId));
 
                            if (!MessageSwitch.IsZero(Address.TimId))
                            {
                                sb.Append('/');
                                sb.Append(Hashes.BinaryToString(Address.TimId));
 
                                if (Address.ChannelId != 0)
                                {
                                    sb.Append('/');
                                    sb.Append(Address.ChannelId.ToString());
                                }
                            }
 
                            SubTopic = await This.Topic.Broker.GetTopic(sb.ToString(), true, false);
 
                            if (!(SubTopic?.Node is ITransducerNode TransducerNode))
                                return;
 
                            await TransducerNode.TransducerDataRequest(TransducerAccessMessage, SamplingMode, TimeoutSeconds);
                        }
                        else
                        {
                            await LogErrorAsync(This, "TransducerRequestError", "Unable to parse Transducer request.");
                            return;
                        }
                        break;
 
                    case TransducerAccessService.SyncReadTransducerBlockDataFromAChannelOfATIM:
                    case TransducerAccessService.SyncReadTransducerSampleDataFromMulitipleChannelsOfATIM:
                    case TransducerAccessService.SyncReadTransducerBlockDataFromMulitipleChannelsOfATIM:
                    case TransducerAccessService.SyncReadTransducerSampleDataFromMultipleChannelsOfMultipleTIMs:
                    case TransducerAccessService.SyncReadTransducerBlockDataFromMultipleChannelsOfMultipleTIMs:
                    case TransducerAccessService.SyncWriteTransducerSampleDataFromAChannelOfATIM:
                    case TransducerAccessService.SyncWriteTransducerBlockDataFromAChannelOfATIM:
                    case TransducerAccessService.SyncWriteTransducerSampleDataFromMulitipleChannelsOfATIM:
                    case TransducerAccessService.SyncWriteTransducerBlockDataFromMulitipleChannelsOfATIM:
                    case TransducerAccessService.SyncWriteTransducerSampleDataFromMultipleChannelsOfMultipleTIMs:
                    case TransducerAccessService.SyncWriteTransducerBlockDataFromMultipleChannelsOfMultipleTIMs:
                    case TransducerAccessService.AsyncReadTransducerBlockDataFromAChannelOfATIM:
                    case TransducerAccessService.CallbackAsyncReadTransducerBlockDataFromAChannelOfATIM:
                    case TransducerAccessService.AsyncReadTransducerStreamDataFromAChannelOfATIM:
                    case TransducerAccessService.CallbackAsyncReadTransducerStreamDataFromAChannelOfATIM:
                    case TransducerAccessService.AsyncReadTransducerBlockDataFromMultipleChannelsOfATIM:
                    case TransducerAccessService.Callback:
                    case TransducerAccessService.AsyncReadTransducerBlockDataFromMultipleChannelOfMultipleTIMs:
                    case TransducerAccessService.CallbackAsyncReadTransducerBlockDataFromMultipleChannelOfMultipleTIMs:
                        return; // TODO
                }
            }
            else if (Message is TedsAccessMessage TedsAccessMessage)
            {
                if (TedsAccessMessage.TryParseRequest(out ChannelAddress Address,
                    out TedsAccessCode TedsAccessCode, out uint TedsOffset,
                    out double TimeoutSeconds))
                {
                    await RemoveErrorAsync(This, "TedsRequestError");
 
                    StringBuilder sb = new StringBuilder();
 
                    sb.Append(This.Topic.FullTopic);
                    sb.Append('/');
                    sb.Append(Hashes.BinaryToString(Address.NcapId));
 
                    if (!MessageSwitch.IsZero(Address.TimId))
                    {
                        sb.Append('/');
                        sb.Append(Hashes.BinaryToString(Address.TimId));
 
                        if (Address.ChannelId != 0)
                        {
                            sb.Append('/');
                            sb.Append(Address.ChannelId.ToString());
                        }
                    }
 
                    SubTopic = await This.Topic.Broker.GetTopic(sb.ToString(), true, false);
 
                    if (!(SubTopic?.Node is ITedsNode TedsNode))
                        return;
 
                    await TedsNode.TedsRequest(TedsAccessMessage, TedsAccessCode, TedsOffset, TimeoutSeconds);
                }
                else
                {
                    await LogErrorAsync(This, "TedsRequestError", "Unable to parse TEDS request.");
                    return;
                }
            }
            else if (Message is DiscoveryMessage DiscoveryMessage)
            {
                if (DiscoveryMessage.TryParseMessage(out ushort _, out DiscoveryData Data))
                {
                    await RemoveErrorAsync(This, "DiscoveryRequestError");
 
                    StringBuilder sb = new StringBuilder();
 
                    sb.Append(This.Topic.FullTopic);
                    if (!MessageSwitch.IsZero(Data.Channel.TimId))
                    {
                        sb.Append('/');
                        sb.Append(Hashes.BinaryToString(Data.Channel.NcapId));
 
                        if (!MessageSwitch.IsZero(Data.Channel.TimId))
                        {
                            sb.Append('/');
                            sb.Append(Hashes.BinaryToString(Data.Channel.TimId));
 
                            if (Data.Channel.ChannelId != 0)
                            {
                                sb.Append('/');
                                sb.Append(Data.Channel.ChannelId.ToString());
                            }
                        }
                    }
 
                    SubTopic = await This.Topic.Broker.GetTopic(sb.ToString(), true, false);
 
                    if (SubTopic?.Node is MeteringNode Node)
                    {
                        bool Broadcast = SubTopic.LocalTopic == "D0" &&
                            await Node.GetParent() is RootTopic;
 
                        LinkedList<INode> ToProcess = new LinkedList<INode>();
                        IEnumerable<INode> ChildNodes;
 
                        if (Broadcast)
                        {
                            ChildNodes = await Node.ChildNodes;
                            if (!(ChildNodes is null))
                            {
                                foreach (INode ChildNode in ChildNodes)
                                    ToProcess.AddLast(ChildNode);
                            }
                        }
                        else
                            ToProcess.AddLast(Node);
 
                        while (!(ToProcess.First is null))
                        {
                            INode ChildNode = ToProcess.First.Value;
                            ToProcess.RemoveFirst();
 
                            bool CheckChildren = Broadcast;
 
                            if (ChildNode is IDiscoverableNode DiscoverableNode)
                                await DiscoverableNode.DiscoveryRequest(DiscoveryMessage);
                            else
                                CheckChildren = ChildNode is DiscoverableTopicNode;
 
                            if (CheckChildren)
                            {
                                ChildNodes = await ChildNode.ChildNodes;
                                if (!(ChildNodes is null))
                                {
                                    foreach (INode ChildNode2 in ChildNodes)
                                        ToProcess.AddLast(ChildNode2);
                                }
                            }
                        }
                    }
                }
                else
                {
                    await LogErrorAsync(This, "DiscoveryRequestError", "Unable to parse TEDS request.");
                    return;
                }
            }
        }
 
        private static Task LogErrorAsync(MqttData This, string EventId, string Message)
        {
            return This.Topic?.Node?.LogErrorAsync(EventId, Message) ?? Task.CompletedTask;
        }
 
        private static Task RemoveErrorAsync(MqttData This, string EventId)
        {
            return This.Topic?.Node?.RemoveErrorAsync(EventId) ?? Task.CompletedTask;
        }
 
        public override Grade DefaultSupport => Grade.Perfect;
 
        public override Task StartReadout(ThingReference ThingReference, ISensorReadout Request, string Prefix, bool Last)
        {
            List<Field> Data = new List<Field>()
            {
                new Int32Field(ThingReference, this.Timestamp, this.Append(Prefix, "#Bytes"),
                    this.value?.Length ?? 0, FieldType.Momentary, FieldQoS.AutomaticReadout)
            };
 
            if (!(this.value is null) && this.value.Length <= 256)
            {
                Data.Add(new StringField(ThingReference, this.Timestamp, "Raw",
                    Convert.ToBase64String(this.value), FieldType.Momentary, FieldQoS.AutomaticReadout));
            }
 
            Request.ReportFields(Last, Data);
 
            return Task.CompletedTask;
        }
    }
}