Working on meteo sampling (e.g. aggregation at daily timestep from hourly meteo) + make a uniform interface with policies

Author: VEZY

docs/src/API/API_public.md

@@ -20,6 +20,7 @@ For mapping-level multi-rate configuration, combine:
 - `ModelSpec(...)`
 - `TimeStepModel(...)`
 - `InputBindings(...)`
+- `MeteoBindings(...)`
 - `OutputRouting(...)`
 - `ScopeModel(...)`
 - `OutputRequest(...)` in `tracked_outputs` for resampled exports
@@ -70,26 +71,39 @@ TimeStepModel(ClockSpec(2.0, 1.0)) |>
 InputBindings(; x=(process=:producer, var=:x))
 ```
 
+### Meteo aggregation bindings
+
+```julia
+ModelSpec(DailyModel()) |>
+TimeStepModel(ClockSpec(24.0, 1.0)) |>
+MeteoBindings(
+    T=MeanWeighted(),                     # default source is :T
+    Ri_SW_f=RadiationEnergy(),            # integrate W m-2 to MJ m-2 over the model window
+    custom_peak=(source=:custom_var, reducer=MaxReducer()),
+)
+```
+
 ### Parameterized window reducers
 
-`Integrate()` defaults to `:sum`; `Aggregate()` defaults to `:mean`.
+`Integrate()` defaults to `SumReducer()`; `Aggregate()` defaults to `MeanReducer()`.
 
 ```julia
 ModelSpec(DailyModel()) |>
 TimeStepModel(ClockSpec(24.0, 1.0)) |>
-InputBindings(; a=(process=:hourly_assim, var=:A, scale="Leaf", policy=Integrate(:sum)))
+InputBindings(; a=(process=:hourly_assim, var=:A, scale="Leaf", policy=Integrate(SumReducer())))
 
 ModelSpec(DailyModel()) |>
 TimeStepModel(ClockSpec(24.0, 1.0)) |>
-InputBindings(; a=(process=:hourly_assim, var=:A, scale="Leaf", policy=Aggregate(:max)))
+InputBindings(; a=(process=:hourly_assim, var=:A, scale="Leaf", policy=Aggregate(MaxReducer())))
 
 ModelSpec(DailyModel()) |>
 TimeStepModel(ClockSpec(24.0, 1.0)) |>
 InputBindings(; a=(process=:hourly_assim, var=:A, scale="Leaf", policy=Integrate(vals -> maximum(vals) - minimum(vals))))
 ```
 
-Supported reducer symbols are:
-`:sum`, `:mean`, `:max`, `:min`, `:first`, `:last`.
+Built-in reducer types are:
+`SumReducer()`, `MeanReducer()`, `MaxReducer()`, `MinReducer()`, `FirstReducer()`, `LastReducer()`.
+The same reducer objects are also used by `MeteoBindings(...)`.
 
 ### Parameterized interpolation mode
 

docs/src/model_execution.md

@@ -15,13 +15,15 @@ For multiscale simulations, model usage is configured in the mapping through `Mo
 
 - `TimeStepModel(...)`: sets model execution clock.
 - `InputBindings(...)`: sets producer, source variable, optional source scale, and policy for each consumer input.
+- `MeteoBindings(...)`: sets weather aggregation rules at the model clock for meteo variables.
 - `OutputRouting(...)`: sets whether an output is canonical (`:canonical`) or stream-only (`:stream_only`).
 - `ScopeModel(...)`: partitions producer streams by scope (`:global`, `:plant`, `:scene`, `:self`) for multi-entity simulations.
 
 Policy parameterization:
-- `Integrate()` defaults to `:sum`; you can pass another reducer, e.g. `Integrate(:mean)` or `Integrate(vals -> maximum(vals) - minimum(vals))`.
-- `Aggregate()` defaults to `:mean`; you can pass reducers such as `Aggregate(:max)`.
+- `Integrate()` defaults to `SumReducer()`; you can pass another reducer, e.g. `Integrate(MeanReducer())` or `Integrate(vals -> maximum(vals) - minimum(vals))`.
+- `Aggregate()` defaults to `MeanReducer()`; you can pass reducers such as `Aggregate(MaxReducer())`.
 - `Interpolate()` defaults to `mode=:linear, extrapolation=:linear`; use `Interpolate(; mode=:hold, extrapolation=:hold)` for hold behavior.
