Change member to private inheritance.

This is the first commit to generate correct, readable Parquet files.

Outputs/ObservationPlaneWriterParquet.h

@@ -15,7 +15,7 @@
 
 namespace corsika::output {
 
-  class ObservationPlaneWriterParquet : public BaseOutput {
+  class ObservationPlaneWriterParquet : public BaseOutput, private ParquetStreamer {
 
   public:
     /**
@@ -24,7 +24,8 @@ namespace corsika::output {
      * @param name    The name of this output.
      */
     ObservationPlaneWriterParquet(std::string const& name)
-        : name_(name)
+        : ParquetStreamer()
+        , name_(name)
         , event_(0){};
 
     /**
@@ -33,20 +34,21 @@ namespace corsika::output {
     void StartOfRun(std::filesystem::path const& directory) final {
 
       // setup the streamer
-      streamer_.Init((directory / "particles.parquet").string());
+      InitStreamer((directory / "particles.parquet").string());
 
       // build the schema
-      streamer_.AddField("event", parquet::Repetition::REQUIRED, parquet::Type::INT32,
-                         parquet::ConvertedType::INT_32);
-      streamer_.AddField("pdg", parquet::Repetition::REQUIRED, parquet::Type::INT32,
-                         parquet::ConvertedType::INT_32);
-      streamer_.AddField("energy", parquet::Repetition::REQUIRED, parquet::Type::FLOAT,
-                         parquet::ConvertedType::NONE);
-      streamer_.AddField("radius", parquet::Repetition::REQUIRED, parquet::Type::FLOAT,
-                         parquet::ConvertedType::NONE);
+      AddField("event", parquet::Repetition::REQUIRED, parquet::Type::INT32,
+               parquet::ConvertedType::INT_32);
+      AddField("pdg", parquet::Repetition::REQUIRED, parquet::Type::INT32,
+               parquet::ConvertedType::INT_32);
+      AddField("energy", parquet::Repetition::REQUIRED, parquet::Type::DOUBLE,
+               parquet::ConvertedType::NONE);
+      AddField("radius", parquet::Repetition::REQUIRED, parquet::Type::DOUBLE,
+               parquet::ConvertedType::NONE);
 
       // and build the streamer
-      streamer_.Build();
+      BuildStreamer();
+
     }
 
     /**
@@ -62,7 +64,7 @@ namespace corsika::output {
     /**
      * Called at the end of each run.
      */
-    void EndOfRun() final { streamer_.Close(); }
+    void EndOfRun() final { writer_.reset(); outfile_->Close(); }
 
     /**
      * Get final text outputs for the config file.
@@ -75,20 +77,17 @@ namespace corsika::output {
      */
     void Write(particles::Code const& pid, units::si::HEPEnergyType const& energy,
                units::si::LengthType const& distance) {
-
       using namespace units::si;
 
       // write the next row
-      writer_ << event_ << static_cast<int>(particles::GetPDG(pid)) << energy / 1_eV
-              << distance / 1_m << parquet::EndRow;
+      (*writer_) << event_ << static_cast<int>(particles::GetPDG(pid)) << energy / 1_eV
+                 << distance / 1_m << parquet::EndRow;
     }
 
     std::string const name_; ///< The name of this output.
 
   private:
-    int event_;                    ///< The current event number we are processing.
-    ParquetStreamer streamer_;     ///< A parquet stream writer helper
-    parquet::StreamWriter writer_; ///< The writer for this file.
+    int event_; ///< The current event number we are processing.
 
   }; // class ObservationPlaneWriterParquet
 

Outputs/OutputManager.h

@@ -21,6 +21,9 @@ namespace corsika::output {
    */
   class OutputManager final {
 
+    /**
+     * Indicates the current state of this manager.
+     */
     enum class OutputState {
       RunNoInit,
       RunInitialized,
@@ -31,7 +34,7 @@ namespace corsika::output {
     OutputState state_{OutputState::RunNoInit}; ///< The current state of this manager.
     std::string const name_;                    ///< The name of this simulation file.
     std::filesystem::path const root_; ///< The top-level directory for the output.
-    inline static auto logger{logging::GetLogger("output_manager")}; ///< A custom logger.
+    inline static auto logger{logging::GetLogger("output")}; ///< A custom logger.
 
     /**
      * The outputs that have been registered with us.
@@ -156,7 +159,7 @@ namespace corsika::output {
     /**
      * Called at the start of each run.
      */
-      void StartOfRun() {
+    void StartOfRun() {
       for (auto& [name, output] : outputs_) {
 
         // construct the path to this output subdirectory
@@ -173,7 +176,7 @@ namespace corsika::output {
     /**
      * Called at the start of each event/shower.
      */
-      void StartOfEvent() {
+    void StartOfEvent() {
 
       // if this is called but we are still in the initialized state,
       // make sure that we transition to RunInProgress
@@ -186,7 +189,7 @@ namespace corsika::output {
     /**
      * Called at the end of each event/shower.
      */
-      void EndOfEvent() {
+    void EndOfEvent() {
 
       for (auto& [name, output] : outputs_) { output.get().EndOfEvent(); }
     }
@@ -195,10 +198,15 @@ namespace corsika::output {
      * Called at the end of each run.
      */
     void EndOfRun() {
-      for (auto& [name, output] : outputs_) { output.get().EndOfRun(); }
+      for (auto& [name, output] : outputs_) {
+        output.get().EndOfRun();
+      }
 
       // and the run has finished
       state_ = OutputState::RunFinished;
+
+      // write any final state information into the config files
+      // for (auto& [name, output] : outputs_) { output.get().EndOfRun(); }
     }
 
   }; // class OutputManager

Outputs/ParquetStreamer.h

 /*
- * (c) Copyright 2019 CORSIKA Project, corsika-project@lists.kit.edu
+ * (c) Copyright 2020 CORSIKA Project, corsika-project@lists.kit.edu
  *
  * This software is distributed under the terms of the GNU General Public
  * Licence version 3 (GPL Version 3). See file LICENSE for a full version of
@@ -23,12 +23,15 @@ namespace corsika::output {
    * This class automates the construction of simple tabular
    * Parquet files using the parquet::StreamWriter.
    */
-  class ParquetStreamer final {
+  class ParquetStreamer {
 
   public:
-    ParquetStreamer() {}
+    ParquetStreamer() = default;
 
-    void Init(std::string const& filepath) {
+    /**
+     * Initialize the streamer to write to a given file.
+     */
+    void InitStreamer(std::string const& filepath) {
 
       // open the file and connect it to our pointer
       PARQUET_ASSIGN_OR_THROW(outfile_, arrow::io::FileOutputStream::Open(filepath));
@@ -42,40 +45,29 @@ namespace corsika::output {
      */
     template <typename... TArgs>
     void AddField(TArgs&&... args) {
-      nodes_.push_back(parquet::schema::PrimitiveNode::Make(args...));
+      fields_.push_back(parquet::schema::PrimitiveNode::Make(args...));
     }
 
     /**
      * Finalize the streamer construction.
      */
-    void Build() {
+    void BuildStreamer() {
 
       // build the top level schema
-      auto schema = std::static_pointer_cast<parquet::schema::GroupNode>(
+      schema_ = std::static_pointer_cast<parquet::schema::GroupNode>(
           parquet::schema::GroupNode::Make("schema", parquet::Repetition::REQUIRED,
-                                           nodes_));
+                                           fields_));
 
       // and build the writer
-      writer_ = parquet::StreamWriter(
-          parquet::ParquetFileWriter::Open(outfile_, schema, builder_.build()));
+      writer_ = std::make_shared<parquet::StreamWriter>(
+          parquet::ParquetFileWriter::Open(outfile_, schema_, builder_.build()));
     }
 
-    /**
-     * Get a reference to the writer for this stream.
-     */
-    parquet::StreamWriter& GetWriter() { return writer_; }
-
-    /**
-     * Close the file.
-     */
-    void Close() { outfile_->Close(); }
-
-    ///
-  private:
-    parquet::StreamWriter writer_;
-    parquet::StreamWriter stream_;               ///< The stream writer to 'outfile'
+  protected:
+    std::shared_ptr<parquet::StreamWriter> writer_;               ///< The stream writer to 'outfile'
     parquet::WriterProperties::Builder builder_; ///< The writer properties builder.
-    parquet::schema::NodeVector nodes_;
+    parquet::schema::NodeVector fields_;         ///< The fields in this file.
+    std::shared_ptr<parquet::schema::GroupNode> schema_;   ///< The schema for this file.
     std::shared_ptr<arrow::io::FileOutputStream> outfile_; ///< The output file.
 
   }; // class ParquetHelper

Processes/ObservationPlane/ObservationPlane.h

@@ -35,10 +35,10 @@ namespace corsika::process::observation_plane {
         : TOutputWriter(args...)
         , plane_(plane)
         , deleteOnHit_(deleteOnHit)
-        , energy_ground_(0_GeV)
+        , energy_ground_(units::si::HEPEnergyType::zero())
         , count_ground_(0) {}
 
-    process::EProcessReturn DoContinuous(setup::Stack::ParticleType const& particle,
+    process::EProcessReturn DoContinuous(setup::Stack::ParticleType& particle,
                                          setup::Trajectory const& trajectory) {
 
       using namespace units::si;
@@ -53,15 +53,23 @@ namespace corsika::process::observation_plane {
         return process::EProcessReturn::eOk;
       }
 
+      const auto energy = particle.GetEnergy();
+
       // write the data to the output
-      this->Write(particle.GetPID(), particle.GetEnergy(),
+      this->Write(particle.GetPID(), energy,
                   (trajectory.GetPosition(1) - plane_.GetCenter()).norm());
 
-      if (deleteOnHit_) { return process::EProcessReturn::eParticleAbsorbed; }
-      return process::EProcessReturn::eOk;
+      if (deleteOnHit_) {
+        count_ground_++;
+        energy_ground_ += energy;
+        particle.Delete();
+        return process::EProcessReturn::eParticleAbsorbed;
+      } else {
+        return process::EProcessReturn::eOk;
+      }
     }
 
-    units::si::LengthType MaxStepLength(setup::Stack::ParticleType const&,
+    units::si::LengthType MaxStepLength(setup::Stack::ParticleType& particle,
                                         setup::Trajectory const& trajectory) {
 
       using namespace units::si;
@@ -74,7 +82,9 @@ namespace corsika::process::observation_plane {
       }
 
       auto const pointOfIntersection = trajectory.GetPosition(timeOfIntersection);
-      return (trajectory.GetR0() - pointOfIntersection).norm() * 1.0001;
+      auto dist = (trajectory.GetR0() - pointOfIntersection).norm() * 1.0001;
+      C8LOG_TRACE("ObservationPlane::MaxStepLength l={} m", dist / 1_m);
+      return dist;
     }
 
     YAML::Node GetConfig() const {
@@ -105,20 +115,22 @@ namespace corsika::process::observation_plane {
       return node;
     }
 
-void ObservationPlane::ShowResults() const {
-  C8LOG_INFO(
-      " ******************************\n"
-      " ObservationPlane: \n"
-      " energy in ground (GeV)     :  {}\n"
-      " no. of particles in ground :  {}\n"
-      " ******************************",
-      energy_ground_ / 1_GeV, count_ground_);
-}
-
-void ObservationPlane::Reset() {
-  energy_ground_ = 0_GeV;
-  count_ground_ = 0;
-}
+    void ShowResults() const {
+      using namespace units::si;
+      C8LOG_INFO(
+          " ******************************\n"
+          " ObservationPlane: \n"
+          " energy in ground (GeV)     :  {}\n"
+          " no. of particles in ground :  {}\n"
+          " ******************************",
+          energy_ground_ / 1_GeV, count_ground_);
+    }
+
+    void Reset() {
+      using namespace units::si;
+      energy_ground_ = 0_GeV;
+      count_ground_ = 0;
+    }
     corsika::units::si::HEPEnergyType GetEnergyGround() const { return energy_ground_; }
 
   private: