tbf/TBF/Rig/RegisterReaders/S640Stream/S640Stream.cs

770 lines
28 KiB
C#

///
/// Copyright (c) 2021 Sensus Slovensko a.s.
///
using System;
using System.IO;
using System.IO.Ports;
using log4net;
using Common;
using TBF.Rig.Generic;
using TBF.Rig.GenericDevices;
namespace TBF.Rig.RegisterReaders.S640Stream
{
/// <summary>
/// This component = instance of this class is a placeholder for a combined main watermeter
/// </summary>
public class S640Stream : ComponentBase, IDevice, IRegReaderDatastream, ISessionDataMngmnt, IOperation
{
private static readonly ILog log = LogManager.GetLogger(typeof(S640Stream));
public override string ToString() { return string.Format("{0}({1})", ClassName, Cfg.ToString(-1)); }
const int MaxDatastreamFramesCount = 80000; /// almost 3h at 8 Hz
readonly S640StreamCfg myCfg;
public int Position { get { return myCfg.Position; } }
public RegisterReaderType RegisterReaderType { get { return RegisterReaderType.DataStream; } }
public double PulsesPerLtr {
get { return 1000.0; }
set { PulsesPerLtr = value; }
}
public double LtrsPerPulse { get { return 1 / PulsesPerLtr; } }
public string QuantityUnits { get; set; }
public Unit VolumeUnits { get { return Unit.ml; } }
public double VolumeScaleFactor { get { return 1; } }
public Unit TimeUnits { get { return Unit.ms; } }
public double TimeScaleFactor { get { return 1; } }
///
/// Required for IRegisterReader interface
///
public int WMPulses { get { return wmPulses; } }
public int WMRefPulses { get { return wmRefPulses; } }
public double WMVolume { get { return wmVolume; } }
public double BeginWMState { get { return beginWMState; } }
public double EndWMState { get { return endWMState; } }
public double WMTestTime { get { return wmTestTime; } }
double beginWMState;
double endWMState;
double wmVolume;
int wmPulses;
int wmRefPulses;
double wmTestTime;
/// <summary>
/// Store/update values to be used as a part of the opto-data log file name.
/// </summary>
/// <param name="test">Currently executed test</param>
/// <param name="repetitionNr">Currently executed repetition number</param>
public void TestIsGoingToStartSoon(Config.Entities.Test test, int repetitionNr)
{
/// Store/update values to be used as a part of the opto-data log file name
testName = test.Name;
testRepeats = test.Repeats;
this.repetitionNr = repetitionNr;
}
///
string testName;
int testRepeats;
int repetitionNr;
///
/// PCB Number from the data stream
///
public Int64 PcbNumber { get { return (pcbNrFrequency1 >= 10) ? pcbNumber1 : 0; } }
long pcbNumber1;
long pcbNumber2;
long pcbNumber3;
long pcbNrFrequency1;
long pcbNrFrequency2;
long pcbNrFrequency3;
///
/// Radio address from the data stream
///
public Int64 RadioAddress { get { return (rAddrFrequency1 >= 10) ? radioAddress1 : 0; } }
long radioAddress1;
long radioAddress2;
long radioAddress3;
long rAddrFrequency1;
long rAddrFrequency2;
long rAddrFrequency3;
///
/// Extracted from datastream, passed to OptoTelegramRaw.UpdateFromString(...)
///
Int64 currentRawVolumeUnwrapped;
Int64 currentRawTimeUnwrapped;
///
/// Indices to determine start / end samples
///
public int TestStartFrameIx;
public int TestEndFrameIx;
int endFrameIdx1;
int endFrameIdx2;
int endFrameIdx3;
private int currentFrameIx;
bool startSampleAcquired;
///
/// Volume of water from the opto telegram
///
private double volumeLtr;
private double volumeLtr0; /// The very first volume from the first valid datastream frame
public double VolumeLtrStart { get { return volumeLtrStart; } } /// Test start volume for metrology
public double VolumeLtrEnd { get { return volumeLtrEnd; } } /// Test end volume for metrology
double volumeLtrStart;
double volumeLtrEnd;
///
/// Timestamp from the opto telegram
///
private double timestampSec;
private double timestampSec0; /// The very first timestamp from the first valid datastream frame
public bool NoSamples { get { return (timestampSecEnd - timestampSecStart) < float.Epsilon; } }
public double TimestampSecStart { get { return timestampSecStart; } }
public double TimestampSecEnd { get { return timestampSecEnd; } }
double timestampSecStart;
double timestampSecEnd;
DatastreamFrame[] optoData;
int optoDataCount;
string optoDataLogFileName;
///
DatastreamFrame toBeFlushed;
int flushedFramesCount;
public int FlushedFramesCount
{
get { return flushedFramesCount; }
set { flushedFramesCount = lastFlushedFramesCount = lastFlushedFramesCount_1 = value; }
}
int lastFlushedFramesCount_1;
int lastFlushedFramesCount;
public int FlushedFramesDelta
{
get
{
int retval = Math.Max(flushedFramesCount - lastFlushedFramesCount, lastFlushedFramesCount - lastFlushedFramesCount_1);
lastFlushedFramesCount_1 = lastFlushedFramesCount;
lastFlushedFramesCount = flushedFramesCount;
return retval;
}
}
///
/// Opto serial port and worker thread related private variables
///
private SerialPort serialPort; /// Used in DebugMode.Normal
private TextReader textReader; /// Used instead of serialPort in DebugMode.Simulate
public S640Stream() { }
public S640Stream(IComponentCfg cfg)
: base(cfg)
{
myCfg = cfg as S640StreamCfg;
}
/// <summary>
/// Clear data related to a specific water meter
/// </summary>
public void StartSession()
{
currentRawVolumeUnwrapped = 0;
currentRawTimeUnwrapped = 0;
pcbNumber1 = 0;
pcbNumber2 = 0;
pcbNumber3 = 0;
pcbNrFrequency1 = 0;
pcbNrFrequency2 = 0;
pcbNrFrequency3 = 0;
radioAddress1 = 0;
radioAddress2 = 0;
radioAddress3 = 0;
rAddrFrequency1 = 0;
rAddrFrequency2 = 0;
rAddrFrequency3 = 0;
flushedFramesCount = 0;
synchronized = false;
synchronized2 = false;
partOfTelegram = string.Empty;
}
public void SaveMark(object mark)
{
}
public void EndSession()
{
datastreamParsingEnabled = false;
}
public override void Initialize()
{
datastreamParsingEnabled = false;
/// Allocate memory for opto-data from iPerl
optoData = new DatastreamFrame[MaxDatastreamFramesCount];
for (int i = 0; i < MaxDatastreamFramesCount; i++) optoData[i] = new DatastreamFrame();
toBeFlushed = new DatastreamFrame();
if (DebugLevel == DebugMode.Normal)
{
/// Prepare serial port
serialPort = new SerialPort(string.Format("COM{0}", myCfg.ComPortNr),
myCfg.BaudRate,
myCfg.Parity,
myCfg.DataBits,
myCfg.StopBits);
serialPort.Handshake = myCfg.Handshake;
serialPort.Open();
log.FatalFormat("{0} initialized: {1}", Name, this);
}
else if (DebugLevel == DebugMode.Replay)
{
serialPort = null;
try
{
if (Position % 2 == 0)
{
textReader = new StreamReader("C:\\TBF\\Simulate\\serialstream-bad.txt");
}
else
{
textReader = new StreamReader("C:\\TBF\\Simulate\\serialstream.txt");
}
}
catch (Exception)
{
throw new Exception(Position % 2 == 0 ? "Missing file C:\\TBF\\Simulate\\serialstream-bad.txt"
: "Missing file C:\\TBF\\Simulate\\serialstream.txt");
}
log.FatalFormat("{0} in replay mode: {1}", Name, this);
}
else
{
serialPort = null;
log.FatalFormat("{0} simulated: {1}", Name, this);
}
}
public void RunDeviceBefore()
{
if (DebugLevel == DebugMode.Normal || DebugLevel == DebugMode.Replay)
{
try
{
ReadDatastream(datastreamParsingEnabled ? DatastreamState.Read : DatastreamState.Flush);
}
catch (Exception e)
{
DebugLevel = DebugMode.FailureDuringOperation;
log.FatalFormat("Opto-data serial port failure : {0}", e.Message);
if (e.InnerException != null)
{
log.FatalFormat("InnerMessage : {0}", e.InnerException.Message);
}
}
}
else if (DebugLevel == DebugMode.Simulate || DebugLevel == DebugMode.FailureDuringOperation)
{
}
}
public void RunDeviceAfter() { }
public void StopDevice()
{
try
{
if (DebugLevel == DebugMode.Normal && serialPort != null)
{
serialPort.Close();
serialPort = null;
}
}
catch
{
}
}
public void StopDevice2() { }
/// <summary>
/// Events: Event.ReadRegisterDone, Event.Error
/// </summary>
/// <returns>ReadWaterMeter instance reference casted to IOperaton</returns>
public IOperation ReadDatastreamOp()
{
return this;
}
/// <summary>
/// Clear data/counters related to a specific tests
/// </summary>
public void Clear()
{
volumeLtr = 0;
volumeLtr0 = 0;
timestampSec = 0;
timestampSec0 = 0;
ReadPulses();
}
public void TestCompleted()
{
/// TODO: Implement
}
int timeFromStart; /// This is to determine when the test start sample should be taken
/// <summary>Start this operation</summary>
public void Start()
{
Clear();
/// Reset opto data
timeFromStart = 0;
optoDataCount = 0;
currentFrameIx = -1;
startSampleAcquired = false;
TestStartFrameIx = 0;
endFrameIdx1 = 0;
endFrameIdx2 = 0;
endFrameIdx3 = 0;
TestEndFrameIx = 0;
timestampSecStart = 0;
timestampSecEnd = 0;
volumeLtrStart = 0;
volumeLtrEnd = 0;
/// File name
optoDataLogFileName = string.Format("{0}_{1}_{2}_{3}.txt",
StateMachine.CycleStartTimeStamp.ToString("HHmmss"),
Position.ToString("D2"),
testName,
repetitionNr);
StartParsingDatastream();
}
/// <summary>Run this operation</summary>
/// <returns>eventDone</returns>
public Event Run()
{
timeFromStart += StateMachine.Period;
ReadPulses();
if (!startSampleAcquired && timeFromStart >= 4 && currentFrameIx >= 0) /// 4 seconds after operation start
{
/// Take the test start sample
startSampleAcquired = true;
TestStartFrameIx = currentFrameIx;
log.InfoFormat("{0} : Run() ... TestStartFrameIx = {1}", Name, TestStartFrameIx);
}
else if (startSampleAcquired)
{
/// Shift data in pipelines
TestEndFrameIx = endFrameIdx3;
endFrameIdx3 = endFrameIdx2;
endFrameIdx2 = endFrameIdx1;
endFrameIdx1 = currentFrameIx;
}
return Event.ReadRegisterDone;
}
/// <summary>Stop this operation</summary>
public void Stop()
{
StopParsingDatastream();
log.InfoFormat("{0} : Stop()", Name);
log.DebugFormat("{0} : TestStartFrameIx = {1}, TestEndFrameIx = {2}, optoDataCount = {3}", Name, TestStartFrameIx, TestEndFrameIx, optoDataCount);
Int64 firstVolumeRaw = optoData[TestStartFrameIx].VolumeRawExt;
int frameIx = TestStartFrameIx;
do { frameIx++; }
while (frameIx < optoDataCount / 2 &&
optoData[frameIx].Flags == FrameFlags.OK &&
optoData[frameIx].VolumeRawExt == firstVolumeRaw);
/// Store the start sample, make a mark
if (frameIx < optoDataCount / 2 && optoData[frameIx].Flags == FrameFlags.OK)
{
TestStartFrameIx = frameIx;
optoData[frameIx].Flags = FrameFlags.OK_TestStart;
timestampSecStart = (double)optoData[frameIx].TimestampExt / DatastreamFrame.TimeScaleFactor;
volumeLtrStart = (double)optoData[frameIx].VolumeRawExt / DatastreamFrame.VolumeScaleFactor;
}
int endFrameIx = 0;
Int64 lastVolumeRaw = optoData[TestStartFrameIx].VolumeRawExt;
while (true)
{
for (int i = 0; i < 8; i++)
{
do { frameIx++; }
while (frameIx < optoDataCount &&
optoData[frameIx].Flags == FrameFlags.OK &&
optoData[frameIx].VolumeRawExt == lastVolumeRaw);
if (frameIx < optoDataCount && frameIx < TestEndFrameIx && optoData[frameIx].Flags == FrameFlags.OK)
{
lastVolumeRaw = optoData[frameIx].VolumeRawExt;
}
}
if (frameIx >= optoDataCount || frameIx >= TestEndFrameIx) break;
if (frameIx < optoDataCount && frameIx < TestEndFrameIx && optoData[frameIx].Flags == FrameFlags.OK)
{
optoData[frameIx].Flags = FrameFlags.OK_FullRound;
lastVolumeRaw = optoData[frameIx].VolumeRawExt;
endFrameIx = frameIx;
}
}
if (endFrameIx != 0)
{
TestEndFrameIx = endFrameIx;
optoData[endFrameIx].Flags = FrameFlags.OK_TestEnd;
timestampSecEnd = (double)optoData[endFrameIx].TimestampExt / DatastreamFrame.TimeScaleFactor;
volumeLtrEnd = (double)optoData[endFrameIx].VolumeRawExt / DatastreamFrame.VolumeScaleFactor;
}
log.DebugFormat("{0} : TestStartFrameIx = {1}, TestEndFrameIx = {2}", Name, TestStartFrameIx, TestEndFrameIx);
DatastreamFrame.FIRFilterFlow(optoData, optoDataCount);
SaveDatastreamLog();
}
void SaveDatastreamLog()
{
string directory = string.Format("C:\\TBF\\ProcessData\\{0}\\{1}\\{2}\\",
StateMachine.CycleStartTimeStamp.ToString("yy"),
StateMachine.CycleStartTimeStamp.ToString("MM"),
StateMachine.CycleStartTimeStamp.ToString("dd"));
string datastreamLogPathName = directory + optoDataLogFileName;
try
{
Directory.CreateDirectory(directory);
using (TextWriter datastreamLogFile = new StreamWriter(datastreamLogPathName))
{
for (int i = 0; i < optoDataCount; i++)
{
datastreamLogFile.WriteLine(optoData[i].ToString());
}
datastreamLogFile.Close();
}
}
catch (Exception exc)
{
File.Delete(datastreamLogPathName);
log.ErrorFormat(string.Format("Error writing into file {0}", datastreamLogPathName));
log.ErrorFormat(string.Format("Exception message: {0}", exc.Message));
}
}
void ReadPulses()
{
beginWMState = volumeLtr0;
endWMState = volumeLtr;
wmVolume = Math.Abs(endWMState - beginWMState);
wmPulses = Convert.ToInt32(Math.Round(wmVolume * PulsesPerLtr));
wmRefPulses = StateMachine.ControlBoardMain.RefPulses;
wmTestTime = timestampSec - timestampSec0;
}
bool datastreamParsingEnabled;
/// <summary> Flush internal buffers and start parsing the opto serial port data </summary>
void StartParsingDatastream()
{
datastreamParsingEnabled = true;
}
/// <summary> Stop parsig the opto serial port data </summary>
void StopParsingDatastream()
{
datastreamParsingEnabled = false;
}
///
/// Variables storing the context of serial port data parsing (ReadOptoSerialPort(...))
///
bool synchronized;
bool synchronized2;
string partOfTelegram;
/// <summary>
/// 9600 Bd, 8 data bits, 1 stop bit, no parity
///
/// Telegram description:
///
/// AAAAAAAAAAAA[tab]BBBBBBBBBB[tab]CCCCCCCCC[tab]DDDDDDDDDD[tab]EE[cr][lf] (49 bytes)
///
/// Data Comment Type Calculate to decimal
/// ----------------------------------------------------------------
/// AAAAAAAAAAAA PCB number Decimal
/// BBBBBBBBBB Radio address Decimal
/// CCCCCCCCC Meter reading Decimal Volume in ml
/// DDDDDDDDDD Timestamp Decimal Time in ms
/// EE Checksum Hexadecimal
/// ----------------------------------------------------------------
///
/// Example:
/// 540210730770 4291524429 000250215 0433411876 56
/// 540210730770 4291524429 000250223 0433412126 4A
/// ...
/// </summary>
/// <param name="streamState">OptoState.Read or OptoState.Flush</param>
void ReadDatastream(DatastreamState streamState)
{
string received = null;
if (DebugLevel == DebugMode.Normal)
{
int nrBytes = serialPort.BytesToRead;
char[] buffer = new char[nrBytes];
serialPort.Read(buffer, 0, nrBytes);
received = new string(buffer);
}
else if (DebugLevel == DebugMode.Replay)
{
string line = textReader.ReadLine();
received = line + '\r' + '\n';
}
if (received != null)
{
string allRcvd = partOfTelegram + received;
while (true)
{
int pos = allRcvd.IndexOf("\r\n");
if (pos < 0)
{
/// No CR+LF found, wait for more characters in the buffer (next invocation)
partOfTelegram = allRcvd;
return;
}
else
{
/// CR+LF found ==> Check datastream state: Read or Flush
if (streamState == DatastreamState.Read)
{
if (pos < DatastreamFrame.FrameLength - 2)
{
/// CR+LF found too early, truncate the beginning incl CR+LF and keep scanning in this loop
allRcvd = allRcvd.Substring(pos + 2);
if (synchronized)
{
optoData[optoDataCount].Counter = optoDataCount;
optoData[optoDataCount++].SetFlags(FrameFlags.SyncError);
}
synchronized = true;
}
// CR+LF found and (pos >= OptoTelegramRaw.Length - 2)
else if (optoData[optoDataCount].UpdateFromString(allRcvd.Substring(pos - DatastreamFrame.FrameLength + 2, DatastreamFrame.FrameLength),
ref currentRawVolumeUnwrapped, ref currentRawTimeUnwrapped,
optoDataCount))
{
OptoTelegramRreceived(optoDataCount++, currentRawVolumeUnwrapped, currentRawTimeUnwrapped, synchronized2);
synchronized2 = synchronized;
allRcvd = allRcvd.Substring(pos + 2);
}
else
{
optoData[optoDataCount].Counter = optoDataCount;
optoDataCount++;
allRcvd = allRcvd.Substring(pos + 2);
}
}
else /// optoState == SerialStreamState.Flush
{
Int64 pcbNumber;
Int64 radioAddress;
if (pos < DatastreamFrame.FrameLength - 2)
{
/// CR+LF found too early, truncate the beginning incl CR+LF and keep scanning in this loop
allRcvd = allRcvd.Substring(pos + 2);
synchronized = true;
}
// CR+LF found and (pos >= OptoTelegram.Length - 2)
else if (toBeFlushed.UpdateFromStringDummy(allRcvd.Substring(pos - DatastreamFrame.FrameLength + 2, DatastreamFrame.FrameLength),
ref currentRawVolumeUnwrapped, ref currentRawTimeUnwrapped,
out pcbNumber,
out radioAddress))
{
flushedFramesCount++;
synchronized2 = synchronized;
allRcvd = allRcvd.Substring(pos + 2);
///
/// Process received PCB number
///
if (pcbNrFrequency1 == 0 || pcbNumber == pcbNumber1)
{
pcbNumber1 = pcbNumber;
pcbNrFrequency1++;
}
else if (pcbNrFrequency2 == 0 || pcbNumber == pcbNumber2)
{
pcbNumber2 = pcbNumber;
pcbNrFrequency2++;
if (pcbNrFrequency2 > pcbNrFrequency1)
{
long tmp = pcbNumber1;
pcbNumber1 = pcbNumber2;
pcbNumber2 = tmp;
tmp = pcbNrFrequency1;
pcbNrFrequency1 = pcbNrFrequency2;
pcbNrFrequency2 = tmp;
}
}
else if (pcbNrFrequency3 == 0 || pcbNumber == pcbNumber3)
{
pcbNumber3 = pcbNumber;
pcbNrFrequency3++;
if (pcbNrFrequency3 > pcbNrFrequency2)
{
long tmp = pcbNumber2;
pcbNumber2 = pcbNumber3;
pcbNumber3 = tmp;
tmp = pcbNrFrequency2;
pcbNrFrequency2 = pcbNrFrequency3;
pcbNrFrequency3 = tmp;
}
if (pcbNrFrequency2 > pcbNrFrequency1)
{
long tmp = pcbNumber1;
pcbNumber1 = pcbNumber2;
pcbNumber2 = tmp;
tmp = pcbNrFrequency1;
pcbNrFrequency1 = pcbNrFrequency2;
pcbNrFrequency2 = tmp;
}
}
///
/// Process received radio address
///
if (rAddrFrequency1 == 0 || radioAddress == radioAddress1)
{
radioAddress1 = radioAddress;
rAddrFrequency1++;
}
else if (rAddrFrequency2 == 0 || radioAddress == radioAddress2)
{
radioAddress2 = radioAddress;
rAddrFrequency2++;
if (rAddrFrequency2 > rAddrFrequency1)
{
long tmp = radioAddress1;
radioAddress1 = radioAddress2;
radioAddress2 = tmp;
tmp = rAddrFrequency1;
rAddrFrequency1 = rAddrFrequency2;
rAddrFrequency2 = tmp;
}
}
else if (rAddrFrequency3 == 0 || radioAddress == radioAddress3)
{
radioAddress3 = radioAddress;
rAddrFrequency3++;
if (rAddrFrequency3 > rAddrFrequency2)
{
long tmp = radioAddress2;
radioAddress2 = radioAddress3;
radioAddress3 = tmp;
tmp = rAddrFrequency2;
rAddrFrequency2 = rAddrFrequency3;
rAddrFrequency3 = tmp;
}
if (rAddrFrequency2 > rAddrFrequency1)
{
long tmp = rAddrFrequency1;
rAddrFrequency1 = rAddrFrequency2;
rAddrFrequency2 = tmp;
tmp = radioAddress1;
radioAddress1 = radioAddress2;
radioAddress2 = tmp;
}
}
}
else
{
allRcvd = allRcvd.Substring(pos + 2);
}
}
}
}
}
}
void OptoTelegramRreceived(int currentIx, long currentRawVolumeUnwrapped, long currentRawTimeUnwrapped, bool async)
{
DatastreamFrame optoTelegram = optoData[currentIx];
currentFrameIx = currentIx;
if (volumeLtr == 0 && volumeLtr0 == 0)
{
volumeLtr0 = volumeLtr = (double)currentRawVolumeUnwrapped / DatastreamFrame.VolumeScaleFactor;
}
else
{
volumeLtr = (double)currentRawVolumeUnwrapped / DatastreamFrame.VolumeScaleFactor;
}
if (timestampSec == 0 && timestampSec0 == 0)
{
timestampSec0 = timestampSec = (double)currentRawTimeUnwrapped / DatastreamFrame.TimeScaleFactor;
}
else
{
timestampSec = (double)currentRawTimeUnwrapped / DatastreamFrame.TimeScaleFactor;
}
}
}
}