///
/// Copyright (c) 2019 Sensus Slovensko a.s.
///
using System;
using System.Collections.Generic;
using System.IO;
using System.IO.Compression;
using System.Threading;
using log4net;
using RecordProcessing;
using RecordProcessing.Records;
namespace RecordProcessing
{
public enum RecordType
{
None, /// No record processing
RF_Test, /// RF_Test @ 433MHz or 868MHz
RF_Test_400_900, /// RF_Test for FlexNet @ 400MHz or 900MHz
CommTest, /// Communication test of water meter
FlowtubeTestHe, /// Flowtube test with Helium
FlowtubeTestAir, /// Flowtube test with air
Count
}
public enum RecordPostproc
{
None, /// Do nothing, leave a file where it is
Move, /// Move a file to destination directory
Compress, /// Compress a file to a destination directory (delete original file)
Count
}
public class RecordProcessing
{
static readonly ILog log = LogManager.GetLogger(typeof(RecordProcessing));
#region Constructor and parameters
public readonly RecordType RecordType;
RecordPostproc recordPostproc;
string testResultsFolder;
bool doExpandTestResultsFolder;
string archiveFolder;
bool doExpandArchiveFolder;
string fileNamePattern;
///
/// Create a dummy RecordProcessing object, worker thread will not be started.
///
public RecordProcessing()
{
this.RecordType = RecordType.None;
}
public RecordProcessing(RecordType recordType,
RecordPostproc recordPostproc,
string testResultsFolder,
bool doExpandTestResultsFolder,
string fileNamePattern)
{
if (recordPostproc != RecordPostproc.None)
{
throw new Exception("Argument 'recordPostproc' should be 'RecordPostproc.None' as no archive folder is defined");
}
this.RecordType = recordType;
this.recordPostproc = recordPostproc;
this.testResultsFolder = testResultsFolder;
this.doExpandTestResultsFolder = doExpandTestResultsFolder;
this.fileNamePattern = fileNamePattern;
this.archiveFolder = string.Empty;
this.doExpandArchiveFolder = false;
}
public RecordProcessing(RecordType recordType,
RecordPostproc recordPostproc,
string sourceFolder,
bool expandSourceFolder,
string fileNamePattern,
string archiveFolder,
bool expandArchiveFolder)
{
this.RecordType = recordType;
this.recordPostproc = recordPostproc;
this.testResultsFolder = sourceFolder;
this.doExpandTestResultsFolder = expandSourceFolder;
this.fileNamePattern = fileNamePattern;
this.archiveFolder = archiveFolder;
this.doExpandArchiveFolder = expandArchiveFolder;
}
#endregion Constructor and parameters
#region Worker thread and its control
public bool Running { get { return running; } }
bool running = false; /// true when worker thread is running
Thread workerThread;
public void StartProcessing()
{
if ((RecordType == RecordType.None) || running) return;
running = true;
workerThread = new Thread(Worker);
workerThread.CurrentCulture = Thread.CurrentThread.CurrentCulture;
workerThread.CurrentUICulture = Thread.CurrentThread.CurrentUICulture;
workerThread.Start();
}
public void StopProcessing()
{
if ((RecordType == RecordType.None) || !running) return;
running = false;
workerThread.Join(2000);
}
void Worker()
{
///
/// Create a reference list of files already existing in the source folder. These file would not be processed.
///
DateTime start = DateTime.Now;
string referenceFolder = doExpandTestResultsFolder ? Path.Combine(testResultsFolder, start.Year.ToString(), start.Month.ToString("D2"), start.Day.ToString("D2")) : testResultsFolder;
IList referenceFiles = new List();
IEnumerable currentFiles = new List();
try { currentFiles = Directory.EnumerateFiles(referenceFolder, fileNamePattern); }
catch (Exception) { }
foreach (var f in currentFiles) referenceFiles.Add(f);
while (running)
{
/// Wait 2.5 s, quit when 'running' flag is reset
for (int i = 0; i < 5; i++)
{
if (!running) break;
Thread.Sleep(500);
}
///
/// Expand folders if required
///
DateTime now = DateTime.Now;
string archiveFolderExt = doExpandArchiveFolder ? Path.Combine(archiveFolder, now.Year.ToString(), now.Month.ToString("D2"), now.Day.ToString("D2")) : archiveFolder;
string sourceFolderExt = doExpandTestResultsFolder ? Path.Combine(testResultsFolder, now.Year.ToString(), now.Month.ToString("D2"), now.Day.ToString("D2")) : testResultsFolder;
if (sourceFolderExt != referenceFolder)
{
/// Folder changes (after midnight when folder expansion was enabled)
/// Check if there are any new files in the original folder so that not a single record is lost
currentFiles = new List();
try { currentFiles = Directory.EnumerateFiles(referenceFolder, fileNamePattern); }
catch (Exception) { }
IList newFilesInPreviousFolder = GetDifference(referenceFiles, currentFiles);
referenceFolder = sourceFolderExt;
referenceFiles.Clear(); /// Clear the reference list of files so that all files in th enew folder are processed
/// Process newly added file, it is assumed there is no more then 1 such file.
if (newFilesInPreviousFolder.Count > 0)
{
ProcessOneFile(newFilesInPreviousFolder[0], archiveFolderExt);
continue;
}
}
/// Determine whether files were added
currentFiles = new List();
try { currentFiles = Directory.EnumerateFiles(sourceFolderExt, fileNamePattern); }
catch (Exception) { }
IList newFiles = GetDifference(referenceFiles, currentFiles);
/// Process newly added files
if (newFiles.Count > 0)
{
if (!ProcessOneFile(newFiles[0], archiveFolderExt))
{
referenceFiles.Add(newFiles[0]); /// Prevent double processing of this file
}
}
}
}
///
/// Determine difference between a current list of files and an original list of files.
///
/// Original list of files
/// Current list of files
/// List of files that were added to the original list
IList GetDifference(IList oriFiles, IEnumerable currentFiles)
{
IList newFiles = new List();
foreach (var f in currentFiles)
{
if (!oriFiles.Contains(f)) newFiles.Add(f);
}
return newFiles;
}
#endregion Worker thread and its control
#region Process one file
///
/// Process one record (one file).
///
/// Record file name
/// Destination folder for a successfully processed records
/// true when the record was deleted from the source directory (moved or compressed)
bool ProcessOneFile(string fileName, string archiveFolder)
{
IRecord record;
/// Extract information from the file
switch (RecordType)
{
case RecordType.RF_Test: record = RFTestRecord.FromFile(fileName); break;
case RecordType.RF_Test_400_900: record = RFTestRecord_400_900.FromFile(fileName); break;
case RecordType.CommTest: record = CommTestRecord.FromFile(fileName); break;
case RecordType.FlowtubeTestHe: record = FlowtubeTestHeRecord.FromFile(fileName); break;
case RecordType.FlowtubeTestAir: record = FlowtubeTestAirRecord.FromFile(fileName); break;
default: record = null; break;
}
/// Process data extracted from the file
bool processedOK = false;
if (record == null)
{
log.ErrorFormat("Record {0} is corrupt", fileName);
}
else
{
try
{
/// Send PcbNumber to anyone who handles SubmitRecordHandler
OnRecordAvailable(this, new SubmitRecordEventArgs(record));
processedOK = true;
log.InfoFormat("Record {0} successfully processed", fileName);
}
catch (Exception)
{
log.ErrorFormat("Processing record {0} failed", fileName);
}
}
/// Move or compress the original file
if (processedOK)
{
if (recordPostproc == RecordPostproc.Move)
{
///
/// Move the file to the archive filder
///
Directory.CreateDirectory(archiveFolder);
string target = Path.Combine(archiveFolder, Path.GetFileName(fileName));
if (File.Exists(target)) File.Delete(target); /// Ensure that the target does not exist
File.Move(fileName, target);
return true;
}
else if (recordPostproc == RecordPostproc.Compress)
{
/// Compress the file to the archive folder
Directory.CreateDirectory(archiveFolder);
using (FileStream outFile = File.Create(Path.Combine(archiveFolder, string.Format("{0}.gz", Path.GetFileName(fileName)))))
{
using (FileStream inFile = File.OpenRead(fileName))
{
using (GZipStream compress = new GZipStream(outFile, CompressionMode.Compress))
{
/// Copy the source file into the compression stream.
inFile.CopyTo(compress);
}
}
}
File.Delete(fileName); /// Delete the original file
return true;
}
else
{
/// Leave the file in its current location
return false;
}
}
else
{
/// TODO: What to do when processing fails? (which is unlikely)
return false;
}
}
///
/// Handler to submit PcbNumber-s extracted from records
///
public event EventHandler SubmitRecordHandler;
///
/// Called from ProcessOneFile()
///
void OnRecordAvailable(object sender, SubmitRecordEventArgs data)
{
if (SubmitRecordHandler == null) return;
try
{
SubmitRecordHandler(sender, data);
}
catch (Exception e)
{
log.Error("SubmitRecordHandler(...) failed", e);
}
}
#endregion Process one file
}
}