tiagocoutinho

import time
import socket

def temp_stream():
  with socket.create_connection(("localhost", 10_123)) as sock:
    sockfd = sock.makefile("rwb", buffering=1)
    while True:
      sockfd.write(b"TEMP?\n")
      payload = sockfd.readline()
      yield float(payload)

tiagocoutinho

import socket
import select

serv = socket.create_server(("", 10_010))

readers = {serv}
while True:
    ready, _, _ = select.select(readers, (), ())
    if serv in ready:
        client, addr = serv.accept()

tiagocoutinho

import random
import string

ascii_alphanumeric = string.ascii_letters + string.digits

def random_name(min_length=32, max_length=32):
    if not (k := random.randint(min_length, max_length)):
        return ""
    first = random.choice(string.ascii_letters)
    return first + "".join(random.choices(ascii_alphanumeric, k=k-1))

tiagocoutinho

"""
$ pip install posix_ipc pytest pytest-xdist

$ pytest -s -v -n 4 test_fixture_run_once_xdist.py
"""

import inspect
import logging
import time

tiagocoutinho

#!/bin/bash

[email protected]:tiagocoutinho
REPOS=( "gepace" "cryocon" "vacuubrand" "xia-pfcu" "mythen-dcs" )

CLONE="git clone --depth=1 --no-single-branch"

function fetch {
        for repo in "${REPOS[@]}"
        do

tiagocoutinho

"""
python usocat.py <listening port> <downstream address>

Example:

python usocat.py 0:10001 192.168.1.5:8000
"""

import logging
import select

tiagocoutinho

import os
import socket
import logging


def cb(sock):
    reader = sock.makefile('rb')
    for line in reader:
        sock.sendall(line[::-1])

tiagocoutinho

opening = "([{"
closing = ")]}"


def balanced(expr):
    stack = []
    for c in expr:
        if c in opening:
            stack.append(c)
        elif c in closing:

tiagocoutinho

import asyncio
import inspect
import functools


def async_(fn):
    """
    Wrap the given function into a coroutine function.
    
    If the function returns a coroutine, that coroutine is awaited and its

tiagocoutinho

def nats(n):
    while True:
        yield n
        n += 1


# another nats implementation
def nats2(n):
    yield n
    yield from nats2(n+1)