Optimize a bit again

Author: VEZY

docs/src/API/API_public.md

@@ -22,7 +22,7 @@ For mapping-level multi-rate configuration, combine:
 - `InputBindings(...)`
 - `OutputRouting(...)`
 - `ScopeModel(...)`
-- `OutputRequest(...)` with `collect_outputs(...)` for resampled exports
+- `OutputRequest(...)` in `tracked_outputs` for resampled exports
 
 `TimeStepModel(...)` accepts:
 - `Real` step counts
@@ -45,7 +45,17 @@ Scope selection detail:
 ```julia
 req_hold = OutputRequest("Leaf", :A; name=:A_hourly, process=:assim, policy=HoldLast())
 req_day = OutputRequest("Leaf", :A; name=:A_daily_sum, process=:assim, policy=Integrate(), clock=ClockSpec(24.0, 1.0))
-out = collect_outputs(sim, [req_hold, req_day]; sink=DataFrame)
+run!(sim, meteo; multirate=true, tracked_outputs=[req_hold, req_day], executor=SequentialEx())
+out = collect_outputs(sim; sink=DataFrame)
+
+# or directly:
+out_status, out = run!(
+    sim,
+    meteo;
+    multirate=true,
+    tracked_outputs=[req_hold, req_day],
+    return_requested_outputs=true,
+)
 ```
 
 - `process` is optional when the source is canonical and unique.

docs/src/model_execution.md

@@ -100,7 +100,19 @@ req = OutputRequest("Leaf", :carbon_assimilation;
     clock=ClockSpec(24.0, 1.0)
 )
 
