ProxyMqttChannelTopicNode.cs

using System;
using System.Collections.Generic;
using System.Text;
using System.Threading.Tasks;
using Waher.Content;
using Waher.Networking.MQTT;
using Waher.Networking.XMPP.Sensor;
using Waher.Runtime.Inventory;
using Waher.Runtime.Language;
using Waher.Script.Units;
using Waher.Things.Attributes;
using Waher.Things.DisplayableParameters;
using Waher.Things.Ieee1451.Ieee1451_0;
using Waher.Things.Ieee1451.Ieee1451_0.Messages;
using Waher.Things.Ieee1451.Ieee1451_0.TEDS;
using Waher.Things.Ieee1451.Ieee1451_0.TEDS.FieldTypes;
using Waher.Things.Ieee1451.Ieee1451_0.TEDS.FieldTypes.TransducerNameTeds;
using Waher.Things.Metering;
using Waher.Things.Mqtt;
using Waher.Things.Mqtt.Model;
using Waher.Things.SensorData;
 
namespace Waher.Things.Ieee1451.Ieee1451_1_6
{
    public class ProxyMqttChannelTopicNode : MqttChannelTopicNode, ITransducerNode, ITedsNode
    {
        private string proxyNodeId;
        private string proxyFieldName;
 
        public ProxyMqttChannelTopicNode()
        {
        }
 
        [Page(1, "IEEE 1451")]
        [Header(29, "Proxy Node ID:", 400)]
        [ToolTip(30, "Node ID of device to represent using this channel node.")]
        [Required]
        public string ProxyNodeId
        {
            get => this.proxyNodeId;
            set => this.proxyNodeId = value;
        }
 
        [Page(1, "IEEE 1451")]
        [Header(31, "Proxy Field Name:", 500)]
        [ToolTip(32, "Name of field to represent using this channel node.")]
        [Required]
        public string ProxyFieldName
        {
            get => this.proxyFieldName;
            set => this.proxyFieldName = value;
        }
 
        public override string LocalId
        {
            get
            {
                if (!string.IsNullOrEmpty(this.EntityName))
                    return this.EntityName;
                else
                    return this.NodeId;
            }
        }
 
        public override Task<string> GetTypeNameAsync(Language Language)
        {
            return Language.GetStringAsync(typeof(Ieee1451Parser), 2, "IEEE 1451.0 Proxy Channel");
        }
 
        public override async Task<IEnumerable<Parameter>> GetDisplayableParametersAsync(Language Language, RequestOrigin Caller)
        {
            LinkedList<Parameter> Parameters = (LinkedList<Parameter>)await base.GetDisplayableParametersAsync(Language, Caller);
 
            Parameters.AddLast(new Int32Parameter("ChannelId",
                await Language.GetStringAsync(typeof(MqttNcapTopicNode), 10, "Channel"),
                this.ChannelId));
 
            Parameters.AddLast(new StringParameter("TimId",
                await Language.GetStringAsync(typeof(MqttNcapTopicNode), 7, "TIM ID"),
                this.TimId));
 
            Parameters.AddLast(new StringParameter("NcapId",
                await Language.GetStringAsync(typeof(MqttNcapTopicNode), 4, "NCAP ID"),
                this.NcapId));
 
            Parameters.AddLast(new StringParameter("ProxyNodeId",
                await Language.GetStringAsync(typeof(MqttNcapTopicNode), 34, "Proxy Node ID"),
                this.ProxyNodeId));
 
            Parameters.AddLast(new StringParameter("ProxyFieldName",
                await Language.GetStringAsync(typeof(MqttNcapTopicNode), 35, "Proxy Field Name"),
                this.ProxyFieldName));
 
            return Parameters;
        }
 
        public override Task<bool> AcceptsParentAsync(INode Parent)
        {
            return Task.FromResult(Parent is ProxyMqttTimTopicNode);
        }
 
        public override Task<bool> AcceptsChildAsync(INode Child)
        {
            return Task.FromResult(false);
        }
 
        public override Grade Supports(MqttTopicRepresentation Topic)
        {
            if (Topic.SegmentIndex > 0 &&
                Topic.CurrentParentTopic?.Node is MqttTimTopicNode &&
                !(Topic.CurrentParentTopic?.Node is MqttChannelTopicNode) &&
                int.TryParse(Topic.CurrentSegment, out _))
            {
                return Grade.Excellent;
            }
            else
                return Grade.NotAtAll;
        }
 
