add upload

README.md

@@ -66,3 +66,15 @@ use the cluster with
 from dask.distributed import Client
 client = Client("<your server's wireguard ip>:31337")
 ```
+
+### time for magic
+
+copy `leylines-support/02-dask.py` into `~/.ipython/profile_default/startup`
+
+this provides 2 new spells: `%dask` connects to your cluster, and `%daskworker` splits off a new
+ipython console on a worker selected by having free RAM available and not being busy. this is useful
+for ad-hoc code testing on a real worker
+
+%dask also installs `client`, a reference to the client, and `tqdmprogress`, which can be used in
+place of `distributed.diagnostics.progress` for a task monitor using `tqdm`, and `upload` which
+uploads a file and returns a delayed function which will fetch the filename on a worker

leylines-support/02-dask.ipy

@@ -1,3 +1,4 @@
+# vim: ft=python
 ipy = __import__("IPython")
 @ipy.core.magic.register_line_magic
 @ipy.core.magic.needs_local_scope
@@ -33,9 +34,22 @@ def dask(line, local_ns):
 
     local_ns['tqdmprogress'] = tqdmprogress
 
+    def upload(client, file):
+        import dask
+        import distributed
+        import os
+        name = os.path.basename(file)
+        client.upload_file(file)
+
+        def get_file():
+            return os.path.join(distributed.get_worker().local_directory, name)
+
+        return dask.delayed(get_file)()
+
     try:
         default = default_client()
         local_ns['client'] = default
+        local_ns['upload'] = lambda file: upload(client, file)
         return
     except ValueError:
         pass
@@ -58,6 +72,7 @@ def dask(line, local_ns):
         else:
             print(f"{node.name} ({node.ip}): down")
     local_ns['client'] = client
+    local_ns['upload'] = lambda file: upload(client, file)
 
 @ipy.core.magic.register_line_magic
 @ipy.core.magic.needs_local_scope