Virtual serial line on linux with socat, python server & client
socat -d -d pty,raw,echo=0,link=/tmp/cryocon_simulator pty,raw,echo=0,link=/tmp/cryocon
| <!DOCTYPE html> | |
| <html> | |
| <head> | |
| <meta charset='utf-8'> | |
| <title>WebRTC Demo</title> | |
| <style> | |
| textarea { width: 100%; height: 100px;} | |
| .video-player { border: 2px solid greenyellow; background-color: gray; width: 100%; } | |
| </style> |
| # pip install aiortc aiohttp opencv-python | |
| import argparse | |
| import asyncio | |
| import logging | |
| from aiohttp import web | |
| from aiortc import RTCPeerConnection, RTCSessionDescription | |
| from aiortc.contrib.media import MediaPlayer, MediaRelay |
| 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 |
| 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) |
| """ | |
| tsb.py -- A telnet <-> serial port bridge | |
| Copyright (C) 2005 Eli Fulkerson | |
| This program is free software; you can redistribute it and/or | |
| modify it under the terms of the GNU General Public License | |
| as published by the Free Software Foundation; either version 2 | |
| of the License, or (at your option) any later version. | |
| This program is distributed in the hope that it will be useful, |
| # I dare you to find an execution that prints out 0 | |
| # (at least using the CPython implementation) | |
| from threading import Thread | |
| C, N = 0, 1_000_000 | |
| def f(inc=1): | |
| global C, N | |
| for i in range(N): | |
| C += inc |
| 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) |
| 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() |
| 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)) |