/*=========================================================================

  Program:   ParaView
  Module:    vtkPVEnSightMasterServerReader.cxx

  Copyright (c) Kitware, Inc.
  All rights reserved.
  See Copyright.txt or http://www.paraview.org/HTML/Copyright.html for details.

     This software is distributed WITHOUT ANY WARRANTY; without even
     the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR
     PURPOSE.  See the above copyright notice for more information.

=========================================================================*/
#include "vtkPVEnSightMasterServerReader.h"

#include "vtkCellData.h"
#include "vtkDataArray.h"
#include "vtkDataArrayCollection.h"
#include "vtkImageData.h"
#include "vtkInformation.h"
#include "vtkInformationVector.h"
#include "vtkMultiProcessController.h"
#include "vtkObjectFactory.h"
#include "vtkPVConfig.h"
#include "vtkPVEnSightMasterServerTranslator.h"
#include "vtkPointData.h"
#include "vtkPolyData.h"
#include "vtkRectilinearGrid.h"
#include "vtkStructuredGrid.h"
#include "vtkTransmitPolyDataPiece.h"
#include "vtkTransmitUnstructuredGridPiece.h"
#include "vtkUnstructuredGrid.h"

#ifdef PARAVIEW_USE_MPI
#include "vtkMPICommunicator.h"
#endif

#include <string>
#include <vector>

#include <ctype.h>

//----------------------------------------------------------------------------
vtkStandardNewMacro(vtkPVEnSightMasterServerReader);

vtkCxxSetObjectMacro(vtkPVEnSightMasterServerReader, Controller, vtkMultiProcessController);
vtkMultiProcessController* vtkPVEnSightMasterServerReader::GetController()
{
  return this->Controller;
}

//----------------------------------------------------------------------------
class vtkPVEnSightMasterServerReaderInternal
{
public:
  std::vector<std::string> PieceFileNames;
  int EnSightVersion;
  int NumberOfTimeSets;
  int NumberOfOutputs;
  std::vector<int> CumulativeTimeSetSizes;
  std::vector<float> TimeSetValues;
};

//----------------------------------------------------------------------------
vtkPVEnSightMasterServerReader::vtkPVEnSightMasterServerReader()
{
  this->Internal = new vtkPVEnSightMasterServerReaderInternal;
  this->Controller = 0;
  this->SetController(vtkMultiProcessController::GetGlobalController());
  this->InformationError = 0;
  this->ExtentTranslator = vtkPVEnSightMasterServerTranslator::New();
  this->NumberOfPieces = 0;
}

//----------------------------------------------------------------------------
vtkPVEnSightMasterServerReader::~vtkPVEnSightMasterServerReader()
{
  this->SetController(0);
  delete this->Internal;
  this->ExtentTranslator->Delete();
}

//----------------------------------------------------------------------------
void vtkPVEnSightMasterServerReader::PrintSelf(ostream& os, vtkIndent indent)
{
  this->Superclass::PrintSelf(os, indent);
  os << indent << "Controller: " << this->Controller << "\n";
  os << indent << "Error: " << this->InformationError << "\n";
  os << indent << "NumberOfPieces: " << this->NumberOfPieces << endl;
}

//----------------------------------------------------------------------------
#ifdef PARAVIEW_USE_MPI
template <class T>
int vtkPVEnSightMasterServerReaderSyncValues(
  T* data, int numValues, int numPieces, vtkMultiProcessController* controller)
{
  // Compare values on all processes that will read real pieces.
  // Returns whether the values match.  If they match, all processes'
  // values are modified to match that of node 0.  This will leave the
  // values unchanged on processes that will read real data, but
  // inform the other processes of the proper values.

  if (!controller)
  {
    return VTK_ERROR;
  }

  vtkMPICommunicator* communicator =
    vtkMPICommunicator::SafeDownCast(controller->GetCommunicator());

  if (!communicator)
  {
    return VTK_ERROR;
  }

  int numProcs = controller->GetNumberOfProcesses();
  int myid = controller->GetLocalProcessId();

  // Collect all the values to node 0.
  T* values = new T[numProcs * numValues];
  communicator->Gather(data, values, numValues, 0);

  int result = VTK_OK;
  // Node 0 compares its values to those from other processes that
  // will actually be reading data.
  if (myid == 0)
  {
    for (int i = 1; (result == VTK_OK) && (i < numPieces); i++)
    {
      for (int j = 0; (result == VTK_OK) && (j < numValues); j++)
      {
        if (values[i * numValues + j] != values[j])
        {
          result = VTK_ERROR;
        }
      }
    }
  }

  // Free buffer where values were collected.
  delete[] values;

  // Broadcast result of comparison to all processes.
  communicator->Broadcast(&result, 1, 0);

  // If the results were okay, broadcast the correct values to all
  // processes so that those that will not read can have the correct
  // values.
  if (result == VTK_OK)
  {
    communicator->Broadcast(data, numValues, 0);
  }

  return result;
}
#else
template <class T>
int vtkPVEnSightMasterServerReaderSyncValues(T*, int, int, vtkMultiProcessController*)
{
  return VTK_OK;
}
#endif

