Multi agent calibration with pyswarm PSO integration.

python/calibration.py

@@ -2,13 +2,18 @@
 import yaml
 from os import chdir
 from pathlib import Path
-from ngen.cal.configuration import General, Model
-from ngen.cal.meta import CalibrationMeta
-from ngen.cal.search import dds, dds_set
+from ngen.cal.configuration import General
+from ngen.cal.search import dds, dds_set, pso_search
+from ngen.cal.strategy import Algorithm
+from ngen.cal.agent import Agent
 
-def main(general: General, model: Model):
+from typing import TYPE_CHECKING
+if TYPE_CHECKING:
+    from typing import Mapping, Any
+
+def main(general: General, model_conf: Mapping[str, Any]):
     #seed the random number generators if requested
-    if( general.random_seed is not None):
+    if general.random_seed is not None:
         import random
         random.seed(general.random_seed)
         import numpy as np
@@ -20,31 +25,42 @@ def main(general: General, model: Model):
     TODO calibrate each "catcment" independely, but there may be something interesting in grouping various formulation params
     into a single variable vector and calibrating a set of heterogenous formultions...
     """
-
-    meta = CalibrationMeta(model, general)
-    start_iteration = general.start_iteration
-    if general.restart:
-        start_iteration = meta.restart()
-
+    start_iteration = 0
+    #Initialize the starting agent
+    agent = Agent(model_conf, general.workdir, general.log, general.restart, general.strategy.parameters)
+    if general.strategy.algorithm == Algorithm.dds:
+        func = dds_set #FIXME what about explicit/dds
+        start_iteration = general.start_iteration
+        if general.restart:
+            start_iteration = agent.restart()
+    elif general.strategy.algorithm == Algorithm.pso: #TODO how to restart PSO?
+        if agent.model.strategy != "uniform":
+            print("Can only use PSO with the uniform model strategy")
+            return
+        if general.restart:
+            print("Restart not supported for PSO search, starting at 0")
+        func = pso_search
 
     print("Starting Iteration: {}".format(start_iteration))
-    print("Starting Best param: {}".format(meta.best_params))
-    print("Starting Best score: {}".format(meta.best_score))
-    print("Starting DDS loop")
+    # print("Starting Best param: {}".format(meta.best_params))
+    # print("Starting Best score: {}".format(meta.best_score))
+    print("Starting calibration loop")
+
 
     #NOTE this assumes we calibrate each catchment independently, it may be possible to design an "aggregate" calibration
     #that works in a more sophisticated manner.
-    if model.strategy == 'explicit':
-        for catchment in model.hy_catchments:
-            dds(start_iteration, general.iterations, catchment, meta)
+    if agent.model.strategy == 'explicit': #FIXME this needs a refactor...should be able to use a calibration_set with explicit loading
+        for catchment in agent.model.adjustables:
+            dds(start_iteration, general.iterations, catchment, agent)
 
-    elif model.strategy == 'independent':
-        for catchment_set in model.hy_catchments:
-            dds_set(start_iteration, general.iterations, catchment_set, meta)
+    elif agent.model.strategy == 'independent':
+        #for catchment_set in agent.model.adjustables:
+        func(start_iteration, general.iterations, agent)
 
-    elif model.strategy == 'uniform':
-        for catchment_set in model.hy_catchments:
-            dds_set(start_iteration, general.iterations, catchment_set, meta)
+    elif agent.model.strategy == 'uniform':
+        #for catchment_set in agent.model.adjustables:
+        #    func(start_iteration, general.iterations, catchment_set, agent)
+        func(start_iteration, general.iterations, agent)
 
 if __name__ == "__main__":
 
@@ -66,5 +82,5 @@ def main(general: General, model: Model):
     # change directory to workdir
     chdir(general.workdir)
 
-    model = Model(model=conf['model']).model
-    main(general, model)
+    #model = Model(model=conf['model']).model
+    main(general, conf['model'])

python/example_config.yaml

@@ -12,24 +12,24 @@ general:
       type: estimation
       #defaults to dds (currently, the only supported algorithm)
       algorithm: "dds"
-      # choices are "kling_gupta", "nnse", "custom", "single_peak", "volume"
-      #objective: nnse
-      # one can also provide a module path to any function that takes
-      # obs, sim array-like arguments and produces a single value float
-      # for example, nnse above could be called this way
-      #objective: "ngen_cal.objectives.normalized_nash_sutcliffe"
-      # Can choose to minimize the objective function or maximixe it
-      # choices are 'min', 'max'.
-      # An an explicit floating point value can be supplied instead, and the
-      # optmization will attempt to converge on that value
-      # Default: min
-      #target: 0.0
+      # To adjuts the neighborhood size parameter of the dds algorithm, uncomment the following two lines
+      #parameters:
+      #  neighborhood: 0.5
+      # To use PSO optmization, select the pso algorithm and configure its parameters as follows
+      #algorithm: "pso"
+      #parameters:
+      #    pool: 4 #number of processors to use (by default, uses 1)
+      #    particles: 8 #number of particles to use (by default, uses 4)
+      #    options: #the PSO parameters (defaults to c1: 0.5, c2: 0.3, w:0.9)
+      #        c1: 0.1
+      #        c2: 0.1
+      #        w: 0.42
     #in theory, could do a senstivity strategy like this
     #sensitivity:
     #  objective: null
 
   # Attempt to restart a previous calibration.
-  # Will look for log and parameter information in the `workdir` to restart
+  # Will look for log and parameter information in an exististing worker dir to restart
   # If the required restart information cannot be found, will start back from the 0 iteration
   # Defaults to false
   #restart: false
@@ -41,17 +41,10 @@ general:
   # defaults to ./
   #workdir: ./
 
-  # This is the range of the hydrograph dates to run the objective function over
-  # To evaluate the entire period, make sure this range is the same as the simulation
-  # run time reange.  Future versions this will become optional and default to evaluating
-  # the entire range
-  evaluation_start: '2015-12-15 12:00:00'
-  evaluation_stop: '2015-12-30 23:00:00'
-
-  # logging locations
-  # location to output stdout/stderr from model runs
-  # defaults to None, which sends all output to /dev/null
-  log_file: "test_log"
+  # Enable model runtime logging (captures standard out and error and writes to file)
+  # logs will be written to <model.type>.log when enabled
+  # defaults to False, which sends all output to /dev/null
+  log: True
   # Name of the best parameter log file, defaults to `name`_best_params.txt
   #parameter_log_file: null
   # Name of the objective function log file, defaults to `name`_objective.txt
@@ -93,11 +86,6 @@ cfe_params: &cfe_params
 #       min: 0.0
 #       max: 21.9
 #       init: 4.05
-  - 
-      name: multiplier
-      min: 10.7
-      max: 9997.3
-      init: 100.0
   - 
       name: expon
       min: 1.0
@@ -138,3 +126,21 @@ model:
     strategy: independent
     params: 
         CFE: *cfe_params
+
+    eval_params:
+      # This is the range of the hydrograph dates to run the objective function over
+      # To evaluate the entire period, you can comment these lines out
+      #evaluation_start: '2015-12-15 12:00:00'
+      #evaluation_stop: '2015-12-30 23:00:00'
+      # choices are "kling_gupta", "nnse", "custom", "single_peak", "volume"
+      objective: "kling_gupta"
+      # one can also provide a module path to any function that takes
+      # obs, sim array-like arguments and produces a single value float
+      # for example, nnse above could be called this way
+      #objective: "ngen_cal.objectives.normalized_nash_sutcliffe"
+      # Can choose to minimize the objective function or maximixe it (only when using the DDS algorithm)
+      # choices are 'min', 'max'.
+      # An an explicit floating point value can be supplied instead, and the
+      # optmization will attempt to converge on that value
+      # Default: min
+      #target: 0.0

python/ngen_cal/changelog.md

@@ -0,0 +1,44 @@
+# V 0.2.0
+- Allow ngen configuration to specify routing output file name to look for.
+- `Evaluatable` objects are now responsible for maintaining the `evaluation_parms` property, including `evaluation_range`
+- `evaluation_range` removed from `Meta`
+- `eval_params` bundeled and moved to model config from global, these include the following configuration keys
+    - `evaluation_start`
+    - `evaluation_stop`
+    - `objective`
+- Add `bounds` property to `Adjustable` interface
+- Use an `Agent` abstraction to connect model runtime, calibration objects, and evaluation criteria
+- All model execution now happens in automatically generated subdirectory
+- Introduce PSO global best serach, only applicable for `uniform` stragegy
+- Allow additional search algorithm parameters to be configured by the user
+    - For DDS, the configuration can supply the `neighborhood` parameter as follows, if not supplied, a default of 0.2 is used
+    ```yaml
+        strategy: 
+        type: estimation
+        algorithm: dds
+        parameters:
+            neighborhood: 0.5
+    ```
+    - For PSO, the configuration can supply the following parameters
+    ```yaml
+        strategy: 
+        type: estimation
+        algorithm: "pso"
+        parameters:
+            pool: 4 #number of processors to use (by default, uses 1)
+            particles: 8 #number of particles to use (by default, uses 4)
+            options: #the PSO parameters (defaults to c1: 0.5, c2: 0.3, w:0.9)
+                c1: 0.1
+                c2: 0.1
+                w: 0.42
+    ```
+- Restart semantics have changed slightly.  Restart is only supported for DDS search, with the following caveats:
+    If a user starts with an independent calibration strategy
+    then restarts with a uniform strategy, this will "work" but probably shouldn't.
+    it works cause the independent writes a param df for the nexus that uniform also uses,
+    so data "exists" and it doesn't know its not conistent...
+    Conversely, if you start with uniform then try independent, it will start back at
+    0 correctly since not all basin params can be loaded.
+    There are probably some similar issues with explicit and independent, since they have
+    similar data semantics.
+- Fix issue with independent strategy not writing ngen forcing configs for target catchments in a way that ngen can handle.

python/ngen_cal/src/ngen/cal/__init__.py

@@ -8,5 +8,5 @@ def pyobject_schema(cls, field_schema):
 from .configuration import General, Model
 from .calibratable import Calibratable, Adjustable, Evaluatable
 from .calibration_set import CalibrationSet, UniformCalibrationSet
-from .meta import CalibrationMeta
+from .meta import JobMeta
 from .plot import *

python/ngen_cal/src/ngen/cal/__main__.py

@@ -2,13 +2,18 @@
 import yaml
 from os import chdir
 from pathlib import Path
-from .configuration import General, Model
-from .meta import CalibrationMeta
-from .search import dds, dds_set
+from ngen.cal.configuration import General
+from ngen.cal.search import dds, dds_set, pso_search
+from ngen.cal.strategy import Algorithm
+from ngen.cal.agent import Agent
 
-def main(general: General, model: Model):
+from typing import TYPE_CHECKING
+if TYPE_CHECKING:
+    from typing import Mapping, Any
+
+def main(general: General, model_conf: Mapping[str, Any]):
     #seed the random number generators if requested
-    if( general.random_seed is not None):
+    if general.random_seed is not None:
         import random
         random.seed(general.random_seed)
         import numpy as np
@@ -20,31 +25,42 @@ def main(general: General, model: Model):
     TODO calibrate each "catcment" independely, but there may be something interesting in grouping various formulation params
     into a single variable vector and calibrating a set of heterogenous formultions...
     """
