jaycosaur’s gists

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 / 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 / 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 / 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 / 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 / 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 / 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 / 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 / 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 / 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

Jacob Richter jaycosaur