Implement a first working version of a dask worker

environment.yml

@@ -7,8 +7,10 @@ channels:
 dependencies:
   - black >=22,<23
   - click
+  - dask
   - humanfriendly
   - isort
+  - more-itertools
   - pip
   - psutil
   - pyflatsurf

flatsurvey/jobs/orbit_closure.py

@@ -240,6 +240,7 @@ def deform(self, deformation):
             ),
         }
 
+    # TODO: Probably all command() implementations can be removed now.
     def command(self):
         command = [self.name()]
         if self._limit != self.DEFAULT_LIMIT:

flatsurvey/pipeline/util.py

@@ -110,21 +110,30 @@ def wrap(**kwargs):
             f"provide_{name}",
         )
         provider.__module__ = "__main__"
-        binding = type(
+
+        binding_type = type(
             f"Partial{prototype.__name__}Binding",
             (pinject.BindingSpec,),
             {
                 f"provide_{name}": provider,
                 "__repr__": lambda self: f"{name} binding to {prototype.__name__}",
+                "__reduce__": lambda self: (PartialBindingSpec_unpickle, ((prototype, name, scope), kwargs))
             },
-        )()
+        )
+
+        binding = binding_type()
         binding.name = name
         binding.scope = scope or "DEFAULT"
         return binding
 
     return wrap
 
 
+def PartialBindingSpec_unpickle(outer_arguments, inner_arguments):
+    return PartialBindingSpec(*outer_arguments)(**inner_arguments)
+
+
+# TODO: Why are the args different between these functions?
 def FactoryBindingSpec(name, prototype, scope=None):
     r"""
     Return a BindingSpec that calls ``prototype`` as a provider for ``name``.
@@ -148,17 +157,29 @@ def FactoryBindingSpec(name, prototype, scope=None):
         f"provide_{name}",
     )
     provider.__module__ = "__main__"
-    binding = type(
+    binding_type = type(
         f"{name}FactoryBinding",
         (pinject.BindingSpec,),
-        {f"provide_{name}": provider, "__repr__": lambda self: f"{name}->{prototype}"},
-    )()
+        {
+            f"provide_{name}": provider,
+            "__repr__": lambda self: f"{name}->{prototype}",
+            # TODO: This is not possible.
+            # "__reduce__": lambda self: (FactoryBindingSpec_unpickle, (name, prototype, scope))
+        },
+    )
+
+    binding = binding_type()
     binding.name = name
     binding.scope = scope or "DEFAULT"
 
     return binding
 
 
+# TODO: This is not possible.
+# def FactoryBindingSpec_unpickle(args):
+#     return FactoryBidingSpec(*args)
+
+
 def provide(name, objects):
     src = compile(
         f"""