-
-    meta = CalibrationMeta(model, general)
-    start_iteration = general.start_iteration
-    if general.restart:
-        start_iteration = meta.restart()
-
+    start_iteration = 0
+    #Initialize the starting agent
+    agent = Agent(model_conf, general.workdir, general.log, general.restart, general.strategy.parameters)
+    if general.strategy.algorithm == Algorithm.dds:
+        func = dds_set #FIXME what about explicit/dds
+        start_iteration = general.start_iteration
+        if general.restart:
+            start_iteration = agent.restart()
+    elif general.strategy.algorithm == Algorithm.pso: #TODO how to restart PSO?
+        if agent.model.strategy != "uniform":
+            print("Can only use PSO with the uniform model strategy")
+            return
+        if general.restart:
+            print("Restart not supported for PSO search, starting at 0")
+        func = pso_search
 
     print("Starting Iteration: {}".format(start_iteration))
-    print("Starting Best param: {}".format(meta.best_params))
-    print("Starting Best score: {}".format(meta.best_score))
-    print("Starting DDS loop")
+    # print("Starting Best param: {}".format(meta.best_params))
+    # print("Starting Best score: {}".format(meta.best_score))
+    print("Starting calibration loop")
+
 
     #NOTE this assumes we calibrate each catchment independently, it may be possible to design an "aggregate" calibration
     #that works in a more sophisticated manner.
