Use of metaclass

climetlab/sources/multi.py

@@ -108,7 +108,7 @@ def _load_sources(self, sources):
                 has_callables = True
                 callables.append(s)
             else:
-                callables.append(lambda: s)
+                callables.append(lambda *args, **kwargs: s)
 
         if not has_callables:
             return sources
@@ -117,12 +117,12 @@ def _load_sources(self, sources):
         if nthreads < 2:
             return [s() for s in sources]
 
-        def _call(s):
-            return s()
+        def _call(s, *args, **kwargs):
+            return s(*args, **kwargs)
 
         with SoftThreadPool(nthreads=nthreads) as pool:
 
-            futures = [pool.submit(_call, s) for s in sources]
+            futures = [pool.submit(_call, s, _observer=pool) for s in sources]
 
             iterator = (f.result() for f in futures)
             sources = list(tqdm(iterator, leave=False, total=len(futures)))

climetlab/sources/multi_url.py

@@ -7,13 +7,9 @@
 # nor does it submit to any jurisdiction.
 #
 
-import os
-
-from climetlab.core.thread import SoftThreadPool
-from climetlab.utils import tqdm
+from climetlab import load_source
 
 from .multi import MultiSource
-from .url import Url
 
 
 class MultiUrl(MultiSource):
@@ -26,38 +22,17 @@ def __init__(self, urls, *args, filter=None, merger=None, force=None, **kwargs):
 
         assert len(urls)
 
-        urls = sorted(urls)
-
-        nthreads = min(self.settings("number-of-download-threads"), len(urls))
-
-        if nthreads < 2:
-            sources = [
-                Url(
-                    url,
-                    filter=filter,
-                    merger=merger,
-                    force=force,
-                )
-                for url in urls
-            ]
-        else:
-            with SoftThreadPool(nthreads=nthreads) as pool:
-
-                futures = [
-                    pool.submit(
-                        Url,
-                        url,
-                        filter=filter,
-                        merger=merger,
-                        observer=pool,
-                        force=force,
-                    )
-                    for url in urls
-                ]
-
-                iterator = (f.result() for f in futures)
-                sources = list(tqdm(iterator, leave=True, total=len(urls)))
-
-                assert all(os.path.exists(s.path) for s in sources)
+        sources = [
+            load_source(
+                "url",
+                url,
+                filter=filter,
+                merger=merger,
+                force=force,
+                # Load lazily so we can do parallel downloads
+                lazily=True,
+            )
+            for url in sorted(urls)
+        ]
 
         super().__init__(sources, filter=filter, merger=merger)

climetlab/utils/lazy.py

@@ -26,7 +26,9 @@ def __init__(self, name, *args, **kwargs):
         self.kwargs = kwargs
         self._source = None
 
-    def __call__(self):
+    def __call__(self, **kwargs):
+        assert self._source is None
+        self.kwargs.update(kwargs)
         return self.source
 
     @property