Valay/real time cards sidecar-updates

metaflow/plugins/cards/card_cli.py

@@ -1,12 +1,19 @@
 from metaflow.client import Task
 from metaflow import JSONType, namespace
-from metaflow.exception import CommandException
+from metaflow.util import resolve_identity
+from metaflow.exception import (
+    CommandException,
+    MetaflowNotFound,
+    MetaflowNamespaceMismatch,
+)
 import webbrowser
 import re
 from metaflow._vendor import click
 import os
 import json
+import uuid
 import signal
+import inspect
 import random
 from contextlib import contextmanager
 from functools import wraps
@@ -375,14 +382,45 @@ def wrapper(*args, **kwargs):
     return wrapper
 
 
-def render_card(mf_card, task, timeout_value=None):
-    rendered_info = None
+def update_card(mf_card, mode, task, data, timeout_value=None):
+    def _reload_token():
+        if data["render_seq"] == "final":
+            # final data update should always trigger a card reload to show
+            # the final card, hence a different token for the final update
+            return "final"
+        elif mf_card.RELOAD_POLICY == mf_card.RELOAD_POLICY_ALWAYS:
+            return "render-seq-%s" % data["render_seq"]
+        elif mf_card.RELOAD_POLICY == mf_card.RELOAD_POLICY_NEVER:
+            return "never"
+        elif mf_card.RELOAD_POLICY == mf_card.RELOAD_POLICY_ONCHANGE:
+            return mf_card.reload_content_token(task, data)
+
+    def _add_token_html(html):
+        if html is not None:
+            return html.replace(mf_card.RELOAD_POLICY_TOKEN, _reload_token())
+
+    def _add_token_json(json_msg):
+        if json_msg is not None:
+            return {"reload_token": _reload_token(), "data": json_msg}
+
+    def _call():
+        # compatibility with old render()-method that doesn't accept the data arg
+        new_render = "data" in inspect.getfullargspec(mf_card.render).args
+        if mode == "render":
+            if new_render:
+                return _add_token_html(mf_card.render(task, data))
+            else:
+                return _add_token_html(mf_card.render(task))
+        elif mode == "render_runtime":
+            return _add_token_html(mf_card.render_runtime(task, data))
+        elif mode == "refresh":
+            return _add_token_json(mf_card.refresh(task, data))
+
     if timeout_value is None or timeout_value < 0:
-        rendered_info = mf_card.render(task)
+        return _call()
     else:
         with timeout(timeout_value):
-            rendered_info = mf_card.render(task)
-    return rendered_info
+            return _call()
 
 
 @card.command(help="create a HTML card")
@@ -414,29 +452,61 @@ def render_card(mf_card, task, timeout_value=None):
     is_flag=True,
     help="Upon failing to render a card, render a card holding the stack trace",
 )
+@click.option(
+    "--id",
+    default=None,
+    show_default=True,
+    type=str,
+    help="ID of the card",
+)
 @click.option(
     "--component-file",
     default=None,
     show_default=True,
     type=str,
-    help="JSON File with Pre-rendered components.(internal)",
+    help="JSON File with Pre-rendered components. (internal)",
 )
 @click.option(
-    "--id",
+    "--mode",
+    default="render",
+    show_default=True,
+    type=str,
+    help="Rendering mode. (internal)",
+)
+@click.option(
+    "--data-file",
     default=None,
     show_default=True,
     type=str,
-    help="ID of the card",
+    help="JSON file containing data to be updated. (internal)",
+)
+@click.option(
+    "--card-uuid",
+    default=None,
+    show_default=True,
+    type=str,
+    help="Card UUID. (internal)",
+)
+@click.option(
+    "--delete-input-files",
+    default=False,
+    is_flag=True,
+    show_default=True,
+    help="Delete data-file and compontent-file after reading. (internal)",
 )
 @click.pass_context
 def create(
     ctx,
     pathspec,
+    mode=None,
     type=None,
     options=None,
     timeout=None,
     component_file=None,
+    data_file=None,
     render_error_card=False,
+    card_uuid=None,
+    delete_input_files=None,
     id=None,
 ):
     card_id = id