-    if model.strategy == 'explicit':
-        for catchment in model.hy_catchments:
-            dds(start_iteration, general.iterations, catchment, meta)
+    if agent.model.strategy == 'explicit': #FIXME this needs a refactor...should be able to use a calibration_set with explicit loading
+        for catchment in agent.model.adjustables:
+            dds(start_iteration, general.iterations, catchment, agent)
 
-    elif model.strategy == 'independent':
-        for catchment_set in model.hy_catchments:
-            dds_set(start_iteration, general.iterations, catchment_set, meta)
+    elif agent.model.strategy == 'independent':
+        #for catchment_set in agent.model.adjustables:
+        func(start_iteration, general.iterations, agent)
 
-    elif model.strategy == 'uniform':
-        for catchment_set in model.hy_catchments:
-            dds_set(start_iteration, general.iterations, catchment_set, meta)
+    elif agent.model.strategy == 'uniform':
+        #for catchment_set in agent.model.adjustables:
+        #    func(start_iteration, general.iterations, catchment_set, agent)
+        func(start_iteration, general.iterations, agent)
 
 if __name__ == "__main__":
 
@@ -53,7 +69,6 @@ def main(general: General, model: Model):
 
     # get the command line parser
     parser = argparse.ArgumentParser(
-        "ngen.cal",
         description='Calibrate catchments in NGEN NWM architecture.')
     parser.add_argument('config_file', type=Path,
                         help='The configuration yaml file for catchments to be operated on')
@@ -67,5 +82,5 @@ def main(general: General, model: Model):
     # change directory to workdir
     chdir(general.workdir)
 
-    model = Model(model=conf['model']).model
-    main(general, model)
+    #model = Model(model=conf['model']).model
+    main(general, conf['model'])

python/ngen_cal/src/ngen/cal/_version.py

@@ -1 +1 @@
-__version__ = '0.1.1'
+__version__ = '0.2.0'