+- The same reducer objects are reused by meteo sampling (`MeteoBindings`) and by windowed policies (`Integrate`, `Aggregate`).
 
 `TimeStepModel(...)` accepts either step counts (`Real`), `ClockSpec`, or fixed `Dates` periods
 (for example `Dates.Hour(1)`, `Dates.Day(1)`). Fixed periods are converted internally using
@@ -41,6 +43,7 @@ Typical pipeline form:
 ```julia
 ModelSpec(MyModel()) |>
 TimeStepModel(ClockSpec(24.0, 1.0)) |>
+MeteoBindings(; T=MeanWeighted()) |>
 InputBindings(; x=(process=:producer, var=:y, policy=HoldLast())) |>
 OutputRouting(; z=:stream_only)
 ```
@@ -87,6 +90,10 @@ mapping = Dict(
 ```
 
 When `multirate=true` is passed to `run!`, the runtime resolves inputs from producer temporal streams according to these policies.
+Meteo rows are also sampled at each model clock. By default, meteo variables are aggregated from
+the finest weather step (for example `T` and `Rh` as weighted means, `Tmin/Tmax`, and radiation
+quantity aliases such as `Ri_SW_q` in MJ m-2). You can override these rules with `MeteoBindings(...)`
+on each `ModelSpec`.
 
 ### Current limitations
 

src/PlantSimEngine.jl

@@ -101,6 +101,7 @@ include("time/runtime/bindings.jl")
 include("time/runtime/input_resolution.jl")
 include("time/runtime/publishers.jl")
 include("time/runtime/output_export.jl")
+include("time/runtime/meteo_sampling.jl")
 
 # Simulation:
 include("run.jl")
@@ -118,18 +119,19 @@ export PreviousTimeStep
 export AbstractModel
 export ScopeId, ClockSpec, ModelKey, OutputKey
 export SchedulePolicy, HoldLast, Interpolate, Integrate, Aggregate
+export AbstractTimeReducer, MeanWeighted, MeanReducer, SumReducer, MinReducer, MaxReducer, FirstReducer, LastReducer, RadiationEnergy
 export OutputCache, HoldLastCache, InterpolateCache, IntegrateCache, AggregateCache
 export TemporalState
 export OutputRequest, collect_outputs
-export ModelList, MultiScaleModel, ModelSpec, TimeStepModel, InputBindings, OutputRouting, ScopeModel
+export ModelList, MultiScaleModel, ModelSpec, TimeStepModel, InputBindings, MeteoBindings, OutputRouting, ScopeModel
 export RMSE, NRMSE, EF, dr
 export Status, TimeStepTable, status
 export init_status!
 export add_organ!
 export @process, process
 export to_initialize, is_initialized, init_variables, dep
 export inputs, outputs, variables, convert_outputs
-export timespec, output_policy, input_bindings, output_routing, model_scope
+export timespec, output_policy, input_bindings, meteo_bindings, output_routing, model_scope
 export run!
 export fit
 

src/mtg/ModelSpec.jl

@@ -1,17 +1,18 @@
 """
-    ModelSpec(model; multiscale=nothing, timestep=nothing, input_bindings=NamedTuple(), output_routing=NamedTuple(), scope=:global)
+    ModelSpec(model; multiscale=nothing, timestep=nothing, input_bindings=NamedTuple(), meteo_bindings=NamedTuple(), output_routing=NamedTuple(), scope=:global)
 
 User-side model configuration wrapper for mapping/model list composition.
 
 `ModelSpec` keeps model implementation and scenario-specific usage metadata in one place.
 This allows modelers to publish reusable models while users decide how models are coupled in
 their simulation setup.
 """
-struct ModelSpec{M,MS,TS,IB,OR,SC}
+struct ModelSpec{M,MS,TS,IB,MB,OR,SC}
     model::M
     multiscale::MS
     timestep::TS
     input_bindings::IB
+    meteo_bindings::MB
     output_routing::OR
     scope::SC
 end
@@ -27,6 +28,7 @@ function ModelSpec(
     multiscale=nothing,
     timestep=nothing,
     input_bindings=NamedTuple(),
+    meteo_bindings=NamedTuple(),
     output_routing=NamedTuple(),
     scope=:global
 )
@@ -40,13 +42,15 @@ function ModelSpec(
 
     normalized_multiscale = _normalize_multiscale_mapping(base_model, base_multiscale)
     normalized_input_bindings = _normalize_input_bindings(input_bindings)
+    normalized_meteo_bindings = _normalize_meteo_bindings(meteo_bindings)
     normalized_output_routing = _normalize_output_routing(output_routing)
     normalized_scope = _normalize_scope_selector(scope)
-    return ModelSpec{typeof(base_model),typeof(normalized_multiscale),typeof(timestep),typeof(normalized_input_bindings),typeof(normalized_output_routing),typeof(normalized_scope)}(
+    return ModelSpec{typeof(base_model),typeof(normalized_multiscale),typeof(timestep),typeof(normalized_input_bindings),typeof(normalized_meteo_bindings),typeof(normalized_output_routing),typeof(normalized_scope)}(
         base_model,
         normalized_multiscale,
         timestep,
         normalized_input_bindings,
+        normalized_meteo_bindings,
         normalized_output_routing,
         normalized_scope
     )
@@ -58,10 +62,11 @@ function ModelSpec(
     multiscale=spec.multiscale,
     timestep=spec.timestep,
     input_bindings=spec.input_bindings,
+    meteo_bindings=spec.meteo_bindings,
     output_routing=spec.output_routing,
     scope=spec.scope
 )
-    ModelSpec(model; multiscale=multiscale, timestep=timestep, input_bindings=input_bindings, output_routing=output_routing, scope=scope)
+    ModelSpec(model; multiscale=multiscale, timestep=timestep, input_bindings=input_bindings, meteo_bindings=meteo_bindings, output_routing=output_routing, scope=scope)
 end
 
 as_model_spec(spec::ModelSpec) = spec
@@ -98,6 +103,16 @@ function with_input_bindings(model_or_spec, bindings)
     return ModelSpec(spec; input_bindings=_normalize_input_bindings(bindings))
 end
 
+"""
+    with_meteo_bindings(model_or_spec, bindings)
+
+Return a `ModelSpec` with explicit meteo aggregation bindings.
+"""
+function with_meteo_bindings(model_or_spec, bindings)
+    spec = as_model_spec(model_or_spec)
+    return ModelSpec(spec; meteo_bindings=_normalize_meteo_bindings(bindings))
+end
+
 """
     with_output_routing(model_or_spec, routing)
 
@@ -139,6 +154,36 @@ end
 
 _normalize_input_bindings(bindings) = bindings
 
+function _normalize_meteo_binding(binding)
+    if binding isa DataType
+        binding <: PlantMeteo.AbstractTimeReducer || error(
+            "Unsupported MeteoBindings reducer type `$(binding)`. ",
+            "Use a PlantMeteo reducer type/instance, callable, or NamedTuple(source=..., reducer=...)."
+        )
+        return binding
+    elseif binding isa PlantMeteo.AbstractTimeReducer
+        return binding
+    elseif binding isa Function
+        return binding
+    elseif binding isa NamedTuple
+        return binding
+    end
+    error(
+        "Unsupported MeteoBindings value `$(binding)` of type `$(typeof(binding))`. ",
+        "Use a PlantMeteo reducer type/instance, callable, or NamedTuple(source=..., reducer=...)."
+    )
+end
+
+function _normalize_meteo_bindings(bindings::NamedTuple)
+    normalized = Pair{Symbol,Any}[]
+    for (k, v) in pairs(bindings)
+        push!(normalized, k => _normalize_meteo_binding(v))
+    end
+    return (; normalized...)
+end
+
+_normalize_meteo_bindings(bindings) = bindings
+
 function _normalize_output_routing(routing::NamedTuple)
     normalized = Pair{Symbol,Symbol}[]
     for (k, v) in pairs(routing)
@@ -184,6 +229,20 @@ Pipe-style transform that sets explicit input bindings on a model/spec.
 InputBindings(bindings) = x -> with_input_bindings(x, bindings)
 InputBindings(; kwargs...) = InputBindings((; kwargs...))
 
+"""
+    MeteoBindings(bindings)
+    MeteoBindings(; kwargs...)
+
+Pipe-style transform that sets explicit weather sampling bindings on a model/spec.
+Each key is the target meteo variable seen by the model.
+Each value can be:
+- a PlantMeteo reducer instance/type (e.g. `MeanWeighted()`, `MaxReducer`)
+- `Function`: custom reducer callable
+- `NamedTuple`: optional fields `source` (Symbol/String) and `reducer`
+"""
+MeteoBindings(bindings) = x -> with_meteo_bindings(x, bindings)
+MeteoBindings(; kwargs...) = MeteoBindings((; kwargs...))
+
 """
     OutputRouting(routing)
     OutputRouting(; kwargs...)

src/mtg/model_spec_validation.jl

@@ -1,21 +1,46 @@
 const _INPUT_BINDING_FIELDS = (:process, :var, :scale, :policy)
 const _MODEL_SCOPE_SELECTORS = (:global, :plant, :scene, :self)
+const _METEO_BINDING_FIELDS = (:source, :reducer)
 
 function _validate_window_reducer(scale::String, process::Symbol, input_var::Symbol, policy_name::Symbol, reducer)
-    if reducer isa Symbol
-        reducer in _WINDOW_REDUCER_SYMBOLS || error(
-            "Invalid reducer `$(reducer)` for policy `$(policy_name)` on input `$(input_var)` ",
-            "in process `$(process)` at scale `$(scale)`. Supported symbols are $(_WINDOW_REDUCER_SYMBOLS)."
+    if reducer isa DataType
+        reducer <: PlantMeteo.AbstractTimeReducer || error(
+            "Invalid reducer type `$(reducer)` for policy `$(policy_name)` on input `$(input_var)` ",
+            "in process `$(process)` at scale `$(scale)`. ",
+            "Expected a PlantMeteo reducer type/instance or a callable."
+        )
+        rr = try
+            reducer()
+        catch
+            error(
+                "Reducer type `$(reducer)` for policy `$(policy_name)` on input `$(input_var)` ",
+                "in process `$(process)` at scale `$(scale)` cannot be instantiated without arguments."
+            )
+        end
+        applicable(rr, [1.0, 2.0]) || error(
+            "Reducer type `$(reducer)` for policy `$(policy_name)` on input `$(input_var)` in process `$(process)` at scale `$(scale)` ",
+            "must be callable on a vector of numeric values."
+        )
+        return nothing
+    elseif reducer isa PlantMeteo.AbstractTimeReducer
+        applicable(reducer, [1.0, 2.0]) || error(
+            "Reducer `$(typeof(reducer))` for policy `$(policy_name)` on input `$(input_var)` in process `$(process)` at scale `$(scale)` ",
+            "must be callable on a vector of numeric values."
+        )
+        return nothing
+    elseif reducer isa Function
+        applicable(reducer, [1.0, 2.0]) || error(
+            "Reducer for policy `$(policy_name)` on input `$(input_var)` in process `$(process)` at scale `$(scale)` ",
+            "must be callable on a vector of numeric values."
         )
         return nothing
     end
 
-    applicable(reducer, [1.0, 2.0]) || error(
-        "Reducer for policy `$(policy_name)` on input `$(input_var)` in process `$(process)` at scale `$(scale)` ",
-        "must be a supported Symbol or a callable accepting a vector of numeric values."
+    error(
+        "Invalid reducer value `$(reducer)` (type `$(typeof(reducer))`) for policy `$(policy_name)` ",
+        "on input `$(input_var)` in process `$(process)` at scale `$(scale)`. ",
+        "Expected a PlantMeteo reducer type/instance or a callable."
     )
-
-    return nothing
 end
 
 function _validate_policy_instance(scale::String, process::Symbol, input_var::Symbol, policy::SchedulePolicy)
@@ -275,6 +300,81 @@ function _validate_output_routing_for_spec(scale::String, process::Symbol, spec:
     return nothing
 end
 
+function _validate_meteo_binding(scale::String, process::Symbol, target_var::Symbol, binding)
+    if binding isa Function || binding isa PlantMeteo.AbstractTimeReducer
+        return nothing
+    elseif binding isa DataType
+        binding <: PlantMeteo.AbstractTimeReducer || error(
+            "Invalid MeteoBindings reducer type for variable `$(target_var)` in process `$(process)` at scale `$(scale)`: ",
+            "expected a subtype of `PlantMeteo.AbstractTimeReducer`."
+        )
+        try
+            binding()
+        catch
+            error(
+                "Invalid MeteoBindings reducer type for variable `$(target_var)` in process `$(process)` at scale `$(scale)`: ",
+                "type `$(binding)` cannot be instantiated without arguments."
+            )
+        end
+        return nothing
+    elseif binding isa NamedTuple
+        extra = setdiff(collect(keys(binding)), collect(_METEO_BINDING_FIELDS))
+        isempty(extra) || error(
+            "Invalid MeteoBindings for variable `$(target_var)` in process `$(process)` at scale `$(scale)`: ",
+            "unsupported fields $(extra)."
+        )
+
+        if haskey(binding, :source)
+            binding.source isa Symbol || binding.source isa AbstractString || error(
+                "Invalid MeteoBindings source for variable `$(target_var)` in process `$(process)` at scale `$(scale)`: ",
+                "`source` must be a Symbol or String."
+            )
+        end
+        if haskey(binding, :reducer)
+            reducer = binding.reducer
+            if reducer isa DataType
+                reducer <: PlantMeteo.AbstractTimeReducer || error(
+                    "Invalid MeteoBindings reducer for variable `$(target_var)` in process `$(process)` at scale `$(scale)`: ",
+                    "`reducer` type must subtype `PlantMeteo.AbstractTimeReducer`."
+                )
+                try
+                    reducer()
+                catch
+                    error(
+                        "Invalid MeteoBindings reducer type for variable `$(target_var)` in process `$(process)` at scale `$(scale)`: ",
+                        "type `$(reducer)` cannot be instantiated without arguments."
+                    )
+                end
+            else
+                (reducer isa PlantMeteo.AbstractTimeReducer || reducer isa Function) || error(
+                    "Invalid MeteoBindings reducer for variable `$(target_var)` in process `$(process)` at scale `$(scale)`: ",
+                    "`reducer` must be a reducer instance/type or a callable."
+                )
+            end
+        end
+        return nothing
+    end
+
+    error(
+        "Invalid MeteoBindings value for variable `$(target_var)` in process `$(process)` at scale `$(scale)`: ",
+        "unsupported type `$(typeof(binding))`."
+    )
+end
+function _validate_meteo_bindings_for_spec(scale::String, process::Symbol, spec::ModelSpec)
+    bindings = meteo_bindings(spec)
+    bindings isa NamedTuple || error(
+        "MeteoBindings for process `$(process)` at scale `$(scale)` must be a NamedTuple, got `$(typeof(bindings))`."
+    )
+
+    for (target_var, binding) in pairs(bindings)
+        target_var isa Symbol || error(
+            "MeteoBindings key for process `$(process)` at scale `$(scale)` must be a Symbol, got `$(typeof(target_var))`."
+        )
+        _validate_meteo_binding(scale, process, target_var, binding)
+    end
+    return nothing
+end
+
 """
     validate_model_specs_configuration(model_specs)
 
@@ -292,6 +392,7 @@ function validate_model_specs_configuration(model_specs)
             _validate_timestep_spec(scale, process, spec)
             _validate_scope_spec(scale, process, spec)
             _validate_input_bindings_for_spec(scale, process, spec, model_specs, known_processes)
+            _validate_meteo_bindings_for_spec(scale, process, spec)
             _validate_output_routing_for_spec(scale, process, spec)
         end
     end