Instantly share code, notes, and snippets.

View aiowatch.py
import asyncio
from pathlib import Path
from typing import Optional
from watchdog.events import FileSystemEvent, FileSystemEventHandler
from watchdog.observers import Observer
class _EventHandler(FileSystemEventHandler):
def __init__(self, queue: asyncio.Queue, loop: asyncio.BaseEventLoop,
View h5recarray.py
import codecs
import json
from typing import Union
import h5py
import numpy as np
import pandas as pd
vlen = np.vectorize(len)
vencode = np.vectorize(codecs.encode)
View metaclass.py
class Manager(object):
classes = set()
class Meta(type):
def __new__(cls, name, bases, d):
print("name:", name, "\nbases:", bases, "\ndict: ", d)
if name is not "Base":
Manager.classes.add(d['data'])
return type.__new__(cls, name, bases, d)
View test.json
{
"subject": {
"pairs": {
"A1-A2": {
"atlases": {
"avg.corrected": {
"region": null,
"x": 49.151580810546875,
"y": -33.148521423339844,
"z": 27.572303771972656
View ipc_compare.py
from multiprocessing import Event, Process, Queue
import time
import zmq
class BaseActor(Process):
ready = Event()
def handle(self, msg):
print("dt =", time.time() - msg['timestamp'])
View schedule_coroutines.py
import asyncio
from threading import Event, Thread
class EventLoopThread(Thread):
def __init__(self):
super().__init__()
self.loop = None
self.ready = Event()
View multiline_lambdas.py
import sys
from PyQt5.QtWidgets import *
from PyQt5.QtGui import *
class MainWindow(QWidget):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.setWindowTitle('Multiline lambdas')
View keybase.md

Keybase proof

I hereby claim:

  • I am mivade on github.
  • I am mivade (https://keybase.io/mivade) on keybase.
  • I have a public key ASCjifbgjSNGV_4F5Ge6JtJZBHGMNpi4miYvMDbZUbD1Fwo

To claim this, I am signing this object:

View benchmark.ipynb
Sorry, something went wrong. Reload?
Sorry, we cannot display this file.
Sorry, this file is invalid so it cannot be displayed.
View randport.py
"""Examples of how to bind a socket to a random port and have that port number
accessible to other processes. Useful for interprocess communications.
"""
import time
import socket
from socket import AF_INET, SOCK_STREAM
from multiprocessing import Process, Queue, Event
from contextlib import contextmanager