        public override async Task<IMqttTopicNode> CreateNew(MqttTopicRepresentation Topic)
        {
            StringBuilder sb = new StringBuilder();
 
            sb.Append("Channel-");
            sb.Append(Topic.Segments[Topic.SegmentIndex - 1]);
            sb.Append('#');
            sb.Append(Topic.CurrentSegment);
 
            return new MqttChannelTopicNode()
            {
                NodeId = await GetUniqueNodeId(sb.ToString()),
                LocalTopic = Topic.CurrentSegment,
                ChannelId = int.Parse(Topic.CurrentSegment),
                TimId = Topic.Segments[Topic.SegmentIndex - 1],
                NcapId = Topic.Segments[Topic.SegmentIndex - 2]
            };
        }
 
        public override async Task StartReadout(ISensorReadout Request, bool DoneAfter)
        {
            try
            {
                await base.StartReadout(Request, false);
 
                Field[] Fields = await this.ReadSensor(Request.Actor);
                await Request.ReportFields(DoneAfter, Fields);
            }
            catch (Exception ex)
            {
                await Request.ReportErrors(DoneAfter, new ThingError(this, ex.Message));
            }
        }
 
        public async Task<Field[]> TryReadSensor(string Actor)
        {
            try
            {
                Field[] Result = await this.ReadSensor(Actor);
                return Result;
            }
            catch (Exception ex)
            {
                await this.LogErrorAsync("ProxyError", ex.Message);
                return null;
            }
        }
 
        public async Task<Field[]> ReadSensor(string Actor)
        {
            MeteringNode Node = await MeteringTopology.GetNode(this.proxyNodeId)
                ?? throw new Exception("Node not found in Metering Topology: " + this.proxyNodeId);
 
            if (!(Node is ISensor Sensor))
                throw new Exception("Node not a sensor node: " + this.proxyNodeId);
 
            List<Field> Fields = new List<Field>();
            List<ThingError> Errors = new List<ThingError>();
            TaskCompletionSource<bool> Completed = new TaskCompletionSource<bool>();
            InternalReadoutRequest Readout = new InternalReadoutRequest(Actor,
                new IThingReference[] { Sensor }, FieldType.Momentary,
                new string[] { this.proxyFieldName }, DateTime.MinValue, DateTime.MaxValue,
                (_, e) =>
                {
                    Fields.AddRange(e.Fields);
                    if (e.Done)
                        Completed.TrySetResult(true);
 
                    return Task.CompletedTask;
                },
                (_, e) =>
                {
                    Errors.AddRange(e.Errors);
                    if (e.Done)
                        Completed.TrySetResult(true);
 
                    return Task.CompletedTask;
                }, null);
 
            await Sensor.StartReadout(Readout);
 
            _ = Task.Delay(this.TimeoutMilliseconds).ContinueWith(
                (_) => Completed.TrySetException(new TimeoutException()));
 
            await Completed.Task;
 
            if (Errors.Count > 0)
            {
                foreach (ThingError Error in Errors)
                    throw new Exception(Error.ErrorMessage);
            }
 
            await this.RemoveErrorAsync("ProxyError");
 
            return Fields.ToArray();
        }
 
        public async Task TransducerDataRequest(TransducerAccessMessage TransducerAccessMessage,
            SamplingMode SamplingMode, double TimeoutSeconds)
        {
            if (!(await this.GetParent() is ProxyMqttTimTopicNode TimNode))
                return;
 
            if (!(await TimNode.GetParent() is ProxyMqttNcapTopicNode NcapNode))
                return;
 
            if (!(await NcapNode.GetParent() is DiscoverableTopicNode CommunicationNode))
                return;
 
            MqttBrokerNode BrokerNode = await CommunicationNode.GetBroker();
            if (BrokerNode is null)
                return;
 
            MqttBroker Broker = await BrokerNode.GetBroker();
            if (Broker is null)
                return;
 
            string Topic = await CommunicationNode.GetFullTopic();
            if (string.IsNullOrEmpty(Topic))
                return;
 
            Field[] Fields = await this.TryReadSensor(string.Empty);
            if (Fields is null)
                return; // TODO: Error response
 
            Field MainField = this.GetMainField(Fields);
            string StringValue;
 
            if (MainField is QuantityField Quantity &&
                Unit.TryParse(Quantity.Unit, out Unit MainUnit))
            {
                double Value = Quantity.Value;
                double NrDecimals = Quantity.NrDecimals;
 
                if (PhysicalUnits.TryCreate(MainUnit, ref Value, ref NrDecimals, out _))
                {
                    byte NrDec = (byte)(NrDecimals < 0 ? 0 : NrDecimals > 255 ? 255 : Math.Round(NrDecimals));
                    StringValue = CommonTypes.Encode(Value, NrDec);
                }
                else
                    StringValue = CommonTypes.Encode(Quantity.Value, Quantity.NrDecimals);
            }
            else if (MainField is null)
                return; // TODO: Error response
            else
                StringValue = MainField.ObjectValue?.ToString() ?? string.Empty;
 
            StringBuilder ToSniffer = BrokerNode.HasSniffers ? new StringBuilder() : null;
 
            byte[] Response = TransducerAccessMessage.SerializeResponse(0, this.NcapIdBinary, this.TimIdBinary,
                (ushort)this.ChannelId, StringValue, MainField.Timestamp, ToSniffer);
 
            if (!(ToSniffer is null))
                await BrokerNode.Information(ToSniffer.ToString());
 
            await Broker.Publish(Topic, MqttQualityOfService.AtLeastOnce, false, Response);
        }
 
