# AUTOGENERATED FILE! PLEASE DON'T EDIT HERE. EDIT THE SOURCE NOTEBOOKS INSTEAD
import k1lib, json, base64, threading; import k1lib.cli as cli
from collections import deque
__all__ = ["log"]
_logObj = {"loaded": False, "logMsgs": deque(), "path": None}
def _thTarget(): # _thTarget
import asyncio, base64, json; from k1lib import kws # _thTarget
async def main(): # _thTarget
async with kws.WsClient("wss://ws.logs.mlexps.com/_k1_ingest") as ws: # _thTarget
while True: # _thTarget
if len(_logObj["logMsgs"]) == 0: await asyncio.sleep(0.01) # _thTarget
else: await ws.send(_logObj["logMsgs"].popleft()) # _thTarget
asyncio.new_event_loop().run_until_complete(main()) # _thTarget
def log(path:str, obj:"any"): # log
"""Logs random debug statements to logs.mlexps.com server.
Example::
k1.log("ggdrive/topic1", "some message")
k1.log("ggdrive/topic1/sub2", {"some": "json", "object": 2})
# I typically do it like this, so that I can filter down only the messages that I want based on severity
k1.log("ggdrive/info", {"some": "json", "object": 2})
k1.log("ggdrive/error", {"some": "json", "object": 2})
Visit the website https://logs.mlexps.com/watch/ggdrive, or
/watch/ggdrive/topic1, or /watch/ggdrive/topic1/sub2 to view all logs
coming in.""" # log
if not _logObj["loaded"]: _logObj["loaded"] = True; threading.Thread(target=_thTarget).start() # log
if not isinstance(obj, (str, float, int)): # log
obj = base64.b64encode(json.dumps(obj).encode()).decode() # log
_logObj["logMsgs"].append(f"{path}/{obj}") # log
if k1lib.settings.startup.import_optionals: # log
try: # log
from scipy import stats # log
__all__.append("pValue") # log
[docs] def pValue(zScore): # log
"""2-sided p value of a particular z score. Requires :mod:`scipy`.""" # log
return stats.norm.sf(abs(zScore))*2 # log
except: pass # log