//----------------------------------------------------------------------------
int vtkPVEnSightMasterServerReader::RequestInformation(
  vtkInformation* request, vtkInformationVector** inputVector, vtkInformationVector* outputVector)
{
  int i;

  // Assume no error will occur.
  this->InformationError = 0;

  // Make sure we have a controller.
  if (!this->Controller)
  {
    vtkErrorMacro("ExecuteInformation requires a Controller.");
    this->InformationError = 1;
    return 0;
  }

  // Parse the input file.  This will set NumberOfPieces.
  int parseResults[2];
  parseResults[0] = this->ParseMasterServerFile();
  parseResults[1] = (parseResults[0] == VTK_OK) ? this->NumberOfPieces : -1;

  // Make sure all nodes could parse the file and agree on the number
  // of pieces.
  if ((vtkPVEnSightMasterServerReaderSyncValues(
         parseResults, 2, this->Controller->GetNumberOfProcesses(), this->Controller) != VTK_OK) ||
    (parseResults[0] != VTK_OK))
  {
    vtkErrorMacro("Error parsing the master server file.");
    this->InformationError = 1;
    return 0;
  }

  // Compare case file versions.
  this->Internal->EnSightVersion = -1;
  int piece = this->Controller->GetLocalProcessId();
  if (piece < this->NumberOfPieces)
  {
    // Let the superclass read the file information on this node.
    this->SuperclassExecuteInformation(request, inputVector, outputVector);
    this->Internal->EnSightVersion = this->EnSightVersion;
  }
  if (vtkPVEnSightMasterServerReaderSyncValues(
        &this->Internal->EnSightVersion, 1, this->NumberOfPieces, this->Controller) != VTK_OK)
  {
    vtkErrorMacro("EnSight version mismatch across nodes.");
    this->InformationError = 1;
    return 0;
  }

  // Make sure all nodes could read their files.
  if (this->Internal->EnSightVersion < 0)
  {
    vtkErrorMacro("Error reading case file on at least one node.");
    this->InformationError = 1;
    return 0;
  }

  // Compare number of time sets.
  vtkDataArrayCollection* timeSets = this->GetTimeSets();
  this->Internal->NumberOfTimeSets = timeSets ? timeSets->GetNumberOfItems() : 0;
  if (vtkPVEnSightMasterServerReaderSyncValues(
        &this->Internal->NumberOfTimeSets, 1, this->NumberOfPieces, this->Controller) != VTK_OK)
  {
    vtkErrorMacro("Number of time sets not equal on all nodes.");
    this->InformationError = 1;
    return 0;
  }

  // Compare the time sets' sizes.
  this->Internal->CumulativeTimeSetSizes.resize(this->Internal->NumberOfTimeSets + 1);
  if (piece < this->NumberOfPieces)
  {
    this->Internal->CumulativeTimeSetSizes[0] = 0;
    for (i = 0; i < this->Internal->NumberOfTimeSets; ++i)
    {
      this->Internal->CumulativeTimeSetSizes[i + 1] =
        (this->Internal->CumulativeTimeSetSizes[i] + timeSets->GetItem(i)->GetNumberOfTuples());
    }
  }
  if (vtkPVEnSightMasterServerReaderSyncValues(&*this->Internal->CumulativeTimeSetSizes.begin(),
        this->Internal->NumberOfTimeSets + 1, this->NumberOfPieces, this->Controller) != VTK_OK)
  {
    vtkErrorMacro("Time set sizes not equal on all nodes.");
    this->InformationError = 1;
    return 0;
  }

  // All time sets have the correct size.  Now compare the values.
  // The CumulativeTimeSetSizes has the sizes on all nodes, even those
  // beyond the number of pieces.

  // Compare time set values.
  this->Internal->TimeSetValues.clear();
  if (piece < this->NumberOfPieces)
  {
    // This is a real piece, fill in the time set values.
    for (i = 0; i < this->Internal->NumberOfTimeSets; ++i)
    {
      vtkDataArray* array = timeSets->GetItem(i);
      int numValues = array->GetNumberOfTuples();
      for (int j = 0; j < numValues; ++j)
      {
        this->Internal->TimeSetValues.push_back(array->GetTuple1(j));
      }
    }
  }
  else
  {
    // This is not a piece, allocate memory to receive the time set values.
    this->Internal->TimeSetValues.resize(
      this->Internal->CumulativeTimeSetSizes[this->Internal->NumberOfTimeSets]);
  }
  if (vtkPVEnSightMasterServerReaderSyncValues(&*this->Internal->TimeSetValues.begin(),
        static_cast<int>(this->Internal->TimeSetValues.size()), this->NumberOfPieces,
        this->Controller) != VTK_OK)
  {
    vtkErrorMacro("Time set values do not match on all nodes.");
    this->InformationError = 1;
    return 0;
  }

  return 1;
}

