OnePredict.h

Go to the documentation of this file.

// OnePredict.h: DP3 step class to predict visibilities from a source model
// Copyright (C) 2023 ASTRON (Netherlands Institute for Radio Astronomy)
// SPDX-License-Identifier: GPL-3.0-or-later
 
 
#ifndef DP3_STEPS_ONEPREDICT_H_
#define DP3_STEPS_ONEPREDICT_H_
 
#include <atomic>
 
#include <xtensor/containers/xtensor.hpp>
 
#include "ApplyBeam.h"
#include "ApplyCal.h"
#include "ResultStep.h"
#include "base/DP3.h"
#include "base/DPBuffer.h"
 
#include "base/ModelComponent.h"
#include "base/PredictBuffer.h"
#include "base/PredictModel.h"
 
#include "model/Patch.h"
#include "model/SourceDBUtil.h"
 
namespace dp3 {
namespace common {
class ParameterSet;
}
 
namespace steps {
 
class OnePredict : public ModelDataStep {
 public:
  OnePredict(const common::ParameterSet&, const std::string& prefix,
             const std::vector<std::string>& source_patterns);
 
  ~OnePredict() override;
 
  common::Fields getRequiredFields() const override {
    common::Fields fields = kUvwField;
    if (operation_ == Operation::kAdd || operation_ == Operation::kSubtract) {
      fields |= kDataField;
    }
    if (apply_cal_step_) {
      fields |= base::GetChainRequiredFields(apply_cal_step_);
    }
    return fields;
  }
 
  common::Fields getProvidedFields() const override {
    // When operation_ == "replace", the output of apply_cal_step_ is passed to
    // the next step. For all other operations, the predicted visibility data
    // is the only provided output.
    common::Fields fields;
 
    if (output_data_name_.empty()) {
      // The predicted visibilities go to the main data buffer
      fields |= kDataField;
    } else {
      // TODO(AST-1241): Handle these dependencies using Fields.
    }
 
    if (operation_ == Operation::kReplace && apply_cal_step_) {
      std::shared_ptr<Step> step = apply_cal_step_;
      do {
        fields |= step->getProvidedFields();
        step = step->getNextStep();
      } while (step);
    }
    return fields;
  }
 
  void SetApplyCal(const common::ParameterSet&, const std::string& prefix);
 
  void SetOperation(const std::string& type);
 
  void SetThreadData(std::mutex* measures_mutex) {
    measures_mutex_ = measures_mutex;
  }
 
  bool process(std::unique_ptr<base::DPBuffer>) override;
 
  void finish() override;
 
  void updateInfo(const base::DPInfo&) override;
 
  void show(std::ostream&) const override;
 
  void showTimings(std::ostream&, double duration) const override;
 
  void setSources(const std::vector<std::string>& sourcePatterns);
 
  base::Direction GetFirstDirection() const override;
 
 private:
  enum class Operation { kReplace, kAdd, kSubtract };
 
  void init(const common::ParameterSet&, const std::string& prefix,
            const std::vector<std::string>& sourcePatterns);
 
  void initializeThreadData();
  everybeam::vector3r_t dir2Itrf(const casacore::MDirection& dir,
                                 casacore::MDirection::Convert& measConverter);
 
  void addBeamToData(const model::Patch& patch, size_t buffer_index,
                     aocommon::xt::UTensor<std::complex<double>, 3>& model_data,
                     double time, bool update_beam, size_t thread,
                     aocommon::xt::UTensor<std::complex<double>, 3>& data0,
                     bool stokesIOnly);
 
  void addBeamToDataRange(
      const model::Patch& patch,
      aocommon::xt::UTensor<std::complex<double>, 3>& model_data, double time,
      size_t thread, aocommon::xt::UTensor<std::complex<double>, 3>& data0,
      const std::pair<size_t, size_t>& baseline_range,
      const std::pair<size_t, size_t>& station_range, std::barrier<>& barrier,
      bool stokesIOnly);
 
  void PredictWithSourceParallelization(base::DPBuffer::DataType& destination,
                                        double time);
  void PredictSourceRange(
      aocommon::xt::UTensor<std::complex<double>, 3>& result, size_t start,
      size_t end, size_t thread_index, std::mutex& mutex, double time,
      bool update_beam);
 
  void CopyPredictBufferToData(
      base::DPBuffer::DataType& destination,
      const aocommon::xt::UTensor<std::complex<double>, 3>& buffer);
 
  std::string name_;
  xt::xtensor<std::complex<float>, 3> input_data_;
  std::string source_db_name_;
  bool correct_time_smearing_ = false;
  bool correct_freq_smearing_ = false;
  Operation operation_;
  std::string output_data_name_;
  bool apply_beam_ = false;
  std::string coefficients_path_;
  bool use_channel_freq_ = false;
  bool one_beam_per_patch_ = false;
  bool thread_over_baselines_ = false;
  double beam_proximity_limit_ = 0.0;
  double beam_evaluation_interval_ = 0.0;
  double previous_beam_time_ = 0.0;
  bool stokes_i_only_ = false;
  bool any_orientation_is_absolute_ = false;  
  base::Direction phase_ref_;
  bool moving_phase_ref_ = false;
 
  std::shared_ptr<ApplyCal> apply_cal_step_;  
  std::shared_ptr<ResultStep> result_step_;   
 
  unsigned int debug_level_ = 0;
 
  std::vector<double> scaled_ncp_uvw_;
 
  std::vector<std::pair<size_t, size_t>> baselines_;
 
  std::vector<int> uvw_split_index_;
 
  xt::xtensor<double, 2> station_uvw_;
 
  std::shared_ptr<std::vector<base::PredictBuffer>> predict_buffers_;
  everybeam::BeamMode beam_mode_ = everybeam::BeamMode::kNone;
  everybeam::ElementResponseModel element_response_model_ =
      everybeam::ElementResponseModel::kDefault;
  std::vector<casacore::MeasFrame> meas_frame_;
  std::vector<casacore::MDirection::Convert> meas_convertors_;
  std::shared_ptr<everybeam::telescope::Telescope> telescope_;
 
  std::string direction_str_;  
  std::vector<std::shared_ptr<model::Patch>> patch_list_;
 
  std::vector<std::pair<std::shared_ptr<base::ModelComponent>,
                        std::shared_ptr<model::Patch>>>
      source_list_;
 
  common::NSTimer timer_;
 
  std::atomic<int64_t> predict_time_ = 0;
  std::atomic<int64_t> apply_beam_time_ = 0;
 
  std::mutex* measures_mutex_;
  std::mutex mutex_;
};
 
}  // namespace steps
}  // namespace dp3
 
#endif