Нотатка. Python. functools.partial.

🔨 functools.partial(func, /, *args, **keywords)

functools.partial

📍 Повертає новий частковий об’єкт, який під час виклику поводитиметься як func, що викликається з позиційними аргументами args і ключовими аргументами keywords. Якщо до виклику надається більше аргументів, вони додаються до args. Якщо надаються додаткові ключові аргументи, вони розширюють і замінюють ключові слова.

📌 Примітка. Більшість функцій планування asyncio не дозволяють передавати ключові аргументи. Для цього скористайтеся functools.partial():

# will schedule "print("Hello", flush=True)"
loop.call_soon(
   functools.partial(print, "Hello", flush=True)
   )

🎧 Використання часткових об’єктів зазвичай зручніше, ніж використання лямбда-виразів, оскільки asyncio може краще відтворювати часткові об’єкти в повідомленнях про налагодження та помилки.

# лямбда-вираз
loop.call_soon(
   lambda : print("Hello")
   )

Як завершити async-завдання, створене з функції що блокує через asyncio.to_thread(): експеримент та рішення.

Маленький експеримент як завершити async task якщо вона була створена з функції що блокує.

Тобто створена з не асинхронної функції (proc2) за допомогою asyncio.to_thread().

	import asyncio
	import threading
	import time
	# Stop event to signal the thread to terminate
	stop_event = threading.Event()
	async def proc1():
	"""
	Infinite loop async
	"""
	i = 0
	while True:
	print(f"proc1 {i}")
	i += 1
	await asyncio.sleep(1)
	def proc2():
	"""
	This function is run in a separate thread
	is blocked until stop_event is set or max 10 try
	"""
	i = 0
	while not stop_event.is_set():
	print(f"proc2 {i}")
	i += 1
	time.sleep(1)
	if i>10:
	print("proc2 end legacy after 10 try")
	break
	async def main():
	print("start main")
	t1=asyncio.create_task(proc1())
	p2=asyncio.to_thread(proc2)
	t2=asyncio.create_task(p2)
	print("sleep 5")
	await asyncio.sleep(5)
	print("cancel t1")
	t1.cancel()
	try:
	await t1
	except asyncio.CancelledError:
	print(f"Task t1 was cancelled.")
	print("cancel t2")
	t2.cancel()
	try:
	await t2
	except asyncio.CancelledError:
	print(f"Task t2 was cancelled.")
	# Signal the thread to stop
	stop_event.set()
	print("end main")
	print("start")
	asyncio.run(main())
	print("done")
	"""
	# Without use stop_event (L54)
	start
	start main
	sleep 5
	proc1 0
	proc2 0
	proc1 1
	proc2 1
	proc2 2
	proc1 2
	proc2 3
	proc1 3
	proc2 4
	proc1 4
	proc2 5
	cancel t1
	Task t1 was cancelled.
	cancel t2
	Task t2 was cancelled.
	end main
	proc2 6
	proc2 7
	proc2 8
	proc2 9
	proc2 10
	proc2 end legacy after 10 try
	done
	"""
	"""
	# With use stop_event (L54)
	start
	start main
	sleep 5
	proc1 0
	proc2 0
	proc1 1
	proc2 1
	proc2 2
	proc1 2
	proc2 3
	proc1 3
	proc2 4
	proc1 4
	cancel t1
	Task t1 was cancelled.
	cancel t2
	Task t2 was cancelled.
	end main
	done
	"""

view raw async_to_thread_cancelling.py hosted with ❤ by GitHub

Як видно з результатів без використання threading.Event(), програма буде постійно працювати навіть після закінчення всіх асинхронних функцій в async loop, тому що окремий потік породжений у цій програмі ще не завершився й async loop буде чекати на нього.

Моніторинг температури та сповіщення про критичні значення. Оновлення проєкту. Python.

Metrics of temperature and alerts.

 Вступ

До складних часів, коли було ще де і ради чого жити, я поступово впроваджував елементи розумного будинку. 

Одним із таких кроків стала інтеграція Bluetooth BLE термометрів Xiaomi Mijia (LYWSD03MMC), прошитих альтернативною прошивкою для зручності взаємодії через MQTT брокер.

Перша версія проєкту

До часів навчання на курсах Python Developer я створив невеликий консольний проєкт (v.0.1.0). Хоча спершу я не повністю розумів внутрішню логіку, проєкт дозволяв збирати інформацію про температуру, використовуючи ноутбук як локальний сервер.

Мотивація для оновлення

З набуттям нових знань та навичок під час навчання я почав сприймати код по-новому. А тут ще й зима настала: овочі на балконі потрібно було вчасно заносити до кімнати, щоб уникнути їх промерзання. Заморозки наближалися, тому я вирішив вдосконалити проєкт, додавши функцію сповіщень про критичні температури.