//----------------------------------------------------------------------------
int vtkPVEnSightMasterServerReader::RequestData(
  vtkInformation* request, vtkInformationVector** inputVector, vtkInformationVector* outputVector)
{
  // Do not execute if ExecuteInformation failed.
  if (this->InformationError)
  {
    return 0;
  }

  // Compare the number of outputs.
  int piece = this->Controller->GetLocalProcessId();
  if (piece < this->NumberOfPieces)
  {
    // Let the superclass read the data and setup the outputs.
    this->SuperclassExecuteData(request, inputVector, outputVector);
  }
  if (vtkPVEnSightMasterServerReaderSyncValues(
        &this->Internal->NumberOfOutputs, 1, this->NumberOfPieces, this->Controller) != VTK_OK)
  {
    vtkErrorMacro("Number of outputs does not match on all nodes.");
    return 0;
  }

  return 1;
}

//----------------------------------------------------------------------------
void vtkPVEnSightMasterServerReader::SuperclassExecuteInformation(
  vtkInformation* request, vtkInformationVector** inputVector, vtkInformationVector* outputVector)
{
  // Fake the filename when the superclass uses it.
  int piece = this->Controller->GetLocalProcessId();
  char* temp = this->CaseFileName;
  this->CaseFileName = const_cast<char*>(this->Internal->PieceFileNames[piece].c_str());
  this->Superclass::RequestInformation(request, inputVector, outputVector);
  this->CaseFileName = temp;
}

//----------------------------------------------------------------------------
void vtkPVEnSightMasterServerReader::SuperclassExecuteData(
  vtkInformation* request, vtkInformationVector** inputVector, vtkInformationVector* outputVector)
{
  // Fake the filename when the superclass uses it.
  int piece = this->Controller->GetLocalProcessId();
  char* temp = this->CaseFileName;
  this->CaseFileName = const_cast<char*>(this->Internal->PieceFileNames[piece].c_str());
  this->Superclass::RequestData(request, inputVector, outputVector);
  this->CaseFileName = temp;
}

//----------------------------------------------------------------------------
static int vtkPVEnSightMasterServerReaderIsSpace(char c)
{
  return isspace(c);
}

//----------------------------------------------------------------------------
static int vtkPVEnSightMasterServerReaderStartsWith(const char* str1, const char* str2)
{
  if (!str1 || !str2 || strlen(str1) < strlen(str2))
  {
    return 0;
  }
  return !strncmp(str1, str2, strlen(str2));
}

//----------------------------------------------------------------------------
// Due to a buggy stream library on the HP and another on Mac OSX, we
// need this very carefully written version of getline.  Returns true
// if any data were read before the end-of-file was reached.
static int vtkPVEnSightMasterServerReaderGetLineFromStream(istream& is, std::string& line)
{
  const int bufferSize = 1024;
  char buffer[bufferSize];
  line = "";
  int haveData = 0;

  // If no characters are read from the stream, the end of file has
  // been reached.
  while ((is.getline(buffer, bufferSize), is.gcount() > 0))
  {
    char* p;
    int ppos, numCharsRead = is.gcount();

    haveData = 1;

    // Get rid of endline character.
    p = buffer;
    ppos = 0;
    while (ppos < bufferSize && ppos < numCharsRead && *p != '\0')
    {
      if (*p == '\r')
      {
        // Change existing method as little as possible.
        *p = '\0';
        line.append(buffer);
        return haveData;
      }
      ++p;
      ++ppos;
    }

    line.append(buffer);

    // If newline character was read, the gcount includes the
    // character, but the buffer does not.  The end of line has been
    // reached.
    if (strlen(buffer) < static_cast<size_t>(is.gcount()))
    {
      break;
    }

    // The fail bit may be set.  Clear it.
    is.clear(is.rdstate() & ~ios::failbit);
  }
  return haveData;
}

