MSWriter.h

Go to the documentation of this file.

// MSWriter.h: DP3 step writing to an MS
// Copyright (C) 2022 ASTRON (Netherlands Institute for Radio Astronomy)
// SPDX-License-Identifier: GPL-3.0-or-later
 
#ifndef DP3_STEPS_MSWRITER_H_
#define DP3_STEPS_MSWRITER_H_
 
#include "OutputStep.h"
 
#include <memory>
#include <thread>
 
#include <aocommon/lane.h>
 
#include <casacore/tables/Tables/ColumnDesc.h>
#include <casacore/tables/Tables/ScalarColumn.h>
#include <casacore/tables/Tables/ArrayColumn.h>
 
#include "base/StManParsetKeys.h"
#include "common/Timer.h"
 
namespace casacore {
class Table;
}
 
namespace dp3 {
namespace common {
class ParameterSet;
}
 
namespace steps {
class InputStep;
 
std::unique_ptr<casacore::DataManager> MakeStMan(
    const std::string& type_name, const std::string& instance_name,
    const casacore::Record& record = casacore::Record());
 
 
 
class MSWriter : public OutputStep {
 public:
  explicit MSWriter(const std::string& out_name, const common::ParameterSet&,
                    const std::string& prefix);
 
  ~MSWriter() override;
 
  common::Fields getRequiredFields() const override {
    return kDataField | kFlagsField | kWeightsField | kUvwField;
  }
 
  bool process(std::unique_ptr<base::DPBuffer> buffer) override;
 
  void finish() override;
 
  void show(std::ostream&) const override;
 
  void updateInfo(const base::DPInfo&) override;
 
  void showTimings(std::ostream&, double duration) const override;
 
  static void WriteHistory(casacore::Table& ms,
                           const common::ParameterSet& parset);
 
  static void UpdateBeam(casacore::Table& main_table,
                         const std::string& out_col_name,
                         const base::DPInfo& info);
 
  static void UpdatePhaseCentre(const std::string& out_name,
                                const casacore::MDirection& new_phase_dir);
 
  static void UpdateSpw(const std::string& out_name, const base::DPInfo& info);
 
  static void UpdateObs(const std::string& out_name, const base::DPInfo& info);
 
  static std::string InsertNumberInFilename(const std::string& name,
                                            size_t number);
 
 private:
  void StartNewMs();
  void FinishMs();
 
  void CreateMs(const std::string& out_name, unsigned int tile_size,
                unsigned int tile_n_chan);
 
  void CopySubTables(casacore::Table& original_table);
 
  void ProcessBuffer(base::DPBuffer& buffer);
 
  void WriteData(casacore::Table& out, base::DPBuffer& buf);
 
  void WriteMeta(casacore::Table& out, const base::DPBuffer& buf);
 
  void CopyMeta(const casacore::Table& in, casacore::Table& out,
                bool copy_time_info);
 
  template <typename T>
  void FillSca(const T& value, casacore::Table& out,
               const casacore::String& column_name) {
    casacore::ScalarColumn<T> out_col(out, column_name);
    out_col.fillColumn(value);
  }
 
  template <typename T>
  void FillArr(const casacore::Array<T>& value, casacore::Table& out,
               const casacore::String& column_name) {
    casacore::ArrayColumn<T> out_col(out, column_name);
    out_col.fillColumn(value);
  }
 
  template <typename T>
  void CopySca(const casacore::Table& in, casacore::Table& out,
               const casacore::String& column_name) {
    casacore::ROScalarColumn<T> in_col(in, column_name);
    casacore::ScalarColumn<T> out_col(out, column_name);
    out_col.putColumn(in_col.getColumn());
  }
 
  template <typename T>
  void CopyArr(const casacore::Table& in, casacore::Table& out,
               const casacore::String& column_name) {
    casacore::ROArrayColumn<T> in_col(in, column_name);
    casacore::ArrayColumn<T> out_col(out, column_name);
    out_col.putColumn(in_col.getColumn());
  }
 
  std::string name_;
  std::string out_name_;
  std::string chunk_name_;
  casacore::Table ms_;
  common::ParameterSet parset_;  
  casacore::String data_col_name_;
  casacore::String flag_col_name_;
  casacore::String weight_col_name_;
  bool overwrite_;  
  bool copy_corr_data_;
  bool copy_model_data_;
  unsigned int tile_size_;
  unsigned int tile_n_chan_;
  unsigned int nr_times_flush_;  
  unsigned int nr_done_;         
  double chunk_duration_ = 0.0;
  double chunk_start_time_ = 0.0;
  size_t current_chunk_index_ = 0;
 
  std::string vds_dir_;       
  std::string cluster_desc_;  
  base::StManParsetKeys st_man_keys_;
  // For now, metadata compression is turned on only when explicitly turned on
  // in compilation. Once we're confident enough about this feature, we can
  // enable it by default.
  bool scalar_flags_ = METADATA_COMPRESSION_DEFAULT;
  bool uvw_compression_ = METADATA_COMPRESSION_DEFAULT;
  bool antenna_compression_ = METADATA_COMPRESSION_DEFAULT;
 
  common::NSTimer timer_;
 
  common::NSTimer writer_timer_;
 
  common::NSTimer create_task_timer_;
 
  aocommon::Lane<std::unique_ptr<base::DPBuffer>> write_queue_{3};
 
  void CreateTask(std::unique_ptr<base::DPBuffer> buffer);
 
  std::thread write_queue_thread_;
 
  void WriteQueueProcess();
 
  bool use_write_thread_{false};
 
  bool is_write_thread_active_{false};
 
  void StopWriteThread();
};
 
}  // namespace steps
}  // namespace dp3
 
#endif