Skip to content

Instantly share code, notes, and snippets.

Avatar

Erik Bernhardsson erikbern

View GitHub Profile
@erikbern
erikbern / asyncio_coroutine_interceptor.py
Last active Jun 2, 2022
Send data to coroutines that do async things
View asyncio_coroutine_interceptor.py
async def intercept_coro(coro, interceptor):
# This roughly corresponds to https://gist.github.com/erikbern/ad7615d22b700e8dbbafd8e4d2f335e1
# The underlying idea is that we can execute a coroutine ourselves and use it to intercept
# any awaitable object. This lets the coroutine await arbitrary awaitable objects, not just
# asyncio futures. See how this is used in object.load.
value_to_send = None
while True:
try:
awaitable = coro.send(value_to_send)
assert inspect.isawaitable(awaitable)
@erikbern
erikbern / loop_hack.py
Last active Apr 3, 2022
Example of how to use async/await programming in Python for non-asyncio purposes
View loop_hack.py
# First, let's create an awaitable object.
# In this case it's a very dumb container of integers.
# Any time a coroutine runs `await Thing(n)` it just resolves to n
# However, we could resolve it to something completely different if we wanted to
class Thing:
def __init__(self, n):
self._n = n
def __await__(self):
View asynchronous_constructor.py
import asyncio
class AsyncConstructorMeta(type):
"""Metaclass to support asynchronous constructors in Python.
Basically we're exploiting the fact that __new__ can return anything in Python.
So we're taking the old __init__ code, removing it from the class, and instead,
we create a custom __new__ method that returns a coroutine wrapping the original
constructor.
@erikbern
erikbern / take_over_globals.py
Last active Nov 5, 2021
Just a proof of concept of how you can inject your own "storage engine" for global variables
View take_over_globals.py
class MyDict(dict):
def __init__(self):
self._dict = {}
def __getitem__(self, k):
print(f'Looking up {k}')
return self._dict[k]
def __setitem__(self, k, v):
print(f'Assigning {k} to {v}')
View exec_hack.py
class MyDict(dict):
def __init__(self):
self._dict = {}
def __getitem__(self, k):
print(f'Looking up {k}')
return self._dict[k]
def __setitem__(self, k, v):
print(f'Assigning {k} to {v}')
View noise_mean_reversion.py
import random
import numpy
from matplotlib import pyplot
rs = numpy.random.randn(1000)
xs = rs[1:-1] - rs[:-2]
ys = rs[2:] - rs[1:-1]
pyplot.scatter(xs, ys)
pyplot.show()
View kx_engineer_productivity.py
import numpy
import scipy.optimize
from matplotlib import pyplot
cs = numpy.linspace(0.01, 0.99, 100)
ks = []
for c in cs:
def f(log_k):
k = numpy.exp(log_k)
View utilization_and_cycle_time.py
import numpy
import random
from matplotlib import pyplot
percentiles = [50, 75, 90, 95, 99]
latencies = [[] for p in percentiles]
loads = []
n = 100000
for k in numpy.linspace(0.01, 1.0, 100):
@erikbern
erikbern / american_community_survey_example.py
Last active May 1, 2022
Download and parse American Community Survey data using Python
View american_community_survey_example.py
# Uses American Community Survey data to estimate property taxes
# https://www.census.gov/programs-surveys/acs/
# The data is a f-ing PITA to parse, but here's an attempt
import bs4
import csv
import io
import os
import requests
import sys
View plot_word_length_in_letters.py
from num2words import num2words
from matplotlib import pyplot
lang = 'de'
words = [num2words(i, lang=lang) for i in range(1000000)]
fig = pyplot.figure()
ax = fig.add_subplot(111)
ax.semilogx([len(word) for word in words], color='green')