vxgmichel

import time
import inspect
import traceback

import trio
import structlog

logger = structlog.get_logger()

vxgmichel

"""Helpers for continued fractions"""

import math
import itertools


def continued_fraction(n, d):
    while d:
        q, r = divmod(n, d)
        n, d = d, r

vxgmichel

#!/usr/bin/env python
"""
Serving ptpython using telnet

I ran into several issues when writing this piece of code, both with ptpython
and prompt_toolkit. Find below a short description for each of them.

Ptpython issues:
----------------

vxgmichel

import time
import asyncio


def task_timing_patch():

    class TaskTimingHandle(asyncio.events.Handle):

        def _add_runtime(self, value):
            try:

vxgmichel

# See the coresponding stackoverflow post:
# https://stackoverflow.com/a/34827291/2846140

import time
import asyncio
import selectors
import contextlib


class TimedSelector(selectors.DefaultSelector):

vxgmichel

import os
import time
import pathlib
from concurrent.futures import ThreadPoolExecutor

from fuse import FUSE, Operations


with ThreadPoolExecutor() as executor:
    mountpoint1 = "./a"

vxgmichel

def generate_tape(tape, state, offset=1):
    head, *tail = tape
    *_, last = tape
    return (
        f"... {head}   " +
        f"   {head}   " * (offset - 1) +
        f"  [{head}|{state}]" +
        "".join(f"   {symbol}   " for symbol in tail) +
        f"   {last}   " * (offset - 1) +

vxgmichel

"""A pytest fixture for running an ssh mock server.

Requires pytest and asyncssh:

    $ pip install pytest asyncssh 
"""

from socket import AF_INET
from unittest.mock import Mock
from contextlib import asynccontextmanager

vxgmichel

import os
import pytest
import subprocess


def get_open_fds():
    cmd = f"lsof -p {os.getpid()}"
    result = subprocess.run(cmd.split(), capture_output=True)
    return {line.split()[-1] for line in result.stdout.splitlines()}

vxgmichel

import attr
import trio


@attr.s
class TaskStatus:

    # Internal state

    _cancel_scope = attr.ib(default=None)