-exported = collect_outputs(sim, [req]; sink=DataFrame)
+run!(sim, meteo; multirate=true, tracked_outputs=[req], executor=SequentialEx())
+exported = collect_outputs(sim; sink=DataFrame)
 ```
 
-This is independent from `tracked_outputs` and allows per-variable resampling policies at export time.
+`tracked_outputs` accepts `OutputRequest` values for these resampled exports.
+You can also return them directly from `run!`:
+
+```julia
+out_status, exported = run!(
+    sim,
+    meteo;
+    multirate=true,
+    tracked_outputs=[req],
+    return_requested_outputs=true,
+)
+```

src/run.jl

@@ -22,12 +22,15 @@ multi-threaded way (`executor=ThreadedEx()`, the default), or in a distributed w
 - `outputs`: the outputs to get in dynamic for each node type of the MTG.
 - `multirate`: experimental feature flag enabling temporal cache-based input resolution for multiscale simulations.
 Current implementation supports `HoldLast` resolution only.
+- `return_requested_outputs`: when `true` in MTG multi-rate runs, return requested resampled outputs directly
+  as second return value.
+- `requested_outputs_sink`: sink used to materialize requested outputs when `return_requested_outputs=true`.
 
 # Returns 
 
-Modifies the status of the object in-place. Users may retrieve the results from the object using 
-the [`status`](https://virtualplantlab.github.io/PlantSimEngine.jl/stable/API/#PlantSimEngine.status-Tuple{Any}) 
-function (see examples).
+Returns status outputs (and optionally requested exports).
+For MTG multi-rate runs with `return_requested_outputs=true`, returns
+`(status_outputs, requested_outputs)`.
 
 # Details 
 
@@ -124,7 +127,9 @@ function run!(
     tracked_outputs=nothing,
     check=true,
     executor=ThreadedEx(),
-    multirate=false
+    multirate=false,
+    return_requested_outputs=false,
+    requested_outputs_sink=DataFrames.DataFrame
 )
     run!(
         DataFormat(object),
@@ -135,7 +140,9 @@ function run!(
         tracked_outputs,
         check,
         executor,
-        multirate
+        multirate,
+        return_requested_outputs,
+        requested_outputs_sink
     )
 end
 
@@ -153,11 +160,14 @@ function run!(
     tracked_outputs=nothing,
     check=true,
     executor=ThreadedEx(),
-    multirate=false
+    multirate=false,
+    return_requested_outputs=false,
+    requested_outputs_sink=DataFrames.DataFrame
 ) where {T<:Union{AbstractArray,AbstractDict},A}
 
     tracked_outputs isa OutputRequest && error("`OutputRequest` is only supported for MTG multi-rate simulations.")
     tracked_outputs isa AbstractVector{<:OutputRequest} && error("`OutputRequest` is only supported for MTG multi-rate simulations.")
+    return_requested_outputs && error("`return_requested_outputs=true` is only supported for MTG multi-rate simulations.")
 
     if executor != SequentialEx()
         @warn string(
@@ -191,11 +201,14 @@ function run!(
     tracked_outputs=nothing,
     check=true,
     executor=ThreadedEx(),
-    multirate=false
+    multirate=false,
+    return_requested_outputs=false,
+    requested_outputs_sink=DataFrames.DataFrame
 ) where {T<:ModelList}
 
     tracked_outputs isa OutputRequest && error("`OutputRequest` is only supported for MTG multi-rate simulations.")
     tracked_outputs isa AbstractVector{<:OutputRequest} && error("`OutputRequest` is only supported for MTG multi-rate simulations.")
+    return_requested_outputs && error("`return_requested_outputs=true` is only supported for MTG multi-rate simulations.")
 
     meteo_adjusted = adjust_weather_timesteps_to_given_length(get_status_vector_max_length(object.status), meteo)
     nsteps = get_nsteps(meteo_adjusted)
@@ -286,11 +299,14 @@ function run!(
     tracked_outputs=nothing,
     check=true,
     executor=ThreadedEx(),
-    multirate=false
+    multirate=false,
+    return_requested_outputs=false,
+    requested_outputs_sink=DataFrames.DataFrame
 ) where {T<:Union{AbstractArray,AbstractDict}}
 
     tracked_outputs isa OutputRequest && error("`OutputRequest` is only supported for MTG multi-rate simulations.")
     tracked_outputs isa AbstractVector{<:OutputRequest} && error("`OutputRequest` is only supported for MTG multi-rate simulations.")
+    return_requested_outputs && error("`return_requested_outputs=true` is only supported for MTG multi-rate simulations.")
 
     dep_graphs = [dep(obj) for obj in collect(values(object))]
     #obj_parallelizable = all([object_parallelizable(graph) for graph in dep_graphs])
@@ -376,17 +392,12 @@ end
 function _multirate_tracked_outputs(tracked_outputs)
     if isnothing(tracked_outputs)
         return nothing, OutputRequest[]
-    elseif tracked_outputs isa Dict
-        return tracked_outputs, OutputRequest[]
     elseif tracked_outputs isa OutputRequest
         return nothing, OutputRequest[tracked_outputs]
     elseif tracked_outputs isa AbstractVector{<:OutputRequest}
         return nothing, collect(tracked_outputs)
     end
-    error(
-        "For MTG multi-rate runs, `tracked_outputs` must be `nothing`, a Dict of status outputs, ",
-        "an `OutputRequest`, or a vector of `OutputRequest`."
-    )
+    return tracked_outputs, OutputRequest[]
 end
 
 function run!(
@@ -399,27 +410,36 @@ function run!(
     tracked_outputs=nothing,
     check=true,
     executor=ThreadedEx(),
-    multirate=false
+    multirate=false,
+    return_requested_outputs=false,
+    requested_outputs_sink=DataFrames.DataFrame
 )
     isnothing(nsteps) && (nsteps = get_nsteps(meteo))
     meteo_adjusted = adjust_weather_timesteps_to_given_length(nsteps, meteo)
     status_outputs, output_requests = _multirate_tracked_outputs(tracked_outputs)
     !multirate && !isempty(output_requests) && error("`OutputRequest` requires `multirate=true`.")
+    return_requested_outputs && !multirate && error("`return_requested_outputs=true` requires `multirate=true`.")
 
     # NOTE : replace_mapping_status_vectors_with_generated_models is assumed to have already run if used
     # otherwise there might be vector length conflicts with timesteps
     sim = GraphSimulation(object, mapping, nsteps=nsteps, check=check, outputs=status_outputs)
-    run!(
+    result = run!(
         sim,
         meteo_adjusted,
         constants,
         extra;
         check=check,
         executor=executor,
         multirate=multirate,
-        tracked_outputs=output_requests
+        tracked_outputs=output_requests,
+        return_requested_outputs=return_requested_outputs,
+        requested_outputs_sink=requested_outputs_sink
     )
 
+    if return_requested_outputs
+        return result
+    end
+
     return outputs(sim)
 end
 
@@ -432,7 +452,9 @@ function run!(
     tracked_outputs=nothing,
     check=true,
     executor=ThreadedEx(),
-    multirate=false
+    multirate=false,
+    return_requested_outputs=false,
+    requested_outputs_sink=DataFrames.DataFrame
 )
 
     dep_graph = object.dependency_graph
@@ -443,6 +465,8 @@ function run!(
         validate_canonical_publishers(object)
         prepare_output_requests!(object, tracked_outputs, timeline)
         configure_temporal_buffers!(object, timeline)
+    elseif return_requested_outputs
+        error("`return_requested_outputs=true` requires `multirate=true`.")
     end
 
     !isnothing(extra) && error("Extra parameters are not allowed for the simulation of an MTG (already used for statuses).")
@@ -475,6 +499,10 @@ function run!(
         resize!(outputs(object)[organ], index - 1)
     end
 
+    if return_requested_outputs
+        return outputs(object), collect_outputs(object; sink=requested_outputs_sink)
+    end
+
     return outputs(object)
 end
 

src/time/runtime/input_resolution.jl

@@ -36,18 +36,66 @@ function _resolved_windowed_value_for_source(
     samples = _resolution_samples(sim, key)
     isnothing(samples) && return nothing, false
 
-    vals = Any[]
-    for (ts, v) in samples
-        if ts >= t_start - 1e-8 && ts <= t_end + 1e-8
-            push!(vals, v)
-        end
-    end
-    isempty(vals) && return nothing, false
+    if policy isa Union{Integrate,Aggregate}
+        reducer = policy.reducer
+        if reducer isa Symbol
+            reducer in _WINDOW_REDUCER_SYMBOLS || error(
+                "Unsupported reducer symbol `$(reducer)`. Supported symbols are $(_WINDOW_REDUCER_SYMBOLS)."
+            )
+
+            found = false
+            n = 0
+            s = 0.0
+            first_v = 0.0
+            last_v = 0.0
+            min_v = 0.0
+            max_v = 0.0
+
+            for (ts, v) in samples
+                ts < t_start - 1e-8 && continue
+                ts > t_end + 1e-8 && continue
+                v isa Real || return nothing, false
+                vf = float(v)
+
+                if !found
+                    found = true
+                    first_v = vf
+                    min_v = vf
+                    max_v = vf
+                end
 
-    all(v -> v isa Real, vals) || return nothing, false
-    vals_real = [float(v) for v in vals]
+                last_v = vf
+                s += vf
+                n += 1
+                vf < min_v && (min_v = vf)
+                vf > max_v && (max_v = vf)
+            end
 
-    if policy isa Integrate || policy isa Aggregate
+            !found && return nothing, false
+
+            if reducer == :sum
+                return s, true
+            elseif reducer == :mean
+                return s / n, true
+            elseif reducer == :max
+                return max_v, true
+            elseif reducer == :min
+                return min_v, true
+            elseif reducer == :first
+                return first_v, true
+            elseif reducer == :last
+                return last_v, true
+            end
+        end
+
+        vals_real = Float64[]
+        for (ts, v) in samples
+            ts < t_start - 1e-8 && continue
+            ts > t_end + 1e-8 && continue
+            v isa Real || return nothing, false
+            push!(vals_real, float(v))
+        end
+        isempty(vals_real) && return nothing, false
         return _window_reduce(vals_real, policy), true
     end
 

test/test-multirate-output-export.jl

@@ -77,4 +77,46 @@ end
     )
     exported_auto = collect_outputs(sim_canonical; sink=DataFrame)
     @test exported_auto[:x_auto][:, :value] == [1.0, 2.0, 3.0, 4.0]
+
+    # Optional direct export return from run! on GraphSimulation.
+    sim_direct = PlantSimEngine.GraphSimulation(
+        mtg,
+        Dict(
+            "Leaf" => (
+                ModelSpec(MRExportSourceModel(Ref(0))) |>
+                TimeStepModel(1.0),
+            ),
+        ),
+        nsteps=4,
+        check=true,
+        outputs=Dict("Leaf" => (:X,)),
+    )
+    out_status, out_requested = run!(
+        sim_direct,
+        meteo4,
+        multirate=true,
+        executor=SequentialEx(),
+        tracked_outputs=[OutputRequest("Leaf", :X; name=:x_direct, policy=HoldLast())],
+        return_requested_outputs=true,
+    )
+    @test haskey(out_status, "Leaf")
+    @test out_requested[:x_direct][:, :value] == [1.0, 2.0, 3.0, 4.0]
+
+    # Optional direct export return from run! on MTG + mapping entry point.
+    out_status_mtg, out_requested_mtg = run!(
+        mtg,
+        Dict(
+            "Leaf" => (
+                ModelSpec(MRExportSourceModel(Ref(0))) |>
+                TimeStepModel(1.0),
+            ),
+        ),
+        meteo4;
+        multirate=true,
+        executor=SequentialEx(),
+        tracked_outputs=[OutputRequest("Leaf", :X; name=:x_mtg, policy=HoldLast())],
+        return_requested_outputs=true,
+    )
+    @test haskey(out_status_mtg, "Leaf")
+    @test out_requested_mtg[:x_mtg][:, :value] == [1.0, 2.0, 3.0, 4.0]
 end