tiagocoutinho

import socket

def make_stream(addr):
    c = socket.create_connection(("0",5001))
    print("Connected!")
    reply = c
    while True:
        request = yield reply
        if not request.strip():
            break

tiagocoutinho

from functools import partial
from typing import Annotated
from fastapi import responses, FastAPI, Form

FormField = Annotated[str, Form()]

app = FastAPI()
get_html = partial(app.get, response_class=responses.HTMLResponse)
put_html = partial(app.put, response_class=responses.HTMLResponse)

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: