This chapter provides information about passing aggregates of jobs to operation functions.
aggregator creates generators of aggregates for use in operation functions via
Such functions may accept a variable number of positional arguments,
*jobs is unpacked into an aggregate, defined as an ordered tuple of jobs.
See also the Python documentation about argument unpacking.
# project.py from flow import FlowProject, aggregator class Project(FlowProject): pass @Project.operation(aggregator=aggregator()) def op1(*jobs): print("Number of jobs in aggregate:", len(jobs)) @Project.operation def op2(job): pass if __name__ == "__main__": Project().main()
aggregator is used with the default arguments, it will create a single aggregate containing all the jobs present in the project.
In the example above,
op1 is an aggregate operation where all the jobs present in the project are passed as a variable number of positional arguments (via
op2 is an operation where only a single job is passed as an argument.
The concept of aggregation may be easier to understand if one realizes that “normal” operation functions are equivalent to aggregate operation functions with an aggregate group size of one job.
Types of Aggregation
Currently, signac-flow allows users to aggregate jobs in the following ways:
All jobs: All of the project’s jobs are passed to the operation function.
Group by state point key(s): The aggregates are grouped by one or more state point keys.
Group by arbitrary key function: The aggregates are grouped by keys determined by a key-function that expects an instance of
Joband return the grouping key.
Grouping into aggregates of a specific size.
Using a completely custom aggregator function when even greater flexibility is needed.
groupby() allows users to aggregate jobs by grouping them by a state point key, an iterable of state point keys whose values define the groupings, or an arbitrary callable of
@Project.operation(aggregator=aggregator.groupby("temperature")) def op3(*jobs): pass
In the above example, the jobs will be aggregated based on the state point key
So, all the jobs having the same value of temperature in their state point will be aggregated together.
groupsof() allows users to aggregate jobs by generating aggregates of a given size.
@Project.operation(aggregator=aggregator.groupsof(2)) def op4(job1, job2=None): pass
In the above example, the jobs will get aggregated in groups of 2 and hence, up to two jobs will be passed as arguments at once.
In case the number of jobs in the project in this example is odd, there will be one aggregate containing only a single job.
In general, the last aggregate from
groupsof() will contain the remaining jobs if the aggregate size does not evenly divide the number of jobs in the project.
If a remainder is expected and valid, users should make sure that the operation function can be called with the reduced number of arguments (e.g. by using
*jobs or providing default arguments as shown above).
Sorting jobs for aggregation
Aggregators allow users to sort the jobs before creating aggregates with the
The sorting order can be defined with the
By default, when no
sort_by parameter is specified, the order of the jobs will be decided by the iteration order of the signac project.
@Project.operation( aggregator=aggregator.groupsof(2, sort_by="temperature", sort_ascending=False) ) def op5(*jobs): pass
In the above example, all the jobs will be sorted by the state point parameter
"temperature" in descending order and then be aggregated as groups of 2.
Selecting jobs for aggregation
signac-flow allows users to selectively choose which jobs to pass into operation functions. This can be used to generate aggregates from only the selected jobs, excluding any jobs that do not meet the selection criteria.
@Project.operation(aggregator=aggregator(select=lambda job: job.sp.temperature > 0)) def op6(*jobs): pass
Similar to the concept of a job id, an aggregate id is a unique hash identifying an aggregate of jobs. The aggregate id is sensitive to the order of the jobs in the aggregate.
The id of an aggregate containing one job is that job’s id.
In order to distinguish between an aggregate id and a job id, the id of aggregates with more than one job will always have a prefix
Users can generate the aggregate id of an aggregate using
Users can also pass an aggregate id to the
--job-id command line flag provided by signac-flow in
Aggregation with FlowGroups
In order to associate an aggregator object with a
FlowGroup, signac-flow provides a
group_aggregator parameter in
By default, no aggregation takes place for a
All the operations in a
FlowGroup will use the same
aggregator object provided to the group’s
# project.py from flow import FlowProject, aggregator class Project(FlowProject): pass group = Project.make_group("agg-group", group_aggregator=aggregator()) @group @Project.operation(aggregator=aggregator()) def op1(*jobs): pass @group @Project.operation def op2(*jobs): pass if __name__ == "__main__": Project().main()
In the above example, when the group
agg-group is executed using
python project.py run -o agg-group, all the jobs in the project are passed as positional arguments for both
op1 is executed using
python project.py run -o op1, all the jobs in the project are passed as positional arguments because a
aggregator is associated with the operation function
op1 (separately from the aggregator used for
op2 is executed using
python project.py run -o op2, only a single job is passed as an argument because no
aggregator is associated with the operation function