# AUTOGENERATED FILE! PLEASE DON'T EDIT HERE. EDIT THE SOURCE NOTEBOOKS INSTEAD
import k1lib, os, dill, time, inspect, json, base64; k1 = k1lib
from typing import List
import k1lib.cli as cli; from k1lib.cli import *
from collections import defaultdict
try: import PIL.Image, PIL; hasPIL = True
except: PIL = k1.dep("PIL"); hasPIL = False
__all__ = ["FromNotebook", "FromPythonFile", "BuildPythonFile", "BuildDashFile", "StartServer", "GenerateHtml", "commonCbs", "serve",
"text", "slider", "html", "analyze", "webToPy", "pyToWeb"]
basePath = os.path.dirname(inspect.getabsfile(k1lib)) + os.sep + "serve" + os.sep
[docs]class FromNotebook(k1.Callback): # FromNotebook
[docs] def __init__(self, fileName, tagName="serve"): # FromNotebook
"""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""" # FromNotebook
super().__init__(); self.fileName = fileName; self.tagName = tagName # FromNotebook
[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() # FromNotebook
[docs]class FromPythonFile(k1.Callback): # FromPythonFile
[docs] def __init__(self, fileName): # FromPythonFile
"""Grabs source code from a python file.""" # FromPythonFile
super().__init__(); self.fileName = fileName # FromPythonFile
[docs] def fetchSource(self): self.l["sourceCode"] = cli.cat(self.fileName) | cli.deref() # FromPythonFile
[docs]class BuildPythonFile(k1.Callback): # BuildPythonFile
[docs] def __init__(self, port=None): # BuildPythonFile
"""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']``""" # BuildPythonFile
super().__init__(); self.port = port; self.suffix = "suffix.py" # BuildPythonFile
[docs] def buildPythonFile(self): # BuildPythonFile
# simple prefix # BuildPythonFile
self.l["pythonFile"] = ["from k1lib.imports import *", *self.l["sourceCode"]] | cli.file() # BuildPythonFile
# grabs random free port if one is not available # BuildPythonFile
if self.port is None: import socket; sock = socket.socket(); sock.bind(('', 0)); self.l["port"] = port = sock.getsockname()[1]; sock.close() # BuildPythonFile
else: port = self.port # BuildPythonFile
# grabs temp meta file for communication # BuildPythonFile
self.l["metaFile"] = metaFile = "" | cli.file(); os.remove(metaFile) # BuildPythonFile
# actually has enough info to build the server # BuildPythonFile
(cli.cat(f"{basePath}{self.suffix}") | cli.op().replace("SOCKET_PORT", f"{port}").replace("META_FILE", metaFile).all()) >> cli.file(self.l["pythonFile"]) # BuildPythonFile
[docs]class BuildDashFile(BuildPythonFile): # BuildDashFile
[docs] def __init__(self): # BuildDashFile
"""Builds the output Python file for a Dash app, ready to be served on localhost""" # BuildDashFile
super().__init__() # BuildDashFile
self.suffix = "suffix-dash.py" # BuildDashFile
[docs]class StartServer(k1.Callback): # StartServer
[docs] def __init__(self, initTime=10): # StartServer
"""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""" # StartServer
super().__init__(); self.initTime = initTime # StartServer
[docs] def startServer(self): # StartServer
None | cli.cmd(f"python {self.l['pythonFile']} &"); count = 0; initTime = self.initTime # StartServer
while not os.path.exists(self.l["metaFile"]): # StartServer
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']}") # StartServer
count += 1; time.sleep(0.1) # StartServer
self.l["meta"] = meta = self.l["metaFile"] | cli.cat(text=False) | cli.aS(dill.loads) # StartServer
[docs]class GenerateHtml(k1.Callback): # GenerateHtml
[docs] def __init__(self, serverPrefix=None, htmlFile=None, title="Interactive demo"): # GenerateHtml
"""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""" # GenerateHtml
super().__init__(); self.serverPrefix = serverPrefix; self.htmlFile = htmlFile; self.title = title # GenerateHtml
[docs] def generateHtml(self): # GenerateHtml
meta = dict(self.l["meta"]); meta = meta | cli.aS(json.dumps) # GenerateHtml
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) # GenerateHtml
self.l["htmlFile"] = cli.cat(f"{basePath}main.html") | replaces.all() | cli.file(self.htmlFile) # GenerateHtml
[docs]def commonCbs(): # commonCbs
"""Grabs common callbacks, including :class:`BuildPythonFile` and :class:`StartServer`""" # commonCbs
return k1.Callbacks().add(BuildPythonFile()).add(StartServer()); # commonCbs
[docs]def serve(cbs): # serve
"""Runs the serving pipeline.""" # serve
import flask, flask_cors # serve
cbs.l = defaultdict(lambda: None) # serve
cbs("begin") # serve
cbs("fetchSource") # fetches cells # serve
cbs("beforeBuildPythonFile"); cbs("buildPythonFile") # builds python server file # serve
cbs("beforeStartServer"); cbs("startServer") # starts serving the model on localhost and add more meta info # serve
cbs("beforeGenerateHtml"); cbs("generateHtml") # produces a standalone html file that provides interactive functionalities # serve
cbs("end") # serve
return cbs # serve
class baseType: # baseType
def __init__(self): # baseType
"""Base type for all widget types""" # baseType
pass # baseType
def getConfig(self): return NotImplemented # baseType
[docs]class text(baseType): # text
[docs] def __init__(self, multiline:bool=True, password:bool=False): # text
"""Represents text, either on single or multiple lines.
If `password` is true, then will set multiline to false automatically,
and creates a text box that blurs out the contents""" # text
super().__init__(); self.multiline = multiline if not password else False; self.password = password # text
def __repr__(self): return f"<text multiline={self.multiline}>" # text
[docs]class html(baseType): # html
def __init__(self): super().__init__() # html
def __repr__(self): return f"<html>" # html
[docs]class slider(baseType): # slider
[docs] def __init__(self, start:float, stop:float, intervals:int=100): # slider
"""Represents a slider from `start` to `stop` with a bunch of
intervals in the middle. If `defValue` is not specified, uses the
middle point between start and stop""" # slider
super().__init__(); self.start = start; self.stop = stop; self.intervals = intervals; self.dt = (stop-start)/intervals # slider
def __repr__(self): return f"<slider {self.start}---{self.intervals}-->{self.stop}>" # slider
def refine(param:str, anno:baseType, default): # anno is not necessarily baseType, can be other types like "int" # refine
if anno == int: return [param, "int", [default, False]] # refine
if anno == float: return [param, "float", [default, False]] # refine
multiline = lambda s: len(s.split("\n")) > 1 or len(s) > 100 # refine
if anno == bool: return [param, "checkbox", default] # refine
if anno == str: return [param, "text", [default, multiline(default or "")]] # refine
if isinstance(anno, text): return [param, "text", [default, anno.multiline, anno.password]] # refine
if isinstance(anno, slider): return [param, "slider", [default, anno.start, anno.stop, anno.dt]] # refine
if isinstance(anno, range): return [param, "slider", [default, anno.start, anno.stop, anno.step]] # refine
byte2Str = aS(base64.b64encode) | op().decode("ascii") # refine
if hasPIL and anno == PIL.Image.Image: return [param, "image", (default | toBytes() | byte2Str) if default is not None else None] # refine
if anno == bytes: return [param, "bytes", (default | byte2Str) if default is not None else None] # refine
if isinstance(anno, list): anno | apply(str) | deref(); return [param, "dropdown", [anno.index(default), anno]] # refine
if isinstance(anno, html): return [param, "html", [default]] # refine
raise Exception(f"Unknown type {anno}") # refine
[docs]def analyze(f): # analyze
spec = inspect.getfullargspec(f); args = spec.args; n = len(args) # analyze
annos = spec.annotations; defaults = spec.defaults or () # analyze
docs = (f.__doc__ or "").split("\n") | grep(":param", sep=True).till() | filt(op().ab_len() > 0) | op().strip().all(2) | (join(" ") | op().split(":") | ~aS(lambda x, y, *z: [y.split(" ")[1], ":".join(z).strip()])).all() | toDict() # analyze
mainDoc = (f.__doc__ or " ").split("\n") | grep(".", sep=True).till(":param") | breakIf(op()[0].startswith(":param")) | join(" ").all() | join(" ") # analyze
# analyze
if len(annos) != n + 1: raise Exception(f"Please annotate all of your arguments ({n} args + 1 return != {len(annos)} annos). Args: {args}, annos: {annos}") # analyze
if len(defaults) != n: raise Exception(f"Please specify default values for all of your arguments ({n} args != {len(defaults)} default values)") # analyze
a = [args, args | lookup(annos), defaults] | transpose() | ~apply(refine) | deref() # analyze
ret = refine("return", annos["return"], None)[1]; defaults = a | cut(0, 2) | toDict() # analyze
annos = a | cut(0, 1) | toDict(); annos["return"] = ret # analyze
if ret == "slider": raise Exception(f"Return value is a slider, which doesn't really make sense. Return float, str or sth like that") # analyze
# analyze
# args:list, annos:dict, defaults:list, docs:dict # analyze
# analyze
return {"args": args, "annos": annos, "defaults": defaults, "docs": docs, # analyze
"mainDoc": mainDoc, "source": inspect.getsource(f), "pid": os.getpid()} # analyze
return args, annos, defaults, docs, mainDoc, d # analyze
[docs]def webToPy(o:str, klass:baseType): # webToPy
o = str(o) # webToPy
if klass == "int": return int(float(o)) # webToPy
if klass == "float": return float(o) # webToPy
if klass == "slider": o = float(o); return int(o) if round(o) == o else o # webToPy
if klass == "text" or klass == "dropdown": return o # webToPy
if klass == "checkbox": return o.lower() == "true" # webToPy
if klass == "bytes": return o | aS(base64.b64decode) # webToPy
if klass == "image": return o | aS(base64.b64decode) | toImg() # webToPy
return NotImplemented # webToPy
[docs]def pyToWeb(o, klass:baseType) -> str: # pyToWeb
if klass in ("int", "float", "text", "checkbox"): return f"{o}" # pyToWeb
if klass == "slider": return NotImplemented # pyToWeb
if klass == "bytes": return o | aS(base64.b64encode) | op().decode() # pyToWeb
if klass == "image": return o | toBytes() | aS(base64.b64encode) # pyToWeb
if klass == "dropdown": return o; # pyToWeb
if klass == "html": return o.encode() | aS(base64.b64encode) | op().decode() # pyToWeb
return NotImplemented # pyToWeb