hiway

from zentropi import Agent
from zentropi import on_timer
from zentropi import on_message


class Leonidas(Agent):
    @on_timer(3)
    def ping(self):
        self.message('Spartans?')
    

hiway

# $ micropython -m upip install micropython-inspect

try:  # python 3.5+
    from inspect import iscoroutinefunction
    import asyncio
except ImportError:  # micropython 1.8+
    from inspect import isgeneratorfunction
    import uasyncio as asyncio  
    def iscoroutinefunction(func):
        return isgeneratorfunction(func) or repr(func).startswith('<closure <generator')

hiway

import asyncio
import logging
import os
import threading
from typing import List

import asyncssh
import janus

LOG_LEVEL = logging.DEBUG

hiway

#!/usr/bin/env python
import asyncio

from prompt_toolkit import CommandLineInterface
from prompt_toolkit.shortcuts import create_prompt_application
from prompt_toolkit.shortcuts import create_asyncio_eventloop


def get_cli(event_loop):
    return CommandLineInterface(application=create_prompt_application(),

hiway

import os
import readline

AUTOCOMPLETE_CACHE = {}


def completer(text, state):
    global AUTOCOMPLETE_CACHE
    buffer = os.path.expanduser(readline.get_line_buffer())
    cached_response = AUTOCOMPLETE_CACHE.get(buffer, [])

hiway

from django.core.management.base import BaseCommand, CommandError


class Command(BaseCommand):
    help = ''

    def handle(self, *args, **options):
        try:
            pass
        except Exception:

hiway

import os

from flask import Flask, request

app = Flask(__name__)


def osascript_multiline_compress(script):
    script = ["-e '{line}'".format(line=line) for line in script.split('\n')]
    return 'osascript ' + ' '.join(script)

hiway

import usocket as socket
import ussl as ssl

def http_get(url):
    _, _, host, path = url.split('/', 3)
    addr = socket.getaddrinfo(host, 443)[0][-1]
    s = socket.socket()
    s.connect(addr)
    s = ssl.wrap_socket(s)
    s.write(bytes('GET /%s HTTP/1.0\r\nHost: %s\r\n\r\n' % (path, host), 'utf8'))

hiway

#!/usr/bin/env python
# coding=utf-8

from zentropi import agent


@agent.on_timer(3)
def say_hello():
    agent.emit('hello', data={'name': 'Your Name'})

hiway

import asyncio
import traceback

from channels import Group

from config.asgi import channel_layer


class ChannelsConsumer(object):
    async def long_running_operation(self, duration):