@@ -452,11 +522,26 @@ def create(
 
     graph_dict, _ = ctx.obj.graph.output_steps()
 
+    if card_uuid is None:
+        card_uuid = str(uuid.uuid4()).replace("-", "")
+
     # Components are rendered in a Step and added via `current.card.append` are added here.
     component_arr = []
     if component_file is not None:
         with open(component_file, "r") as f:
             component_arr = json.load(f)
+        # data is passed in as temporary files which can be deleted after use
+        if delete_input_files:
+            os.remove(component_file)
+
+    # Load data to be refreshed for runtime cards
+    data = {}
+    if data_file is not None:
+        with open(data_file, "r") as f:
+            data = json.load(f)
+        # data is passed in as temporary files which can be deleted after use
+        if delete_input_files:
+            os.remove(data_file)
 
     task = Task(full_pathspec)
     from metaflow.plugins import CARDS
@@ -500,18 +585,20 @@ def create(
 
         if mf_card:
             try:
-                rendered_info = render_card(mf_card, task, timeout_value=timeout)
+                rendered_info = update_card(
+                    mf_card, mode, task, data, timeout_value=timeout
+                )
             except:
                 if render_error_card:
                     error_stack_trace = str(UnrenderableCardException(type, options))
                 else:
                     raise UnrenderableCardException(type, options)
         #
 
-    if error_stack_trace is not None:
+    if error_stack_trace is not None and mode != "refresh":
         rendered_info = error_card().render(task, stack_trace=error_stack_trace)
 
-    if rendered_info is None and render_error_card:
+    if rendered_info is None and render_error_card and mode != "refresh":
         rendered_info = error_card().render(
             task, stack_trace="No information rendered From card of type %s" % type
         )
@@ -532,12 +619,20 @@ def create(
         card_id = None
 
     if rendered_info is not None:
-        card_info = card_datastore.save_card(save_type, rendered_info, card_id=card_id)
-        ctx.obj.echo(
-            "Card created with type: %s and hash: %s"
-            % (card_info.type, card_info.hash[:NUM_SHORT_HASH_CHARS]),
-            fg="green",
-        )
+        if mode == "refresh":
+            card_datastore.save_data(
+                card_uuid, save_type, rendered_info, card_id=card_id
+            )
+            ctx.obj.echo("Data updated", fg="green")
+        else:
+            card_info = card_datastore.save_card(
+                card_uuid, save_type, rendered_info, card_id=card_id
+            )
+            ctx.obj.echo(
+                "Card created with type: %s and hash: %s"
+                % (card_info.type, card_info.hash[:NUM_SHORT_HASH_CHARS]),
+                fg="green",
+            )
 
 
 @card.command()
@@ -655,7 +750,6 @@ def list(
     as_json=False,
     file=None,
 ):
-
     card_id = id
     if pathspec is None:
         list_many_cards(
@@ -687,3 +781,89 @@ def list(
         show_list_as_json=as_json,
         file=file,
     )
+
+
+@card.command(help="Run local card viewer server")
+@click.option(
+    "--run-id",
+    default=None,
+    show_default=True,
+    type=str,
+    help="Run ID of the flow",
+)
+@click.option(
+    "--port",
+    default=8324,
+    show_default=True,
+    type=int,
+    help="Port on which Metaflow card server will run",
+)
+@click.option(
+    "--namespace",
+    "user_namespace",
+    default=None,
+    show_default=True,
+    type=str,
+    help="Namespace of the flow",
+)
+@click.option(
+    "--max-cards",
+    default=30,
+    show_default=True,
+    type=int,
+    help="Maximum number of cards to be shown at any time by the server",
+)
+@click.pass_context
+def server(ctx, run_id, port, user_namespace, max_cards):
+    from .card_server import create_card_server, CardServerOptions
+    user_namespace = resolve_identity() if user_namespace is None else user_namespace
+    run = _get_run_object(ctx.obj, run_id, user_namespace)
+    options = CardServerOptions(
+        run_object=run,
+        only_running=False,
+        follow_resumed=False,
+        flow_datastore=ctx.obj.flow_datastore,
+        max_cards=max_cards,
+    )
+    create_card_server(options, port, ctx.obj)
+
+
+def _get_run_object(obj, run_id, user_namespace):
+    from metaflow import Flow, Run, Task
+
+    flow_name = obj.flow.name
+    try:
+        if run_id is not None:
+            namespace(None)
+        else:
+            _msg = "Searching for runs in namespace: %s" % user_namespace
+            obj.echo(_msg, fg="blue", bold=False)
+            namespace(user_namespace)
+        flow = Flow(pathspec=flow_name)
+    except MetaflowNotFound:
+        raise CommandException("No run found for *%s*." % flow_name)
+
+    except MetaflowNamespaceMismatch:
+        raise CommandException(
+            "No run found for *%s* in namespace *%s*. You can switch the namespace using --namespace"
+            % (flow_name, user_namespace)
+        )
+
+    if run_id is None:
+        run_id = flow.latest_run.pathspec
+
+    else:
+        assert len(run_id.split("/")) == 1, "run_id should be of the form <runid>"
+        run_id = "/".join([flow_name, run_id])
+
+    try:
+        run = Run(run_id)
+    except MetaflowNotFound:
+        raise CommandException("No run found for runid: *%s*." % run_id)
+    except MetaflowNamespaceMismatch:
+        raise CommandException(
+            "No run found for runid: *%s* in namespace *%s*. You can switch the namespace using --namespace"
+            % (run_id, user_namespace)
+        )
+    obj.echo("Using run-id %s" % run_id, fg="blue", bold=False)
+    return run

metaflow/plugins/cards/card_client.py

@@ -2,7 +2,7 @@
 from metaflow.datastore import FlowDataStore
 from metaflow.metaflow_config import CARD_SUFFIX
 from .card_resolver import resolve_paths_from_task, resumed_info
-from .card_datastore import CardDatastore
+from .card_datastore import CardDatastore, CardNameSuffix
 from .exception import (
     UnresolvableDatastoreException,
     IncorrectArguementException,
@@ -57,6 +57,15 @@ def __init__(
         # Tempfile to open stuff in browser
         self._temp_file = None
 
+    def _get_data(self) -> Optional[dict]:
+        # currently an internal method to retrieve a card's data.
+        data_paths = self._card_ds.extract_data_paths(
+            card_type=self.type, card_hash=self.hash, card_id=self._card_id
+        )
+        if len(data_paths) == 0:
+            return None
+        return self._card_ds.get_card_data(data_paths[0])
+
     def get(self) -> str:
         """
         Retrieves the HTML contents of the card from the
@@ -172,7 +181,7 @@ def _get_card(self, index):
         if index >= self._high:
             raise IndexError
         path = self._card_paths[index]
-        card_info = self._card_ds.card_info_from_path(path)
+        card_info = self._card_ds.card_info_from_path(path, suffix=CardNameSuffix.CARD)
         # todo : find card creation date and put it in client.
         return Card(
             self._card_ds,
@@ -252,6 +261,7 @@ def get_cards(
         # Exception that the task argument should be of form `Task` or `str`
         raise IncorrectArguementException(_TYPE(task))
 
+    origin_taskpathspec = None
     if follow_resumed:
         origin_taskpathspec = resumed_info(task)
         if origin_taskpathspec: