espx/klib.py

def woke_up():
    with uses("machine") as machine:
        return machine.reset_cause() == machine.DEEPSLEEP_RESET


def poweroff(wake_in=0):
    with uses("machine") as machine:
        if wake_in:
            rtc = machine.RTC()
            rtc.irq(trigger=rtc.ALARM0, wake=machine.DEEPSLEEP)
            rtc.alarm(rtc.ALARM0, int(wake_in * 1000))
        machine.deepsleep()


def turbo():

    class Turbo:
        impl = __import__("machine")

        def __enter__(self):
            # memleak ?
            self.last = self.impl.freq()
            self.impl.freq(160000000)

        def __exit__(*self):
            # memleak ?
            # __import__('machine').freq(80000000)
            self.impl.freq(self.last)
            del self.last

    return Turbo()


async def sha256(f, block=512):
    with uses("uhashlib", "ubinascii"):
        h = uhashlib.sha256()
        if isinstance(f, str):
            try:
                f = open(f, "rb")
            except OSError:
                f = None
        while f:
            if use.aio:
                await uses.aio.idle()

            data = f.read(block)
            if data:
                h.update(data)
            else:
                break
        if f:
            f.close()
        return ubinascii.hexlify(h.digest()).decode()  # .lower()

async def urlopen(url, data=None, method="GET"):
    if data is not None and method == "GET":
        method = "POST"

    proto, dummy, host = url.split("/", 2)
    if host.find("/")>0:
        host,path = host.split("/",1)
    else:
        path=""

    ssl = proto[4] == "s"

    if ":" in host:
        host, port = host.split(":", 1)
        port = int(port)
    elif ssl:
        port = 443
    else:
        port = 80

    with uses("usocket") as usocket:
        addr = usocket.getaddrinfo(host, port)[0][4]
        s = usocket.socket()
        s.connect(addr)
        if ssl:
            s = __import__("ussl").wrap_socket(s)
        sw = s.write
        sw(method)
        sw(b" ")
        sw(b"/")
        sw(path)
        sw(b" HTTP/1.0\r\nHost: ")
        sw(host)
        sw(b"\r\n")

        if data:
            sw(b"Content-Length: ")
            sw(str(len(data)))
            sw(b"\r\n")
        sw(b"\r\n")
        if data:
            sw(data)

        l = s.readline()
        protover, status, msg = l.split(None, 2)
        status = int(status)

        if status == 404:
            raise OSError(2, "404 [%s]" % url)
        elif status != 200:
            print( status , method, url )

        while True:
            l = s.readline().lower()
            if not l or l == b"\r\n":
                break
            if l.startswith(b"transfer-encoding:"):
                if b"chunked" in l:
                    raise ValueError("Unsupported " + l)
            elif l.startswith(b"location:"):
                raise NotImplementedError("Redirect N/I")
            if use.aio:
                await use.aio.idle()
        return s

def reboot():
    __import__('machine').reset()

def gpio(*argv,**kw):
    import sys
    with uses('m.%s'% sys.platform) as mod:
        try:
            return mod.gpio(*argv,**kw)
        finally:
            del mod

#this one could gc, when all opened files are closed, may need rewrite as a module
def fopen(*argv, **kw):
    import m.fopen
    return m.fopen.fopen(*argv, **kw)

def Pin(pin,mode='r',value=0, **kw):
    import m.pin
    thepin=gpio(pin,mode)
    return m.pin.Pin(pin, thepin, mode,value, **kw)


def HBridge(*argv, **kw):
    import m.hbridge
    return m.hbridge.HBridge(*argv, **kw)

def init_adc(*argv, **kw):
    try:
        return ADC
    except:
        print("loading ADC(MCP3008)")
        import m.MCP3008
    return ADC



def ping(*argv, **kw):
    with uses('m.uping'):
        m.uping.ping(*argv,**kw)