        public Field GetMainField(Field[] Fields)
        {
            foreach (Field Field in Fields)
            {
                if (Field.Type == FieldType.Momentary && Field.Name == this.proxyFieldName)
                    return Field;
            }
 
            return null;
        }
 
        public async Task TedsRequest(TedsAccessMessage TedsAccessMessage,
            TedsAccessCode TedsAccessCode, uint TedsOffset, double TimeoutSeconds)
        {
            if (!(await this.GetParent() is ProxyMqttTimTopicNode TimNode))
                return;
 
            if (!(await TimNode.GetParent() is ProxyMqttNcapTopicNode NcapNode))
                return;
 
            if (!(await NcapNode.GetParent() is DiscoverableTopicNode CommunicationNode))
                return;
 
            MqttBrokerNode BrokerNode = await CommunicationNode.GetBroker();
            if (BrokerNode is null)
                return;
 
            MqttBroker Broker = await BrokerNode.GetBroker();
            if (Broker is null)
                return;
 
            string Topic = await CommunicationNode.GetFullTopic();
            if (string.IsNullOrEmpty(Topic))
                return;
 
            byte[] Response;
            StringBuilder ToSniffer;
 
            switch (TedsAccessCode)
            {
                case TedsAccessCode.MetaTEDS:
                    ToSniffer = BrokerNode.HasSniffers ? new StringBuilder() : null;
                    Response = TedsAccessMessage.SerializeResponse(0, this.NcapIdBinary,
                        this.TimIdBinary, (ushort)this.ChannelId, ToSniffer,
                        new TedsId(99, 255, (byte)TedsAccessCode.MetaTEDS, 2, 1));
                    break;
 
                case TedsAccessCode.ChanTEDS:
                    Field[] Fields = await this.TryReadSensor(TedsAccessCode.ToString());
                    if (Fields is null)
                        return;
 
                    List<TedsRecord> Records = new List<TedsRecord>();
                    Field MainField = this.GetMainField(Fields);
 
                    if (MainField is QuantityField Quantity &&
                        Unit.TryParse(Quantity.Unit, out Unit MainUnit))
                    {
                        double Value = Quantity.Value;
                        double NrDecimals = Quantity.NrDecimals;
 
                        if (PhysicalUnits.TryCreate(MainUnit, ref Value, ref NrDecimals, out PhysicalUnits Ieee1451Unit))
                            Records.Add(new Ieee1451_0.TEDS.FieldTypes.TransducerChannelTeds.PhysicalUnits(Ieee1451Unit));
                    }
 
                    ToSniffer = BrokerNode.HasSniffers ? new StringBuilder() : null;
 
                    Response = TedsAccessMessage.SerializeResponse(0, this.NcapIdBinary,
                        this.TimIdBinary, (ushort)this.ChannelId, ToSniffer,
                        new TedsId(99, 255, (byte)TedsAccessCode.ChanTEDS, 2, 1),
                        Records.ToArray());
                    break;
 
                case TedsAccessCode.XdcrName:
                    ToSniffer = BrokerNode.HasSniffers ? new StringBuilder() : null;
                    Response = TedsAccessMessage.SerializeResponse(0, this.NcapIdBinary,
                        this.TimIdBinary, (ushort)this.ChannelId, ToSniffer,
                        new TedsId(99, 255, (byte)TedsAccessCode.XdcrName, 2, 1),
                        new Format(true),
                        new Ieee1451_0.TEDS.FieldTypes.TransducerNameTeds.Content(this.EntityName));
                    break;
 
                default:
                    return;
            }
 
            if (!(ToSniffer is null))
                await BrokerNode.Information(ToSniffer.ToString());
 
            await Broker.Publish(Topic, MqttQualityOfService.AtLeastOnce, false, Response);
        }
 
    }
}