Skip to content

Commit

Permalink
Update op weighted average (#231)
Browse files Browse the repository at this point in the history
* added weighted average operator for workflow
  • Loading branch information
cehbrecht committed Sep 27, 2023
1 parent 6c6aff5 commit aeb6b1e
Show file tree
Hide file tree
Showing 3 changed files with 17 additions and 0 deletions.
8 changes: 8 additions & 0 deletions rook/operator.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@
from rook.utils.average_utils import (
run_average_by_time,
run_average_by_dim,
run_weighted_average,
)
from rook.utils.subset_utils import run_subset
from rook.utils.concat_utils import run_concat
Expand Down Expand Up @@ -81,6 +82,13 @@ def _get_runner(self):
return run_average_by_dim


class WeightedAverage(Operator):
prefix = "weighted_average"

def _get_runner(self):
return run_weighted_average


class Concat(Operator):
prefix = "concat"

Expand Down
3 changes: 3 additions & 0 deletions rook/utils/average_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,5 +16,8 @@ def run_average_by_dim(args):
def run_weighted_average(args):
from daops.ops.average import average_over_dims

args["apply_fixes"] = False
args["dims"] = ["latitude", "longitude"]

result = average_over_dims(**args)
return result.file_uris
6 changes: 6 additions & 0 deletions rook/workflow.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@
from .operator import (
AverageByTime,
AverageByDimension,
WeightedAverage,
Subset,
Concat,
)
Expand Down Expand Up @@ -84,6 +85,7 @@ def __init__(self, output_dir):
self.subset_op = Subset(output_dir)
self.average_time_op = AverageByTime(output_dir)
self.average_dim_op = AverageByDimension(output_dir)
self.weighted_average_op = WeightedAverage(output_dir)
self.prov = Provenance(output_dir)

def validate(self, wfdoc):
Expand Down Expand Up @@ -148,6 +150,10 @@ def _run_step(self, step_id, step, inputs=None):
collection = step["in"]["collection"]
result = self.average_dim_op.call(step["in"])
self.prov.add_operator(step_id, step["in"], collection, result)
elif "weighted_average" == step["run"]:
collection = step["in"]["collection"]
result = self.weighted_average_op.call(step["in"])
self.prov.add_operator(step_id, step["in"], collection, result)
elif "concat" == step["run"]:
collection = step["in"]["collection"]
result = self.concat_op.call(step["in"])
Expand Down

0 comments on commit aeb6b1e

Please sign in to comment.