Skip to content

Instantly share code, notes, and snippets.

View jaycosaur's full-sized avatar
🐦

Jacob Richter jaycosaur

🐦
4 followers · 0 following
  • Sydney, Australia
View GitHub Profile
@jaycosaur
jaycosaur / person.ts
Last active August 9, 2020 13:06
Implicit vs Explicit typing [typescript] - Typescript to Python field guide
interface Person {
name: string;
age: number;
height: number;
friends: Person[];
}
// Jane implicitly implements the person type
const jane = {
name: "Jane",
@jaycosaur
jaycosaur / multiplex_queue.py
Last active February 1, 2020 12:11
Multiplex class joins messages from multiple queues together using QueuePipe's. Useful for fan-in jobs.
from queue import Queue
from threading import Thread
from typing import Any, List, Iterator
class PipeClose:
pass
class QueuePipe:
@jaycosaur
jaycosaur / ticker.py
Created February 1, 2020 11:32
Emulation of Ticker class in golang. Emits the time at intervals of 'time_to_wait' seconds on internal queue. Can be stopped by calling 'stop' method.
from queue import Queue, Empty
from threading import Thread
from typing import Any
import time
class Ticker:
_alive = True
_q = Queue()
_stop_q = Queue()
@jaycosaur
jaycosaur / queue_pipe.py
Created February 1, 2020 11:30
QueuePipe joins forwards messages from one theading queue to another, can be stopped by invoking the stop() method.
from queue import Queue
from threading import Thread
class KillSignal:
pass
class QueuePipe:
def __init__(self, queue_in: Queue, queue_out: Queue) -> None:
@jaycosaur
jaycosaur / fanout_queue.py
Last active July 18, 2022 19:26
Async and Sync queue message multicasting to multiple queues. This is the implementation of a message fanout strategy for worker threads and processes. Note this doesn't create worker threads / processes, it only manages (in a blocking way) multicasting messages.
from typing import Type, Set, Any
from multiprocessing import Queue
import asyncio
class MulticastQueue:
def __init__(self, queue_constructor: Type[Queue] = Queue) -> None:
self.subscribers: Set[Queue] = set()
self.constructor = queue_constructor
def register(self) -> Queue:
@jaycosaur
jaycosaur / sync_to_async_queue.py
Created January 27, 2020 21:39
Working with asyncio and sync worker processes is always a pain, not anymore with Janus and this convertor. Async -> Sync is also trivial to implement.
from queue import Queue
from threading import Thread
from typing import cast
import asyncio
import janus
# note that this will never cleanup the thread. You will need to implement your own shutdown logic / methods.
def sync_to_async_queue(queue: Queue, loop: asyncio.AbstractEventLoop) -> asyncio.Queue:
conversion_queue: janus.Queue = janus.Queue(loop=loop)
@jaycosaur
jaycosaur / select.py
Last active February 1, 2020 12:16
Select statement golang emulation in python.
from queue import Queue
from threading import Thread
from typing import Any, List, Iterator
class PipeClose:
pass
class PipeExtra:
@jaycosaur
jaycosaur / vision-multiprocess-shared-memory-example.py
Created October 29, 2019 09:06
import cv2
import time
import numpy as np
import multiprocessing as mp
log_const = 30
shared_frame = mp.RawArray('B', 2764800)
camera_dims = (720, 1280, 3)
@jaycosaur
jaycosaur / vision-multiprocess-example.py
Created October 29, 2019 09:05
import cv2
import time
import numpy as np
import multiprocessing as mp
log_const = 30
class FPS:
@jaycosaur
jaycosaur / vision-threaded-example.py
Created October 29, 2019 09:04
import cv2
import time
import numpy as np
from threading import Thread
from queue import Queue
log_const = 30