//----------------------------------------------------------------------------
int vtkPVEnSightMasterServerReader::ParseMasterServerFile()
{
  // Clear old list of pieces.
  this->Internal->PieceFileNames.clear();

  // Construct the file name to open.
  std::string sfilename;
  if (!this->CaseFileName)
  {
    vtkErrorMacro("A case file name must be specified.");
    return VTK_ERROR;
  }
  if (this->FilePath)
  {
    sfilename = this->FilePath;
    sfilename += this->CaseFileName;
    vtkDebugMacro("Full path to file: " << sfilename.c_str());
  }
  else
  {
    sfilename = this->CaseFileName;
  }

  // Open the file for reading.
  ifstream fin(sfilename.c_str(), ios::in);
  if (!fin)
  {
    vtkErrorMacro("Unable to open file: " << sfilename.c_str());
    return VTK_ERROR;
  }

  // A little state machine to read a sequence of lines.
  // First: Determine what kind of file this is.
  // We treat a case file like a degenerate sos file with one server.
  int readingFormat = 0;
  // Only case files have the SERVER tag.
  int readingServers = 0;
  int numServers = 0;

  // Read all data lines in the file.
  std::string line;
  while (vtkPVEnSightMasterServerReaderGetLineFromStream(fin, line))
  {
    // This section determines the type of file: case of SOS.
    if (!readingFormat && vtkPVEnSightMasterServerReaderStartsWith(line.c_str(), "FORMAT"))
    {
      // The FORMAT section starts here.
      readingFormat = 1;
    }
    else if (readingFormat && !numServers &&
      vtkPVEnSightMasterServerReaderStartsWith(line.c_str(), "type:"))
    {
      // Remove the leading "type:" to get the remaining value.
      const char* p = line.c_str();
      p += strlen("type:");
      // Remove leading spaces.
      while (*p && vtkPVEnSightMasterServerReaderIsSpace(*p))
      {
        ++p;
      }
      if (!*p)
      {
        vtkErrorMacro("Error parsing file type from: " << line.c_str());
        return VTK_ERROR;
      }
      // These are the special cases (case files).
      if (strncmp(p, "ensight", 7) == 0 || strncmp(p, "ensight gold", 12) == 0)
      {
        // Handle the case file line an sos file with one server.
        numServers = 1;
        this->Internal->PieceFileNames.push_back(this->CaseFileName);
        // We could exit the state machine here
        // because we have nothing else to do from this state.
        // We assume the line "SERVERS" will not appear in any case file.
      }
      else if (strncmp(p, "master_server gold", 18) != 0)
      { // We might as well have this sanity check here.
        vtkErrorMacro("Unexpected file type: '" << p << "'");
        return VTK_ERROR;
      }
      readingFormat = 0;
    }

    // The rest of this stuff is for picking the number of servers
    // and case file names from the SOS file.
    else if (!readingServers && vtkPVEnSightMasterServerReaderStartsWith(line.c_str(), "SERVERS"))
    {
      // The SERVERS section starts here.
      readingServers = 1;
    }
    else if (readingServers && !numServers &&
      vtkPVEnSightMasterServerReaderStartsWith(line.c_str(), "number of servers:"))
    {
      // Found the number of servers line.
      if (sscanf(line.c_str(), "number of servers: %i", &numServers) < 1)
      {
        vtkErrorMacro("Error parsing number of servers from: " << line.c_str());
        return VTK_ERROR;
      }
    }
    else if (readingServers && vtkPVEnSightMasterServerReaderStartsWith(line.c_str(), "casefile:"))
    {
      // Found a casefile line.  Record the name.
      const char* p = line.c_str();
      p += strlen("casefile:");
      while (*p && vtkPVEnSightMasterServerReaderIsSpace(*p))
      {
        ++p;
      }
      if (!*p)
      {
        vtkErrorMacro("Error parsing case file name from: " << line.c_str());
        return VTK_ERROR;
      }
      this->Internal->PieceFileNames.push_back(p);
    }
  }

  // Make sure we got the expected number of servers (pieces).
  if (static_cast<int>(this->Internal->PieceFileNames.size()) != numServers)
  {
    vtkErrorMacro("Number of servers specified by file does not match "
                  "actual number of servers provided by file.");
    return VTK_ERROR;
  }

  // The number of pieces is minimum of the number of EnSight servers
  // specified in the file and the number of processes to read the
  // data.
  int numProcs = this->Controller->GetNumberOfProcesses();

  // Make sure we have enoght processes to read all the pieces.
  if (numProcs < numServers)
  {
    vtkErrorMacro("Not enough processes (" << numProcs << ") to read all Ensight server files ("
                                           << numServers << ")");
  }

  this->NumberOfPieces = (numServers < numProcs) ? numServers : numProcs;

  return VTK_OK;
}

//----------------------------------------------------------------------------
int vtkPVEnSightMasterServerReader::CanReadFile(const char* fname)
{
  // We may have to read quite a few lines of the file to do this test
  // for real.  Just check the extension.
  size_t len = strlen(fname);
  if ((len >= 4) && (strcmp(fname + len - 4, ".sos") == 0))
  {
    return 1;
  }
  else if ((len >= 5) && (strcmp(fname + len - 5, ".case") == 0))
  {
    return 1;
  }
  return 0;
}
