# AUTOGENERATED FILE! PLEASE DON'T EDIT
import k1lib, os, dill, time, inspect, json; k1 = k1lib
import k1lib.cli as cli
from collections import defaultdict
__all__ = ["FromNotebook", "FromPythonFile", "BuildPythonFile", "StartServer", "GenerateHtml", "commonCbs", "serve"]
basePath = os.path.dirname(inspect.getabsfile(k1lib)) + os.sep + "serve" + os.sep
[docs]class FromNotebook(k1.Callback):
[docs] def __init__(self, fileName, tagName="serve"):
"""Grabs source code from a Jupyter notebook. Will grab cells with the comment
like ``# serve`` in the first line.
:param tagName: tag name on the first line of the cell to pull out from"""
super().__init__(); self.fileName = fileName; self.tagName = tagName
[docs] def fetchSource(self): self.l["sourceCode"] = cli.nb.cells(self.fileName) | cli.nb.pretty(whitelist=[self.tagName]) | cli.filt(cli.op()["cell_type"] == "code") | (cli.op()["source"] | ~cli.head(1)).all() | cli.joinStreams() | cli.deref()
[docs]class FromPythonFile(k1.Callback):
[docs] def __init__(self, fileName):
"""Grabs source code from a python file."""
super().__init__(); self.fileName = fileName
[docs] def fetchSource(self): self.l["sourceCode"] = cli.cat(self.fileName) | cli.deref()
[docs]class BuildPythonFile(k1.Callback):
[docs] def __init__(self, port=None):
"""Builds the output Python file, ready to be served on localhost.
:param port: which port to run on localhost. If not given, then a port will
be picked at random, and will be available at ``cbs.l['port']``"""
super().__init__(); self.port = port
[docs] def buildPythonFile(self):
# simple prefix
self.l["pythonFile"] = ["from k1lib.imports import *", *self.l["sourceCode"]] | cli.file()
# grabs random free port if one is not available
if self.port is None: import socket; sock = socket.socket(); sock.bind(('', 0)); self.l["port"] = port = sock.getsockname()[1]; sock.close()
else: port = self.port
# grabs temp meta file for communication
self.l["metaFile"] = metaFile = "" | cli.file(); os.remove(metaFile)
# actually has enough info to build the server
(cli.cat(f"{basePath}suffix.py") | cli.op().replace("SOCKET_PORT", f"{port}").replace("META_FILE", metaFile).all()) >> cli.file(self.l["pythonFile"])
[docs]class StartServer(k1.Callback):
[docs] def __init__(self, initTime=10):
"""Starts the server, verify that it starts okay and dumps meta information (including
function signatures) to ``cbs.l``
:param initTime: time to wait in seconds until the server is online before declaring it's unsuccessful"""
super().__init__(); self.initTime = initTime
[docs] def startServer(self):
None | cli.cmd(f"python {self.l['pythonFile']} &"); count = 0; initTime = self.initTime
while not os.path.exists(self.l["metaFile"]):
if count > initTime/0.1: raise Exception(f"Tried to start server up, but no responses yet. Port: {self.l['port']}, pythonFile: {self.l['pythonFile']}, metaFile: {self.l['metaFile']}")
count += 1; time.sleep(0.1)
self.l["meta"] = meta = self.l["metaFile"] | cli.cat(text=False) | cli.aS(dill.loads)
meta["annoStrs"] = meta["annos"].items() | cli.apply(lambda x: x.__name__, 1) | cli.toDict()
meta["goodTypeStrs"] = meta["goodTypes"] | cli.apply(lambda x: x.__name__) | cli.deref()
[docs]class GenerateHtml(k1.Callback):
[docs] def __init__(self, serverPrefix=None, htmlFile=None, title="Interactive demo"):
"""Generates a html file that communicates with the server.
:param serverPrefix: prefix of server for back and forth requests, like "https://example.com/proj1". If
empty, tries to grab ``cbs.l["serverPrefix"]``, which you can deposit from your own callback. If
that's not available then it will fallback to ``localhost:port``
:param htmlFile: path of the target html file. If not specified then a temporary file
will be created and made available in ``cbs.l["htmlFile"]``
:param title: title of html page"""
super().__init__(); self.serverPrefix = serverPrefix; self.htmlFile = htmlFile; self.title = title
[docs] def generateHtml(self):
meta = dict(self.l["meta"]); del meta["annos"]; del meta["goodTypes"]; meta = meta | cli.aS(json.dumps)
replaces = cli.op().replace("META_JSON", meta)\
.replace("SERVER_PREFIX", self.serverPrefix or self.l["serverPrefix"] or f"http://localhost:{self.l['port']}")\
.replace("TITLE", self.title)
self.l["htmlFile"] = cli.cat(f"{basePath}main.html") | replaces.all() | cli.file(self.htmlFile)
[docs]def commonCbs():
"""Grabs common callbacks, including :class:`BuildPythonFile` and :class:`StartServer`"""
return k1.Callbacks().add(BuildPythonFile()).add(StartServer());
[docs]def serve(cbs):
"""Runs the serving pipeline."""
import flask, flask_cors
cbs.l = defaultdict(lambda: None)
cbs("begin")
cbs("fetchSource") # fetches cells
cbs("beforeBuildPythonFile"); cbs("buildPythonFile") # builds python server file
cbs("beforeStartServer"); cbs("startServer") # starts serving the model on localhost and add more meta info
cbs("beforeGenerateHtml"); cbs("generateHtml") # produces a standalone html file that provides interactive functionalities
cbs("end")
return cbs