Skip to content

Instantly share code, notes, and snippets.

View maxfischer2781's full-sized avatar

Max Kühn maxfischer2781

  • Karlsruhe, Germany
View GitHub Profile
@maxfischer2781
maxfischer2781 / monitored_pipe.py
Last active January 20, 2020 12:40
Variant of usim.Pipe that allows load monitoring
from usim import Pipe, instant
from usim._primitives.notification import Notification
class MonitoredPipe(Pipe):
def __init__(self, throughput: float):
super().__init__(throughput)
self._monitor = Notification()
async def load(self):
@maxfischer2781
maxfischer2781 / align.py
Last active January 5, 2020 17:26
Python string alignment
from itertools import accumulate
a = ['a', '30', '-', 'foot', 'und', 'ete', 'cted']
b = ['a', '30-foot', 'undetected']
def alignment(a: "Iterable[str]", b: "Iterable[str]") -> "List[List[int]]":
if ''.join(a) != ''.join(b):
raise ValueError("a and b must be fragments of the same string")
target_subs = accumulate(b)
@maxfischer2781
maxfischer2781 / stacked_context.py
Created September 19, 2019 15:56
Python context manager that is stack-aware in a thread safe manner
import threading, random
class StackContext:
def __init__(self):
self._state = threading.local()
self._state.stack = []
def __enter__(self):
this_context = random.random()
@maxfischer2781
maxfischer2781 / clone_gridmapdir.py
Last active June 9, 2020 07:45
Copy a gridmapdir to a new location, preserving its mappings
#!/usr/bin/python
"""
Copy a gridmapdir to a new location, preserving its mappings
This script clones a gridmapdir, taking into account relative hardlinks.
This allows copying a gridmapdir to a different device, without losing mappings
from identities to accounts.
.. code:: bash
@maxfischer2781
maxfischer2781 / switch_type.py
Created August 23, 2019 13:10
Extended class based variant of switch statements in Python
r"""
Another class-based implementation of a ``switch`` type
Toy example for building types that emulate ``switch``-statements.
Supports type-based matching, and can be extended to destructuring/pattern
matching and value-based matching.
``switch``-types must derive from the ``Switch`` class. Each match case
is a method ``def case(self, name: pattern [, name: pattern [, ...]]):``.
Cases are seemingly evaluated in-order, without guarantee on side-effects.
@maxfischer2781
maxfischer2781 / IterableQueue
Created August 9, 2019 16:35
An iterable variant of multiprocessing.Queue
from multiprocessing.queues import Queue
from multiprocessing import get_context, Process, cpu_count
import os
class IterableQueue(Queue):
"""
``multiprocessing.Queue`` that can be iterated to ``get`` values
:param sentinel: signal that no more items will be received
@maxfischer2781
maxfischer2781 / private_deco.py
Last active July 30, 2020 08:21
Python private class attributes via decorator
"""
This is a toy example for implementing private fields via decorators
Provides a base type ``Class`` and decorator ``trusted`` to handle
private/public data fields. Classes that derive from ``Class`` can own
private data, and methods marked as ``trusted`` can access it.
Every ``trusted`` method receives a privileged ``self`` that operates on
private data; public data is accessible as ``self.__public__``. Regular
methods and all external clients only operate on the public data.
@maxfischer2781
maxfischer2781 / measure_instance_resources.py
Created June 19, 2019 16:39
Script to measure the memory used by instances for various configurations
from itertools import product, islice
import argparse
import math
import string
import psutil
import keyword
CLI = argparse.ArgumentParser('Test memory consumption of instances')
CLI.add_argument(
@maxfischer2781
maxfischer2781 / random_pw.py
Created February 6, 2019 12:05
Create a random string/password
#!/usr/bin/env python3
from __future__ import print_function
import random
import math
import sys
import argparse
#: alphabet of 2**6 letters
ALPHABET_64 = "0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ+="
@maxfischer2781
maxfischer2781 / mp_gc.py
Created November 27, 2018 16:59
MVCE for early finalisation using Python multiprocessing and spawn/fork method
import os
import sys
import time
import multiprocessing
def print_pid(*args, **kwargs):
print('[%s]' % os.getpid(), *args, **kwargs)