small things

Signed-off-by: Ayush Kamat <[email protected]>

latch_cli/centromere/ctx.py

@@ -1,5 +1,7 @@
+import os
 import re
 import sys
+import time
 import traceback
 from dataclasses import dataclass
 from pathlib import Path
@@ -220,8 +222,6 @@ def __init__(
                                 elif system == "Darwin":
                                     res = subprocess.run(["open", new_meta]).returncode
                                 elif system == "Windows":
-                                    import os
-
                                     res = os.system(str(new_meta.resolve()))
                                 else:
                                     res = None
@@ -283,6 +283,10 @@ def __init__(
             if not self.disable_auto_version:
                 hash = hash_directory(self.pkg_root)
                 self.version = f"{self.version}-{hash[:6]}"
+
+                if os.environ.get("LATCH_NEW_VERSION_ALWAYS") is not None:
+                    self.version = f"{self.version}-{int(time.monotonic())}"
+
                 click.echo(f"  {self.version}\n")
 
             if self.nucleus_check_version(self.version, self.workflow_name):

latch_cli/extras/common/utils.py

@@ -1,4 +1,5 @@
 import textwrap
+import types
 from typing import Type, Union, get_args, get_origin
 
 from typing_extensions import TypeGuard
@@ -25,6 +26,9 @@ def is_primitive_value(val: object) -> TypeGuard[Union[None, str, bool, int, flo
 
 
 def type_repr(t: Type, *, add_namespace: bool = False) -> str:
+    if getattr(t, "__name__", None) == "NoneType":
+        return "None"
+
     if is_primitive_type(t) or t is LatchFile or t is LatchDir:
         return t.__name__
 

latch_cli/extras/nextflow/dag.py

@@ -4,7 +4,8 @@
     from functools import lru_cache as cache
 
 import json
-from dataclasses import dataclass, field
+import sys
+from dataclasses import asdict, dataclass, field
 from enum import Enum
 from pathlib import Path
 from typing import Dict, List, Optional, Set, Tuple, TypedDict
@@ -28,6 +29,11 @@ class Vertex:
     id: str
     label: str
     type: VertexType
+    statement: str
+    ret: List[str] = field(hash=False)
+    outputNames: List[str] = field(hash=False)
+    module: str
+    unaliased: str
 
 
 @dataclass(frozen=True)
@@ -42,7 +48,11 @@ class _VertexContentJson(TypedDict):
     id: str
     label: str
     type: VertexType
-    processMeta: Optional[Dict]
+    statement: str
+    ret: List[str]
+    outputNames: List[str]
+    module: str
+    unaliased: str
 
 
 class _VertexJson(TypedDict):
@@ -80,13 +90,8 @@ def from_path(cls, p: Path) -> Self:
         vertices: List[Vertex] = []
         for v in payload["vertices"]:
             c = v["content"]
-            vertices.append(
-                Vertex(
-                    id=c["id"],
-                    label=identifier_from_str(c["label"])[:128],
-                    type=c["type"],
-                )
-            )
+            c["label"] = identifier_from_str(c["label"])[:128]
+            vertices.append(Vertex(**c))
 
         edges: List[Edge] = []
         edge_set: Set[Tuple[str, str]] = set()
@@ -98,7 +103,8 @@ def from_path(cls, p: Path) -> Self:
                 # disallow multiple edges
                 continue
 
-            edges.append(Edge(**c))
+            edge = Edge(**c)
+            edges.append(edge)
             edge_set.add(t)
 
         return cls(vertices, edges)
@@ -120,14 +126,14 @@ def dest(self, e: Edge) -> Vertex:
         return self._vertices_by_id()[e.dest]
 
     @cache
-    def ancestors(self) -> Dict[Vertex, List[Vertex]]:
-        res: Dict[Vertex, List[Vertex]] = {}
+    def ancestors(self) -> Dict[Vertex, List[Tuple[Vertex, Edge]]]:
+        res: Dict[Vertex, List[Tuple[Vertex, Edge]]] = {}
         for v in self.vertices:
             res[v] = []
 
         by_id = self._vertices_by_id()
         for edge in self.edges:
-            res[by_id[edge.dest]].append(by_id[edge.src])
+            res[by_id[edge.dest]].append((by_id[edge.src], edge))
 
         return res
 
@@ -144,14 +150,14 @@ def inbound_edges(self) -> Dict[Vertex, List[Edge]]:
         return res
 
     @cache
-    def descendants(self) -> Dict[Vertex, List[Vertex]]:
-        res: Dict[Vertex, List[Vertex]] = {}
+    def descendants(self) -> Dict[Vertex, List[Tuple[Vertex, Edge]]]:
+        res: Dict[Vertex, List[Tuple[Vertex, Edge]]] = {}
         for v in self.vertices:
             res[v] = []
 
         by_id = self._vertices_by_id()
         for edge in self.edges:
-            res[by_id[edge.src]].append(by_id[edge.dest])
+            res[by_id[edge.src]].append((by_id[edge.dest], edge))
 
         return res
 
@@ -214,20 +220,18 @@ def _resolve_subworkflows_helper(
 
             sub_dag = dags[v.label]
             for sub_v in sub_dag.vertices:
-                new_vertices.append(
-                    Vertex(
-                        id="_".join([v.id, sub_v.id]),
-                        label=sub_v.label,
-                        type=sub_v.type,
-                    )
-                )
+                args = asdict(sub_v)
+                args["id"] = "_".join([v.id, sub_v.id])
+
+                new_vertices.append(Vertex(**args))
 
             for sub_e in sub_dag.edges:
                 new_edges.append(
                     Edge(
                         label=sub_e.label,
                         src="_".join([v.id, sub_e.src]),
                         dest="_".join([v.id, sub_e.dest]),
+                        branch=sub_e.branch,
                     )
                 )
 
@@ -254,6 +258,7 @@ def _resolve_subworkflows_helper(
                             label=e.label,
                             src=src,
                             dest=dest,
+                            branch=e.branch,
                         )
                     )
 
@@ -285,7 +290,7 @@ def resolve_subworkflows(cls, dags: Dict[str, Self]) -> Dict[str, Self]:
         return res
 
     def _toposort_helper(self, cur: Vertex, res: List[Vertex]):
-        for x in self.ancestors()[cur]:
+        for x, _ in self.ancestors()[cur]:
             self._toposort_helper(x, res)
 
         res.append(cur)

latch_cli/extras/nextflow/serialize.py

@@ -8,4 +8,4 @@ def serialize_nf(
     image_name: str,
     dkr_repo: str,
 ):
-    serialize(nf_wf, output_dir, image_name, dkr_repo)
+    serialize(nf_wf, output_dir, image_name, dkr_repo, write_spec=True)