diff --git a/components/omega/configs/Default.yml b/components/omega/configs/Default.yml index 9db42e178466..41ba8c98d4d3 100644 --- a/components/omega/configs/Default.yml +++ b/components/omega/configs/Default.yml @@ -181,3 +181,16 @@ Omega: EndTime: 0001-07-31_00:00:00 Contents: - Tracers + Analysis: + GlobalStats: + Enable: true + Fields: ["NormalVelocity", "PseudoThickness", "Temperature", "Salinity"] + SpatialStats: ["Max", "Min", "Mean", "StdDev"] + ReductionPeriod: ["1Day", "1Month"] + SampleFreq: ["6Hours", "2Days"] + Filename: global.stats.$Y + Stream: + FileFreq: 1 + FileFreqUnits: years + CustomGroup: + Enable: false diff --git a/components/omega/src/CMakeLists.txt b/components/omega/src/CMakeLists.txt index 4ea242099586..d9bd109ff707 100644 --- a/components/omega/src/CMakeLists.txt +++ b/components/omega/src/CMakeLists.txt @@ -6,6 +6,7 @@ add_library(OmegaLibFlags INTERFACE) target_include_directories( OmegaLibFlags INTERFACE + ${OMEGA_SOURCE_DIR}/src/analysis ${OMEGA_SOURCE_DIR}/src/base ${OMEGA_SOURCE_DIR}/src/infra ${OMEGA_SOURCE_DIR}/src/ocn @@ -72,7 +73,7 @@ if(GKlib_FOUND) endif() # Add source files for the library -file(GLOB_RECURSE _LIBSRC_FILES infra/*.cpp base/*.cpp ocn/*.cpp timeStepping/*.cpp) +file(GLOB_RECURSE _LIBSRC_FILES analysis/*.cpp infra/*.cpp base/*.cpp ocn/*.cpp timeStepping/*.cpp) add_library(${OMEGA_LIB_NAME} ${_LIBSRC_FILES}) diff --git a/components/omega/src/analysis/Analysis.cpp b/components/omega/src/analysis/Analysis.cpp new file mode 100644 index 000000000000..12819803f245 --- /dev/null +++ b/components/omega/src/analysis/Analysis.cpp @@ -0,0 +1,504 @@ +//===-- analysis/Analysis.cpp - analysis module -----------------*- C++ -*-===// +// +// The Analysis module provides in-situ computation of analysis fields from +// the ocean model state during simulation runtime. This implementation file +// contains the core orchestration logic including operator chain parsing, +// dependency graph construction, alarm-based scheduling, and recursive +// computation with caching. +// +//===----------------------------------------------------------------------===// + +#include "Analysis.h" +#include "IOStream.h" +#include "TimeStepper.h" +#include "analysisGroups/Groups.h" + +namespace OMEGA { + +// Create the static class members +Analysis *Analysis::DefAnalysis = nullptr; +std::map> Analysis::AllAnalysisObjects; + +//------------------------------------------------------------------------------ +// Parses a frequency string like "1day" or "6hour" into numeric and unit +// components. Returns a vector with [0] = numeric part, [1] = units part. +// Appends 's' to units if not already present for consistency with +// TimeInterval. +std::vector parseFreqStr(const std::string &FreqStr) { + + std::string DigitStr; + std::string UnitsStr; + + // Find where digits end and units begin + size_t Pos = FreqStr.find_first_not_of("0123456789"); + if (Pos != std::string::npos) { + DigitStr = FreqStr.substr(0, Pos); + UnitsStr = FreqStr.substr(Pos); + } + + // Validate that we found both components + if (DigitStr.empty() || UnitsStr.empty()) { + ABORT_ERROR("Analysis: Invalid frequency string found in Config: {}", + FreqStr); + } + + // Ensure units end with 's' for consistency with TimeInterval + if (UnitsStr.back() != 's') { + UnitsStr += 's'; + } + + return {DigitStr, UnitsStr}; + +} // end parseFreqStr + +//------------------------------------------------------------------------------ +// Initializes the Analysis module by registering all base operator types +// with the factory, then creating the default Analysis instance. This function +// must be called after HorzMesh, VertCoord, and TimeStepper have been +// initialized, as it retrieves pointers to their default instances. +void Analysis::init() { + + // Register all built-in operators with the factory + registerAllBaseAnalysisOperators(); + + // Retrieve default instances of required components + auto DefEnv = MachEnv::getDefault(); + auto Mesh = HorzMesh::getDefault(); + auto VCoord = VertCoord::getDefault(); + auto DefTimeStepper = TimeStepper::getDefault(); + Clock *OmegaClock = DefTimeStepper->getClock(); + Config *OmegaConfig = Config::getOmegaConfig(); + + // Create the default Analysis instance + Analysis::DefAnalysis = + create("Default", DefEnv, Mesh, VCoord, OmegaClock, OmegaConfig); + +} // end init + +//------------------------------------------------------------------------------ +// Creates a new named Analysis instance using the constructor and registers +// it in the AllAnalysisObjects map. Returns a pointer to the new instance, +// or nullptr if an Analysis with this name already exists. +Analysis *Analysis::create(const std::string &Name, const MachEnv *Env, + const HorzMesh *Mesh, const VertCoord *VCoord, + Clock *ModelClock, Config *Options) { + + // Check for duplicate names + if (AllAnalysisObjects.find(Name) != AllAnalysisObjects.end()) { + LOG_ERROR("Attempted to create an Analysis with name {} " + "but one with that name already exists", + Name); + return nullptr; + } + + // Create new Analysis instance via constructor + auto NewAnalysis = + new Analysis(Name, Env, Mesh, VCoord, ModelClock, Options); + + // Store in map and return pointer + AllAnalysisObjects.emplace(Name, NewAnalysis); + + return NewAnalysis; + +} // end create + +//------------------------------------------------------------------------------ +// Constructs a new Analysis instance by reading the Analysis configuration +// node, instantiating enabled AnalysisGroup objects, building the operator +// dependency graph, setting up alarms, and initializing all operators. +// Pre-defined AnalysisGroup types (e.g., GlobalStats) are dispatched by name; +// user-defined custom groups will be supported in future versions. +Analysis::Analysis(const std::string &InName, const MachEnv *InEnv, + const HorzMesh *InMesh, const VertCoord *InVCoord, + Clock *InModelClock, Config *Options) { + + // Store member variables + Name = InName; + Env = InEnv; + Mesh = InMesh; + VCoord = InVCoord; + ModelClock = InModelClock; + + // Set up name prefix for operator and stream names + std::string NamePrefix = Name + "_"; + if (Name == "Default") { + NamePrefix = ""; + } + + // Read the Analysis configuration node + Error Err; + Config AnalysisCfg("Analysis"); + Err += Options->get(AnalysisCfg); + CHECK_ERROR_ABORT(Err, "Analysis: Analysis group not in Config"); + + // Loop over all AnalysisGroup nodes in the config + for (Config::Iter It = AnalysisCfg.begin(); It != AnalysisCfg.end(); ++It) { + std::string GroupName; + AnalysisCfg.getName(It, GroupName); + Config GroupCfg(GroupName); + Err += AnalysisCfg.get(GroupCfg); + + // Check if this group is enabled + bool GroupEnabled = false; + GroupCfg.get("Enable", GroupEnabled); + + if (GroupEnabled) { + // Dispatch to pre-defined AnalysisGroup constructors by name + if (GroupName == "GlobalStats") { + GlobalStats GlobalStatsGroup(NamePrefix + GroupName, GroupCfg, + this); + continue; + } + + // User-defined custom groups not yet supported + ABORT_ERROR("Analysis: custom analysis group enabled in config, but " + "composable operators are not yet supported."); + } + } + + // Build the operator dependency graph + buildOperatorDependencies(); + + // Set up alarm-based scheduling + setComputeAlarms(); + + // Initialize all operators with mesh/env pointers + initializeAllOps(); + +} // end constructor + +//------------------------------------------------------------------------------ +// Parses an underscore-delimited operator chain string and instantiates +// operators for each node in the chain. For example, +// "Temperature_SpatialMean_TimeMean1day" parses into three operators. +// If an intermediate operator already exists (shared by another chain), +// it is reused rather than duplicated. This natural sharing mechanism +// avoids redundant computation of common intermediate results. +void Analysis::parseChainAndBuildOps(const std::string &OpChainStr) { + + // Split the chain string on underscore delimiters + std::vector ChainVec; + std::stringstream OpChainSS(OpChainStr); + std::string Part; + while (std::getline(OpChainSS, Part, '_')) { + ChainVec.push_back(Part); + } + + // Build the chain left-to-right, checking at each step if the + // operator already exists (via Field registry). If not, create it. + std::string CurChainStr; + for (const auto &ChainNode : ChainVec) { + std::string Upstream = CurChainStr; + + // Build up the current chain string progressively + if (CurChainStr.empty()) { + CurChainStr = ChainNode; + } else { + CurChainStr += ("_" + ChainNode); + } + + // Check if this operator already exists by looking for its output Field + if (!Field::exists(CurChainStr)) { + + // Spatial operators (SpatialMean, SpatialMax, etc.) + if (ChainNode.find("Spatial") != std::string::npos) { + registerAnalysisOp(ChainNode, {Upstream}, makeOpConfig()); + continue; + } + + // Temporal operators (TimeMean, etc.) with period embedded in name + if (ChainNode.find("Time") != std::string::npos) { + // Extract operator type and period string (e.g., "TimeMean1day") + std::size_t Pos = ChainNode.find_first_of("0123456789"); + if (Pos == std::string::npos) { + ABORT_ERROR("Analysis: Improper temporal window string, {}", + ChainNode); + } + std::string TimeOp = ChainNode.substr(0, Pos); + std::string FreqStr = ChainNode.substr(Pos); + registerAnalysisOp(TimeOp, {Upstream}, + makeOpConfig(opParam("Period", FreqStr))); + continue; + } + ABORT_ERROR("Analysis: Error trying to parse {}. No Field or " + "Operator named {}", + OpChainStr, ChainNode); + } + } + +} // end parseChainAndBuildOps + +//------------------------------------------------------------------------------ +// Instantiates a single operator via the AnalysisOpFactory and wraps it in +// an OperatorNode. The factory inspects the primary upstream Field's metadata +// (scalar type, rank, memory location) to select the correct templated +// specialization of the operator. The new node is appended to the OpNodes +// vector. +void Analysis::registerAnalysisOp(const std::string &OpName, + const std::vector &UpstreamNames, + Config Options) { + + // Create the operator via factory (type-dispatched based on Field metadata) + auto NewOp = AnalysisOpFactory::createOp(OpName, UpstreamNames, Options); + + if (NewOp) { + // Wrap in an OperatorNode and add to the graph + auto Node = std::make_unique(); + Node->Op = std::move(NewOp); + OpNodes.push_back(std::move(Node)); + } + +} // end registerAnalysisOp + +//------------------------------------------------------------------------------ +// Performs post-hoc dependency resolution by iterating over all operator nodes +// and matching each node's input field names against other nodes' output field +// names. When a match is found, adds a pointer to the Upstreams vector, +// forming the edges of the dependency graph. Avoids adding duplicate upstream +// pointers. In future versions, this will be replaced by signature-based +// deduplication during graph construction. +void Analysis::buildOperatorDependencies() { + // For each operator node, find its upstream dependencies + for (auto &Node : OpNodes) { + auto InputNames = Node->Op->getInputFieldNames(); + + // For each required input field + for (const auto &InputName : InputNames) { + // Search all other nodes for matching output + for (auto &PotentialUpstream : OpNodes) { + auto UpstreamOutputs = PotentialUpstream->Op->getOutputFieldNames(); + + for (const auto &UpstreamOutput : UpstreamOutputs) { + if (UpstreamOutput == InputName) { + // Found a match - check if already in Upstreams vector + if (std::find(Node->Upstreams.begin(), Node->Upstreams.end(), + PotentialUpstream.get()) == + Node->Upstreams.end()) { + // Not found, so add the upstream dependency + Node->Upstreams.push_back(PotentialUpstream.get()); + } + break; + } + } + } + } + } + +} // end buildOperatorDependencies + +//------------------------------------------------------------------------------ +// Sets ComputeAlarms on terminal operator nodes and propagates them upstream. +// Terminal nodes that write to output streams receive alarm pointers from +// those streams. Temporal reduction operators receive two alarms: an +// accumulation alarm (owned by Analysis) that triggers sample collection, +// and an output alarm (from the stream) that triggers finalization and write. +// Discrete-sampling operators receive only the stream's output alarm. +void Analysis::setComputeAlarms() { + + // Get model timestep and current time for alarm initialization + TimeInterval Timestep = ModelClock->getTimeStep(); + TimeInstant CurrentTime = ModelClock->getCurrentTime(); + + // Set alarms on terminal operators (those that write to output streams) + for (auto &Node : OpNodes) { + std::string OpType = Node->Op->getOperatorType(); + bool IsTimeReduction = (OpType.find("Time") != std::string::npos); + + if (!Node->StreamNames.empty()) { + // This is a terminal operator that writes to one or more streams + + if (IsTimeReduction) { + // Temporal reduction operators need two alarms: + // 1. Accumulation alarm (owned by Analysis) - triggers sample + // collection + auto AccumulationAlarm = std::make_unique( + "Compute_" + OpType, Timestep, CurrentTime); + + // Attach alarm to clock and store pointer + ModelClock->attachAlarm(AccumulationAlarm.get()); + Node->ComputeAlarms.push_back(AccumulationAlarm.get()); + AccumulationAlarms.push_back(std::move(AccumulationAlarm)); + + // 2. Output alarm (from stream) - triggers finalization and write + // Time reduction operators have exactly one associated stream + auto StreamAlarm = IOStream::getAlarm(Node->StreamNames[0]); + Node->ComputeAlarms.push_back(StreamAlarm); + + // Tell the operator which alarm triggers period finalization + Node->Op->setPeriodAlarm(StreamAlarm); + + } else { + // Discrete sampling operators: borrow alarm pointers from streams + for (const auto &StreamName : Node->StreamNames) { + auto StreamAlarm = IOStream::getAlarm(StreamName); + + // Avoid adding duplicate alarm pointers + if (std::find(Node->ComputeAlarms.begin(), + Node->ComputeAlarms.end(), + StreamAlarm) == Node->ComputeAlarms.end()) { + Node->ComputeAlarms.push_back(StreamAlarm); + } + } + } + } + } + + // Propagate alarms from terminal nodes to all upstream dependencies + propagateAlarmsUpstream(); + +} // end setComputeAlarms + +//------------------------------------------------------------------------------ +// Iteratively propagates alarm pointers from downstream operators to upstream +// operators. An upstream operator must be computed whenever any of its +// downstream consumers needs fresh data, so each downstream alarm is added +// to the upstream's ComputeAlarms vector. Iteration continues until no +// further changes occur (fixed point algorithm). +void Analysis::propagateAlarmsUpstream() { + + // Iterate until no changes occur (fixed point) + bool Changed = true; + while (Changed) { + Changed = false; + + // For each node, find all downstream operators + for (auto &Node : OpNodes) { + for (const auto &OtherNode : OpNodes) { + // Check if Node is in OtherNode's upstream list + for (const auto *Upstream : OtherNode->Upstreams) { + if (Upstream == Node.get()) { + // Node is upstream of OtherNode, so add OtherNode's alarms + for (auto *DownstreamAlarm : OtherNode->ComputeAlarms) { + // Only add if not already present + if (std::find(Node->ComputeAlarms.begin(), + Node->ComputeAlarms.end(), + DownstreamAlarm) == + Node->ComputeAlarms.end()) { + Node->ComputeAlarms.push_back(DownstreamAlarm); + Changed = true; + } + } + } + } + } + } + } + +} // end propagateAlarmsUpstream + +//------------------------------------------------------------------------------ +// Main computational loop called once per timestep. Iterates over all +// operator nodes and checks if any of their alarms are ringing. If so, +// calls computeRecursive to compute the operator and all its upstream +// dependencies, using cache validation to avoid redundant work. +void Analysis::computeAll() { + + TimeInstant CurrentTime = ModelClock->getCurrentTime(); + + // Check each operator node for ringing alarms + for (auto &Node : OpNodes) { + bool ShouldCompute = false; + + // Check if any of this node's alarms are ringing + for (auto *Alarm : Node->ComputeAlarms) { + if (Alarm->isRinging()) { + ShouldCompute = true; + break; + } + } + + // Trigger recursive computation if any alarm is ringing + if (ShouldCompute) { + computeRecursive(Node.get(), CurrentTime); + } + } + +} // end computeAll + +//------------------------------------------------------------------------------ +// Recursively computes an operator node and all its upstream dependencies, +// using timestamp-based cache validation to prevent redundant computation. +// If the node's output is already valid for this timestamp (cache hit), +// returns immediately. Otherwise, recursively computes all upstream +// dependencies first, then calls the node's compute() method. +void Analysis::computeRecursive(OperatorNode *Node, + const TimeInstant &TimeStamp) { + + // Check cache: if already computed for this timestamp, return immediately + if (Node->Op->isCacheValid(TimeStamp)) { + return; + } + + // Recursively compute all upstream dependencies first (depth-first) + for (auto *Upstream : Node->Upstreams) { + computeRecursive(Upstream, TimeStamp); + } + + // Now all inputs are fresh - compute this operator + Node->Op->compute(TimeStamp); + +} // end computeRecursive + +//------------------------------------------------------------------------------ +// Calls initialize() on all operators after the dependency graph is complete +// and all Fields exist. This allows operators to store pointers to mesh, +// environment, and other resources they need during compute() calls. +void Analysis::initializeAllOps() { + + // Initialize each operator with context pointers + for (auto &OpNode : OpNodes) { + OpNode->Op->initialize(Env, Mesh, VCoord, makeOpConfig()); + } + +} // end initializeAllOps + +//------------------------------------------------------------------------------ +// Returns a reference to the ModelClock pointer. This is used by +// AnalysisGroup instances during stream creation to access the clock +// for alarm initialization. +Clock *&Analysis::getModelClock() { return ModelClock; } // end getModelClock + +//------------------------------------------------------------------------------ +// Returns a vector of non-owning pointers to all registered operator nodes. +// Used primarily for testing and validation of the dependency graph. +const std::vector Analysis::getOpNodes() { + std::vector OpPtrs; + for (auto &Node : OpNodes) { + OpPtrs.push_back(Node.get()); + } + return OpPtrs; +} // end getOpNodes + +//------------------------------------------------------------------------------ +// Checks whether an operator node with the given full instance name already +// exists in the OpNodes vector. Returns true if found, false otherwise. +// Used during chain parsing to avoid creating duplicate operators. +bool Analysis::OpNodeExists(const std::string &FullOpName) { + + for (const auto &Node : OpNodes) { + if (FullOpName == Node->Op->getName()) { + return true; + } + } + + return false; + +} // end OpNodeExists + +//------------------------------------------------------------------------------ +// Retrieves the default Analysis instance +Analysis *Analysis::getDefault() { return DefAnalysis; } // end getDefault + +// Removes all defined Analysis instances and cleans up static resources +void Analysis::finalize() { + DefAnalysis = nullptr; + AllAnalysisObjects.clear(); +} // end finalize + +//------------------------------------------------------------------------------ +// Destructor - deallocates all memory +Analysis::~Analysis() {} + +} // end namespace OMEGA + +//===----------------------------------------------------------------------===// diff --git a/components/omega/src/analysis/Analysis.h b/components/omega/src/analysis/Analysis.h new file mode 100644 index 000000000000..b7c4a3bb773d --- /dev/null +++ b/components/omega/src/analysis/Analysis.h @@ -0,0 +1,252 @@ +#ifndef OMEGA_ANALYSIS_H +#define OMEGA_ANALYSIS_H + +//===-- analysis/Analysis.h - OMEGA Analysis ----------*- C++ -*-===// +// +/// \file +/// \brief Defines core Analysis framework for in-situ computation +/// +/// The Analysis module provides in-situ computation of analysis fields from +/// the ocean model state during simulation runtime. Analysis fields are +/// computed on-the-fly and written to output streams at user-specified +/// intervals, providing an alternative to extensive offline +/// post-processing. +/// +/// The framework is built on a composable operator architecture where +/// operators can be chained together to produce analysis outputs. Each +/// operator performs a single, well-defined transformation (e.g., spatial +/// reduction, temporal averaging, binary operations). Operators declare +/// their input field dependencies at construction, and the Analysis +/// orchestrator resolves dependencies to form a directed acyclic graph +/// (DAG). +/// +/// The orchestrator uses an alarm-based scheduling model to trigger +/// operator computation. Each operator node contains pointers to one or +/// more alarms; when any alarm rings, the operator computes its output. +/// Upstream dependencies are computed recursively on-demand, with +/// timestamp-based caching to prevent redundant computation when multiple +/// downstream operators share an intermediate result. +/// +/// For detailed design documentation including algorithmic formulation, +/// dependency resolution, and operator registration, see +/// components/omega/doc/design/Analysis.md +/// +//===----------------------------------------------------------------------===// + +#include "AnalysisOpFactory.h" +#include "AnalysisOperator.h" +#include "Config.h" +#include "DataTypes.h" +#include "Dimension.h" +#include "Error.h" +#include "Field.h" +#include "HorzMesh.h" +#include "Logging.h" +#include "MachEnv.h" +#include "TimeMgr.h" +#include "VertCoord.h" +#include "operators/Ops.h" + +#include +#include +#include +#include + +namespace OMEGA { + +/// Parses a frequency string into numeric and unit components +/// Frequency strings are of the form "1day", "6hour", "1month", etc. +/// Returns a vector with two elements: [0] = numeric part, [1] = units +/// If units do not end with 's', appends 's'. +std::vector +parseFreqStr(const std::string &FreqStr ///< [in] frequency string to parse +); + +/// Internal representation of a node in the Analysis operator dependency +/// graph. Each node contains an operator instance, pointers to its upstream +/// dependencies, the names of output streams that consume its output, and +/// non-owning pointers to alarms that trigger its computation. +struct OperatorNode { + std::unique_ptr Op; ///< Operator instance (owned) + std::vector Upstreams; ///< Upstream deps (non-owning) + std::vector StreamNames; ///< Output streams; empty if + ///< intermediate operator + std::vector ComputeAlarms; ///< Compute alarms (non-owning) +}; + +/// The Analysis class is the top-level orchestrator for the in-situ +/// analysis framework. It is responsible for: +/// - Reading analysis configuration and constructing AnalysisGroup instances +/// - Parsing operator chain strings and instantiating operators via factory +/// - Resolving operator dependencies to form a directed acyclic graph (DAG) +/// - Managing the alarm-based scheduling system for operator computation +/// - Owning accumulation alarms for temporal reduction operators +/// - Triggering recursive operator computation on each timestep +/// +/// The class maintains a vector of OperatorNode instances representing all +/// registered operators. Each node stores pointers to its upstream +/// dependencies and the alarms that trigger its computation. The Analysis +/// object owns accumulation alarms for temporal reduction operators, while +/// output alarms are borrowed from IOStream instances. +class Analysis { + public: + /// Initializes the Analysis module by registering all base operators + /// and creating the default Analysis instance. This function must be + /// called after HorzMesh, VertCoord, and TimeStepper are initialized, + /// as it retrieves pointers to the default instances of these objects. + static void init(); + + /// Creates a new named Analysis instance and registers it in the + /// AllAnalysisObjects map. Each Analysis instance maintains its own + /// set of operators and dependency graph. + static Analysis * + create(const std::string &Name, ///< [in] name for new Analysis instance + const MachEnv *Env, ///< [in] machine environment + const HorzMesh *Mesh, ///< [in] horizontal mesh + const VertCoord *VCoord, ///< [in] vertical coordinate + Clock *ModelClock, ///< [in] pointer to model clock + Config *Options ///< [in] configuration options + ); + + /// Computes all analysis fields whose alarms are ringing at the current + /// timestep. This function is called once per timestep from the main + /// driver. It iterates over all operator nodes and triggers recursive + /// computation for any node with a ringing alarm. + void computeAll(); + + /// Parses an underscore-delimited operator chain string and instantiates + /// all operators in the chain that do not yet exist as Fields. For + /// example, "Temperature_SpatialMean_TimeMean1day" parses into three + /// operators. If intermediate operators already exist (shared by other + /// chains), they are reused rather than duplicated. + void parseChainAndBuildOps( + const std::string &OpChainStr ///< [in] underscore-delimited chain string + ); + + /// Instantiates a single operator via the factory and appends it as + /// an OperatorNode. The factory selects the correct templated + /// specialization based on the primary upstream Field's metadata + /// (scalar type, rank, memory location). + void + registerAnalysisOp(const std::string &OpName, ///< [in] operator type name + const std::vector + &UpstreamNames, ///< [in] upstream field names + Config Options ///< [in] operator configuration + ); + + /// Returns a reference to the model clock pointer. This is used by + /// AnalysisGroup instances during stream creation to access the clock + /// for alarm initialization. + Clock *&getModelClock(); + + /// Returns a vector of non-owning pointers to all registered operator + /// nodes. Used primarily for testing and validation of the dependency graph. + const std::vector getOpNodes(); + + /// Checks whether an operator node with the given full instance name + /// already exists in the OpNodes vector. Returns true if found, false + /// otherwise. Used during chain parsing to avoid creating duplicate + /// operators. + bool OpNodeExists(const std::string &FullOpName ///< [in] full op name + ); + + /// Retrieves the default Analysis instance. The preference is to pass + /// the Analysis pointer as an argument, but retrieval is necessary for + /// sharing info between initialization and run phases. + static Analysis *getDefault(); + + /// Destructor - deallocates all memory and deletes the Analysis instance + ~Analysis(); + + /// Removes all defined Analysis instances and cleans up static resources + static void finalize(); + + private: + /// Accumulation alarms owned by Analysis for temporal reduction + /// operators. These alarms control how frequently samples are added to + /// running sums (e.g., every timestep or at a coarser interval). Output + /// alarms are borrowed from IOStream instances and stored in + /// OperatorNode::ComputeAlarms. + std::vector> AccumulationAlarms; + + /// Pointer to the default Analysis instance for easy retrieval + static Analysis *DefAnalysis; + + /// Map of all Analysis instances, paired with names for retrieval + static std::map> AllAnalysisObjects; + + /// Private constructor for creating a new Analysis instance. + /// Called by the create() factory method. Reads configuration, + /// constructs AnalysisGroup instances, and builds the operator + /// dependency graph. + Analysis(const std::string &Name, ///< [in] name for new instance + const MachEnv *Env, ///< [in] machine environment + const HorzMesh *Mesh, ///< [in] horizontal mesh + const VertCoord *VCoord, ///< [in] vertical coordinate + Clock *ModelClock, ///< [in] pointer to model clock + Config *Options ///< [in] configuration options + ); + + std::string Name; ///< Name of this Analysis instance + const MachEnv *Env; ///< Machine environment for MPI operations + const HorzMesh *Mesh; ///< Horizontal mesh for spatial operations + const VertCoord *VCoord; ///< Vertical coordinate for vertical ops + Clock *ModelClock; ///< Pointer to model clock for time mgmt + + /// All registered operator nodes forming the dependency graph + std::vector> OpNodes; + + /// Registers all built-in operator types with the AnalysisOpFactory. + /// Called once during init() before any operators are instantiated. + /// Each operator template is registered with all supported array type + /// variants (scalar types, ranks, memory locations). Defined in Ops.cpp. + static void registerAllBaseAnalysisOperators(); + + /// Post-hoc dependency resolution: iterates over all operator nodes and + /// matches input field names against other nodes' output field names to + /// populate the Upstreams vectors. This forms the edges of the + /// dependency graph. In future versions, this will be replaced by + /// signature-based deduplication during graph construction. + void buildOperatorDependencies(); + + /// Sets ComputeAlarms on terminal nodes by borrowing alarm pointers from + /// associated IOStream instances. For temporal reduction operators, also + /// creates accumulation alarms and adds them to ComputeAlarms. Then + /// calls propagateAlarmsUpstream() to propagate alarms to upstream + /// dependencies. + void setComputeAlarms(); + + /// Calls initialize() on all operators after the dependency graph is + /// complete and all Fields exist. This allows operators to store + /// pointers to mesh, environment, and other resources needed during + /// compute(). + void initializeAllOps(); + + /// Iteratively propagates alarm pointers from downstream operators to + /// upstream operators. An upstream operator must be computed whenever + /// any of its downstream consumers needs data, so each downstream alarm + /// is added to the upstream's ComputeAlarms vector. Iteration continues + /// until no further changes occur (fixed point). + void propagateAlarmsUpstream(); + + /// Recursively computes an operator node and all its upstream + /// dependencies. Uses timestamp-based cache validation to prevent + /// redundant computation when multiple downstream operators share an + /// intermediate result. If the node's LastComputed timestamp matches the + /// current TimeStamp, returns immediately (cache hit). Otherwise, + /// recursively computes all upstream dependencies first, then calls the + /// node's compute() method. + void + computeRecursive(OperatorNode *Node, ///< [in] node to compute + const TimeInstant &TimeStamp ///< [in] current timestamp + ); + + // Forbid copy and move construction + Analysis(const Analysis &) = delete; + Analysis(Analysis &&) = delete; + +}; // end class Analysis + +} // namespace OMEGA +//===----------------------------------------------------------------------===// +#endif // OMEGA_ANALYSIS_H diff --git a/components/omega/src/analysis/AnalysisGroup.cpp b/components/omega/src/analysis/AnalysisGroup.cpp new file mode 100644 index 000000000000..9395cb5cc46d --- /dev/null +++ b/components/omega/src/analysis/AnalysisGroup.cpp @@ -0,0 +1,180 @@ +//===- analysis/AnalysisGroup.cpp - AnalysisGroup implementation -*- C++-*-===// +// +// Implementation of AnalysisGroup base class methods. The key method is +// createAnalysisGroupStreams(), which groups operator chains by their output +// characteristics (period and type), validates temporal reduction periods +// against the restart interval, and creates IOStream objects with appropriate +// configurations for writing analysis output. +// +//===----------------------------------------------------------------------===// + +#include "AnalysisGroup.h" + +namespace OMEGA { + +//------------------------------------------------------------------------------ +// Returns the name of this AnalysisGroup instance +std::string AnalysisGroup::getName() { return GroupName; } // end getName + +//------------------------------------------------------------------------------ +// Groups operator chains by output period and type (temporal reduction vs. +// discrete sampling), validates temporal reduction periods against the restart +// interval, and creates IOStream objects for the group's output. Each stream +// is associated with the appropriate operator nodes, and the operators' output +// fields are added to the stream contents. +void AnalysisGroup::createAnalysisGroupStreams(const std::string &GroupName, + Config &AnalysisGroupOptions, + Analysis *AnalysisManager) { + + Error Err1; + Error Err2; + + // Read optional Stream config node from AnalysisGroup configuration + Config AnalysisStreamOptions("Stream"); + Err1 = AnalysisGroupOptions.get(AnalysisStreamOptions); + std::map ParamOverrides; + + // If Stream config exists, extract parameter overrides + if (Err1.isSuccess()) { + for (Config::Iter It = AnalysisStreamOptions.begin(); + It != AnalysisStreamOptions.end(); ++It) { + std::string Key = It->first.as(); + std::string Value; + AnalysisStreamOptions.get(Key, Value); + ParamOverrides[Key] = Value; + } + } + + // Create StreamParams with defaults and apply any overrides + StreamParams StreamCfg; + StreamCfg.apply(ParamOverrides); + + // Parse filename into prefix and timestamp template + // e.g., "analysis.$Y.$M" -> prefix="analysis", template=".$Y.$M" + std::string FilenameStr; + Err1 = AnalysisGroupOptions.get("Filename", FilenameStr); + CHECK_ERROR_ABORT(Err1, "Analysis: Filename not found in Config"); + + std::string FilenamePrefix; + std::string FilenameTemplate; + size_t Pos = FilenameStr.find("$"); + if (Pos != std::string::npos) { + if (Pos == 0) { + FilenamePrefix.clear(); + FilenameTemplate = FilenameStr; + } else { + // Split at '$'. If a separator (e.g., '.' or '_') immediately precedes + // the template, keep it in the template for backwards compatibility. + FilenamePrefix = FilenameStr.substr(0, Pos); + if (!FilenamePrefix.empty() && + (FilenamePrefix.back() == '.' || FilenamePrefix.back() == '_')) { + FilenamePrefix.pop_back(); + FilenameTemplate = FilenameStr.substr(Pos - 1); + } else { + FilenameTemplate = FilenameStr.substr(Pos); + } + } + } else { + // No timestamp template - use entire string as prefix + FilenamePrefix = FilenameStr; + } + + // Group operator chains by their output stream characteristics + // Chains with the same frequency and type go to the same stream + // Map: StreamName -> list of operator instance names + std::map> StreamToOpNames; + + for (const auto &Info : OpChainInfos) { + std::string StreamName; + + // Construct stream name based on frequency and type + if (Info.IsTimeReduction) { + StreamName = GroupName + "_" + Info.FreqStr + "TimeStats"; + } else { + StreamName = GroupName + "_" + Info.FreqStr + "Samples"; + } + + // Add this operator chain to the appropriate stream + StreamToOpNames[StreamName].push_back(Info.ChainStr); + } + + // Create IOStream objects and associate operator nodes + for (const auto &[StreamName, OpNames] : StreamToOpNames) { + + // Determine stream type from name suffix + bool IsTimeReduction = + (StreamName.find("TimeStats") != std::string::npos); + + // Extract period string from stream name + // e.g., "GlobalStats_1dayTimeStats" -> "1day" + size_t UnderscorePos = StreamName.find_last_of("_"); + std::string PeriodWithSuffix = StreamName.substr(UnderscorePos + 1); + std::string PeriodStr; + if (IsTimeReduction) { + PeriodStr = + PeriodWithSuffix.substr(0, PeriodWithSuffix.find("TimeStats")); + } else { + PeriodStr = + PeriodWithSuffix.substr(0, PeriodWithSuffix.find("Samples")); + } + + // Parse period string into numeric frequency and time units + std::vector ParsedStr = parseFreqStr(PeriodStr); + I4 Freq = std::stoi(ParsedStr[0]); + TimeUnits FreqUnits = TimeUnitsFromString(ParsedStr[1]); + TimeInterval PeriodInterval(Freq, FreqUnits); + + // Validate temporal reduction periods against restart interval + // This ensures proper checkpoint/restart behavior for time-averaged + // fields + if (IsTimeReduction) { + auto RestartAlarm = IOStream::getAlarm("RestartWrite"); + bool IsDivisible = + RestartAlarm->getInterval()->isDivisibleBy(PeriodInterval); + if (!IsDivisible) { + ABORT_ERROR( + "Analysis: The RestartWrite interval is not divisible by " + "the averaging period, {} {}. Currently, temporal averaging " + "is only available over intervals where " + "RestartPeriod % PeriodInterval == 0", + ParsedStr[0], ParsedStr[1]); + } + } + + // Build filename for this stream (includes period and type in name) + StreamCfg.Params["Filename"] = + FilenamePrefix + "_" + PeriodStr + + (IsTimeReduction ? "TimeStats" : "Samples") + FilenameTemplate; + StreamCfg.Params["Freq"] = ParsedStr[0]; + StreamCfg.Params["FreqUnits"] = ParsedStr[1]; + + // Create the IOStream with configured parameters + auto NewStreamCfg = StreamCfg.toConfig(); + auto RefClock = AnalysisManager->getModelClock(); + IOStream::create(StreamName, NewStreamCfg, RefClock); + + // Retrieve the newly created stream + auto Stream = IOStream::get(StreamName); + + // Associate operator nodes with this stream and populate stream contents + auto OpNodes = AnalysisManager->getOpNodes(); + for (auto *Node : OpNodes) { + std::string OpInstanceName = Node->Op->getName(); + + // Check if this operator belongs to this stream + if (std::find(OpNames.begin(), OpNames.end(), OpInstanceName) != + OpNames.end()) { + // Add stream name to operator's list (used for alarm association) + Node->StreamNames.push_back(StreamName); + + // Add all of the operator's output fields to the stream + for (const auto &FieldName : Node->Op->getOutputFieldNames()) { + Stream->addField(FieldName); + } + } + } + } + +} // end createAnalysisGroupStreams + +} // end namespace OMEGA diff --git a/components/omega/src/analysis/AnalysisGroup.h b/components/omega/src/analysis/AnalysisGroup.h new file mode 100644 index 000000000000..b1762d41211d --- /dev/null +++ b/components/omega/src/analysis/AnalysisGroup.h @@ -0,0 +1,156 @@ +#ifndef OMEGA_ANALYSISGROUP_H +#define OMEGA_ANALYSISGROUP_H + +//===-- analysis/AnalysisGroup.h - AnalysisGroup base class -----*- C++ -*-===// +// +/// \file +/// \brief Defines the AnalysisGroup base class for bundled analysis groups +/// +/// AnalysisGroup is the abstract base class for bundled analysis groups that +/// encapsulate common analysis patterns. In the initial implementation (v1), +/// concrete derived classes (e.g., GlobalStats) handle config parsing, operator +/// construction, and stream creation for pre-defined analysis group types. +/// Future versions will support user-defined custom groups specified entirely +/// in config using composable operator chains. +/// +/// Each AnalysisGroup reads its configuration, constructs operator chains +/// by calling Analysis::parseChainAndBuildOps(), and creates associated +/// IOStream objects for output. The base class provides helper structures +/// (OpChainInfo, StreamParams) and methods (createAnalysisGroupStreams) that +/// facilitate grouping operator chains by output frequency and type, validating +/// temporal reduction periods, and creating appropriately configured streams. +/// +//===----------------------------------------------------------------------===// + +#include "Analysis.h" +#include "Config.h" +#include "IOStream.h" +#include +#include + +namespace OMEGA { + +/// AnalysisGroup is the abstract base class for bundled analysis groups. +/// Concrete derived classes (e.g., GlobalStats) encapsulate the configuration +/// parsing, operator construction, and stream creation logic for named analysis +/// group types. The base class provides utilities for grouping operator chains +/// by their output characteristics and creating IOStream objects with +/// appropriate configurations. Future versions will support user-defined custom +/// groups where users specify composable operator chains directly in the config +/// file. +class AnalysisGroup { + + public: + /// Virtual destructor allows polymorphic deletion of derived classes + virtual ~AnalysisGroup() = default; + + /// Returns the name of this AnalysisGroup instance + std::string getName(); + + /// Groups operator chains by output period and type (time reduction vs. + /// discrete sampling), validates temporal reduction periods against the + /// restart interval, and creates IOStream objects for the group's output. + /// Called by derived class constructors after all operator chains have + /// been registered with the Analysis orchestrator. + void createAnalysisGroupStreams( + const std::string &GroupName, ///< [in] name of this group + Config &AnalysisGroupOptions, ///< [in] group configuration options + Analysis *AnalysisManager ///< [in] Analysis orchestrator instance + ); + + protected: + /// Metadata about a single operator chain within this AnalysisGroup. + /// Stores the chain string (operator instance name), the output frequency, + /// and whether the chain performs temporal reduction or discrete sampling. + struct OpChainInfo { + std::string ChainStr; ///< Operator instance name (output Field name) + std::string FreqStr; ///< Period/frequency string (e.g., "1day", "6hour") + bool IsTimeReduction; ///< true for temporal reduction; false for discrete + ///< samples + }; + + /// Template for constructing IOStream configurations for this group's + /// output. Provides default values for all IOStream creation parameters. + /// Derived classes can override defaults using group-specific config options + /// via the apply() method. The toConfig() method converts the parameters to + /// a Config object suitable for IOStream::create(). + struct StreamParams { + /// Constructor initializes all IOStream parameters with default values. + /// Empty string values indicate parameters that must be set by derived + /// classes or will be omitted from the final stream configuration. + StreamParams() + : Params{ + {"UsePointerFile", "false"}, + {"PointerFilename", ""}, + {"Filename", ""}, + {"Mode", "write"}, + {"IfExists", "append"}, + {"Precision", "double"}, + {"Freq", ""}, + {"FreqUnits", ""}, + {"FileFreq", ""}, + {"FileFreqUnits", ""}, + {"UseStartEnd", "false"}, + {"StartTime", ""}, + {"EndTime", ""}, + } {} + + /// Applies group-specific configuration overrides to the default + /// parameter values. Only known parameters can be overridden; unknown + /// keys trigger an error. + void apply(const std::map + &Overrides ///< [in] parameter overrides + ) { + for (const auto &[Key, Value] : Overrides) { + auto It = Params.find(Key); + + // Validate that the key exists in our parameter map + if (It == Params.end()) { + ABORT_ERROR("Analysis: Unknown Stream config parameter, {}", + Key); + } + + It->second = Value; + } + } + + /// Converts the parameter map to a Config object suitable for passing + /// to IOStream::create(). Only parameters with non-empty values are + /// included in the Config. Contents are left empty and must be added + /// after stream creation. + Config toConfig() const { + Config Cfg; + + // Add only parameters with non-empty values + for (const auto &[Key, Value] : Params) { + if (!Value.empty()) { + Cfg.add(Key, Value); + } + } + + // Contents are populated after stream creation via IOStream::addField + std::vector EmptyStrVec{""}; + Cfg.add("Contents", EmptyStrVec); + + return Cfg; + } + + /// Map of all IOStream configuration options for this group's output + std::map Params; + }; + + std::string GroupName; ///< Name of this AnalysisGroup instance + + /// Metadata for all operator chains in this group (output field name, + /// frequency, and type). Populated by derived class constructors. + std::vector OpChainInfos; + + /// Full operator chain strings for all chains in this group. This vector + /// stores the complete underscore-delimited chain strings for reference. + std::vector OpChainStrings; + +}; // end class AnalysisGroup + +} // end namespace OMEGA + +#endif diff --git a/components/omega/src/analysis/AnalysisOpFactory.cpp b/components/omega/src/analysis/AnalysisOpFactory.cpp new file mode 100644 index 000000000000..a3675997fbd7 --- /dev/null +++ b/components/omega/src/analysis/AnalysisOpFactory.cpp @@ -0,0 +1,115 @@ +//===-- analysis/AnalysisOpFactory.cpp - Factory implementation --*- C++-*-===// +// +// Implementation of AnalysisOpFactory static methods for operator registration +// and creation. The factory maintains a runtime registry (Meyer's singleton) +// that maps fully-qualified operator type keys to constructor functions. +// Registration occurs during program initialization via static initializers. +// Creation performs type-safe dispatch by inspecting Field metadata to select +// the appropriate templated operator specialization. +// +//===----------------------------------------------------------------------===// + +#include "AnalysisOpFactory.h" +#include + +namespace OMEGA { + +//------------------------------------------------------------------------------ +// Registers an operator variant by adding it to the factory registry. Checks +// for duplicate registration and aborts if the key already exists, which +// indicates a programming error (likely duplicate registration in static +// initializers). +void AnalysisOpFactory::registerOperator(const std::string &Label, + CreatorFunc Creator) { + + auto &Reg = registry(); + + // Check for duplicate registration (programming error) + if (Reg.find(Label) != Reg.end()) { + ABORT_ERROR("AnalysisOpFactory: Operator type {} is already registered", + Label); + } + + // Add constructor function to registry + Reg[Label] = Creator; +} // end registerOperator + +//------------------------------------------------------------------------------ +// Creates an operator instance by inspecting the primary upstream Field's +// metadata to construct a fully-qualified type key, looking up the constructor +// in the registry, and invoking it. The primary Field (first in UpstreamNames) +// determines the array type template parameter for the operator. +std::unique_ptr +AnalysisOpFactory::createOp(const std::string &OpType, + const std::vector &UpstreamNames, + Config Options) { + + // Validate that all upstream Fields exist before attempting creation + for (const auto &FieldName : UpstreamNames) { + auto FieldPtr = Field::get(FieldName); + if (!FieldPtr) { + ABORT_ERROR("Field '{}' not found for operator creation", FieldName); + } + } + + // Extract metadata from primary upstream Field (determines array type) + auto FieldPtr = Field::get(UpstreamNames[0]); + ArrayDataType DType = FieldPtr->getType(); + int Rank = FieldPtr->getNumDims(); + ArrayMemLoc MemLoc = FieldPtr->getMemoryLocation(); + + // Map metadata to array type name (e.g., "Array2DR8_Device") + std::string arrayTypeName = getArrayTypeName(DType, Rank, MemLoc); + + // Build fully-qualified operator key (e.g., "SpatialMean_Array2DR8_Device") + std::string FullOpType = OpType + "_" + arrayTypeName; + + // Look up constructor in registry + auto &Reg = registry(); + auto it = Reg.find(FullOpType); + + if (it == Reg.end()) { + ABORT_ERROR("Operator type {} not found", FullOpType); + } + + // Invoke the registered constructor function + return it->second(UpstreamNames, Options); +} // end createOp + +//------------------------------------------------------------------------------ +// Checks whether an operator variant with the given key exists in the registry. +// Returns true if found, false otherwise. +bool AnalysisOpFactory::hasOperator(const std::string &Type) { + auto &Reg = registry(); + return Reg.find(Type) != Reg.end(); +} // end hasOperator + +//------------------------------------------------------------------------------ +// Constructs an array type name string from Field metadata by expanding the +// OMEGA_ANALYSIS_ARRAY_TYPES macro and checking each combination. Returns a +// string like "Array2DR8_Device" that is used as part of the operator variant +// key. Aborts if the metadata combination is not supported. +std::string AnalysisOpFactory::getArrayTypeName(ArrayDataType DType, int Rank, + ArrayMemLoc MemLoc) { +// Define a macro to check if metadata matches this array type +#define TRY_ARRAY_TYPE(dt, r, ml, ArrayT) \ + if (DType == dt && Rank == r && MemLoc == ml) { \ + return std::string(#ArrayT) + "_" + #ml; \ + } + + // Expand macro over all supported array type combinations + OMEGA_ANALYSIS_ARRAY_TYPES(TRY_ARRAY_TYPE) + +#undef TRY_ARRAY_TYPE + + // If we reach here, the array type is not supported + ABORT_ERROR( + "Unsupported array type/Rank/location: DType={}, Rank={}, MemLoc={}", + static_cast(DType), Rank, static_cast(MemLoc)); + + return {}; +} // end getArrayTypeName + +} // end namespace OMEGA + +//===----------------------------------------------------------------------===// diff --git a/components/omega/src/analysis/AnalysisOpFactory.h b/components/omega/src/analysis/AnalysisOpFactory.h new file mode 100644 index 000000000000..f394f8118e23 --- /dev/null +++ b/components/omega/src/analysis/AnalysisOpFactory.h @@ -0,0 +1,181 @@ +#ifndef OMEGA_ANALYSISOPFACTORY_H +#define OMEGA_ANALYSISOPFACTORY_H + +//===-- analysis/AnalysisOpFactory.h - Operator factory ---------*- C++ -*-===// +// +/// \file +/// \brief Defines the AnalysisOpFactory for runtime operator creation +/// +/// The AnalysisOpFactory provides a runtime registry that maps operator type +/// names to constructor functions, enabling dynamic instantiation of templated +/// operators without hard-coded switch statements. The factory: +/// - Supports decentralized registration via template helpers +/// - Performs type-safe dispatch based on Field metadata (scalar type, rank, +/// memory location) +/// - Enables extensibility by allowing new operators to integrate without +/// modifying orchestration code +/// +/// Each operator template (e.g., SpatialMaxOp) is registered for all +/// supported array type combinations via registerAllArrayVariants(). At +/// operator creation time, the factory inspects the primary upstream Field's +/// metadata and selects the matching templated specialization. +/// +//===----------------------------------------------------------------------===// + +#include "AnalysisOperator.h" +#include "Config.h" + +#include + +namespace OMEGA { + +/// Macro defining all supported array type combinations for analysis operators. +/// Used by registerAllArrayVariants() to register operator variants for all +/// combinations of scalar type, rank, and memory location. The macro takes a +/// single argument X which should be a macro that processes each +/// (DType, Rank, MemLoc, ArrayT) tuple. +#define OMEGA_ANALYSIS_ARRAY_TYPES(X) \ + /* 1D Arrays */ \ + X(ArrayDataType::I4, 1, ArrayMemLoc::Both, Array1DI4) \ + X(ArrayDataType::I4, 1, ArrayMemLoc::Device, Array1DI4) \ + X(ArrayDataType::I8, 1, ArrayMemLoc::Both, Array1DI8) \ + X(ArrayDataType::I8, 1, ArrayMemLoc::Device, Array1DI8) \ + X(ArrayDataType::R4, 1, ArrayMemLoc::Both, Array1DR4) \ + X(ArrayDataType::R4, 1, ArrayMemLoc::Device, Array1DR4) \ + X(ArrayDataType::R8, 1, ArrayMemLoc::Both, Array1DR8) \ + X(ArrayDataType::R8, 1, ArrayMemLoc::Device, Array1DR8) \ + /* 2D Arrays */ \ + X(ArrayDataType::I4, 2, ArrayMemLoc::Both, Array2DI4) \ + X(ArrayDataType::I4, 2, ArrayMemLoc::Device, Array2DI4) \ + X(ArrayDataType::I8, 2, ArrayMemLoc::Both, Array2DI8) \ + X(ArrayDataType::I8, 2, ArrayMemLoc::Device, Array2DI8) \ + X(ArrayDataType::R4, 2, ArrayMemLoc::Both, Array2DR4) \ + X(ArrayDataType::R4, 2, ArrayMemLoc::Device, Array2DR4) \ + X(ArrayDataType::R8, 2, ArrayMemLoc::Both, Array2DR8) \ + X(ArrayDataType::R8, 2, ArrayMemLoc::Device, Array2DR8) \ + /* 3D Arrays */ \ + X(ArrayDataType::I4, 3, ArrayMemLoc::Both, Array3DI4) \ + X(ArrayDataType::I4, 3, ArrayMemLoc::Device, Array3DI4) \ + X(ArrayDataType::I8, 3, ArrayMemLoc::Both, Array3DI8) \ + X(ArrayDataType::I8, 3, ArrayMemLoc::Device, Array3DI8) \ + X(ArrayDataType::R4, 3, ArrayMemLoc::Both, Array3DR4) \ + X(ArrayDataType::R4, 3, ArrayMemLoc::Device, Array3DR4) \ + X(ArrayDataType::R8, 3, ArrayMemLoc::Both, Array3DR8) \ + X(ArrayDataType::R8, 3, ArrayMemLoc::Device, Array3DR8) // \ +// /* 4D Arrays */ \ +// X(ArrayDataType::I4, 4, ArrayMemLoc::Both, Array4DI4) \ +// X(ArrayDataType::I4, 4, ArrayMemLoc::Device, Array4DI4) \ +// X(ArrayDataType::I8, 4, ArrayMemLoc::Both, Array4DI8) \ +// X(ArrayDataType::I8, 4, ArrayMemLoc::Device, Array4DI8) \ +// X(ArrayDataType::R4, 4, ArrayMemLoc::Both, Array4DR4) \ +// X(ArrayDataType::R4, 4, ArrayMemLoc::Device, Array4DR4) \ +// X(ArrayDataType::R8, 4, ArrayMemLoc::Both, Array4DR8) \ +// X(ArrayDataType::R8, 4, ArrayMemLoc::Device, Array4DR8) \ +// /* 5D Arrays */ \ +// X(ArrayDataType::I4, 5, ArrayMemLoc::Both, Array5DI4) \ +// X(ArrayDataType::I4, 5, ArrayMemLoc::Device, Array5DI4) \ +// X(ArrayDataType::I8, 5, ArrayMemLoc::Both, Array5DI8) \ +// X(ArrayDataType::I8, 5, ArrayMemLoc::Device, Array5DI8) \ +// X(ArrayDataType::R4, 5, ArrayMemLoc::Both, Array5DR4) \ +// X(ArrayDataType::R4, 5, ArrayMemLoc::Device, Array5DR4) \ +// X(ArrayDataType::R8, 5, ArrayMemLoc::Both, Array5DR8) \ +// X(ArrayDataType::R8, 5, ArrayMemLoc::Device, Array5DR8) \ + +/// AnalysisOpFactory is a singleton factory class that manages runtime +/// registration and creation of analysis operators. The factory maintains a +/// registry mapping operator type keys to constructor functions. Keys are +/// formed from base operator name, array type, and memory location (e.g., +/// "SpatialMax_Array2DR8_Device"). This enables type-safe dispatch: the +/// factory inspects the primary upstream Field's metadata at creation time +/// and selects the correct templated specialization. +/// +/// All methods are static; the underlying registry is a Meyer's singleton +/// guaranteed to be initialized before first use. Operators self-register +/// during program initialization via registerAllArrayVariants(). +class AnalysisOpFactory { + public: + /// Function signature for operator constructors. Takes upstream field names + /// and configuration options, returns a unique_ptr to a new operator + /// instance. + using CreatorFunc = std::function( + const std::vector &UpstreamNames, Config Options)>; + + /// Registers a single operator variant in the factory by associating a + /// string label (key) with a constructor function. The label typically + /// includes the operator type, array type, and memory location. Aborts if + /// the label is already registered (duplicate key). + static void registerOperator( + const std::string &Label, ///< [in] unique key for this operator variant + CreatorFunc Creator ///< [in] constructor function for this variant + ); + + /// Creates an operator instance by inspecting the primary upstream Field's + /// metadata (scalar type, rank, memory location) to build a fully-qualified + /// type key, looking up the constructor in the registry, and invoking it + /// with the provided arguments. Aborts if the operator type or array variant + /// is not registered. + static std::unique_ptr + createOp(const std::string &OpType, ///< [in] operator type name + const std::vector + &UpstreamNames, ///< [in] upstream field names + Config Options ///< [in] operator configuration + ); + + /// Returns a list of all registered operator variant keys. Useful for + /// validation and generating informative error messages when an operator + /// type is not found. + static std::vector availableOperators(); + + /// Checks whether an operator variant with the given key is registered. + /// Returns true if found, false otherwise. + static bool + hasOperator(const std::string &Type ///< [in] operator variant key + ); + + /// Registers all array type variants of a templated operator class by + /// expanding OMEGA_ANALYSIS_ARRAY_TYPES over all supported (DType, Rank, + /// MemLoc, ArrayT) combinations. For each combination, constructs a key + /// from baseName + "_" + ArrayT + "_" + MemLoc and registers a lambda + /// that instantiates OperatorTemplate. This enables a single call + /// to register dozens of operator variants. + template