🧪 Автоматизация тестирования с помощью PyTest
📌 PyTest упрощает процесс автоматизации тестирования, позволяя создавать простые и понятные тесты для функций Python. В данном примере показано, как использовать PyTest для параметризации тестов, что помогает повысить надежность и повторяемость в крупных проектах.
Подпишись 👉🏻 @KodduuPython 🤖
import pytest
# Пример функции сложения
def add(a: int, b: int) -> int:
return a + b
# Параметризация тестов для проверки функции сложения
@pytest.mark.parametrize("a, b, expected", [
(1, 1, 2),
(2, 3, 5),
(10, -10, 0),
(-1, -1, -2)
])
def test_add_parametrized(a, b, expected):
assert add(a, b) == expected
# Запуск тестов
# В командной строке выполните: pytest имя_файла.py
📌 PyTest упрощает процесс автоматизации тестирования, позволяя создавать простые и понятные тесты для функций Python. В данном примере показано, как использовать PyTest для параметризации тестов, что помогает повысить надежность и повторяемость в крупных проектах.
Подпишись 👉🏻 @KodduuPython 🤖
⚡2
🎮 Реализация мультиплеера с WebSockets
📌 Этот код демонстрирует базовую настройку сервера WebSockets для обработки подключений клиентов в реальном времени. Он позволяет передавать данные между подключенными игроками и синхронизировать состояние игры. Полезен для создания простых мультиплеерных игр, где необходимо реал-тайм взаимодействие.
Подпишись 👉🏻 @KodduuPython 🤖
import asyncio
import websockets
import json
connected_clients = set()
async def handle_connection(websocket, path):
# Добавляем нового клиента в набор подключенных
connected_clients.add(websocket)
try:
async for message in websocket:
data = json.loads(message)
action = data.get("action")
if action == "broadcast":
await broadcast(data)
elif action == "sync":
await sync_game_state(websocket, data)
except websockets.exceptions.ConnectionClosed:
print(f"Клиент отключился: {websocket}")
finally:
connected_clients.remove(websocket)
async def broadcast(data):
message = json.dumps(data)
for client in connected_clients:
await client.send(message)
async def sync_game_state(websocket, data):
# Здесь можно обработать синхронизацию состояния игры
await websocket.send(json.dumps({"status": "synced"}))
async def main():
async with websockets.serve(handle_connection, "localhost", 8765):
await asyncio.Future() # Ожидание завершения
# Пример использования
asyncio.run(main())
📌 Этот код демонстрирует базовую настройку сервера WebSockets для обработки подключений клиентов в реальном времени. Он позволяет передавать данные между подключенными игроками и синхронизировать состояние игры. Полезен для создания простых мультиплеерных игр, где необходимо реал-тайм взаимодействие.
Подпишись 👉🏻 @KodduuPython 🤖
❤2
Ускорение работы с данными с помощью numpy
📌 Этот код демонстрирует, насколько быстрее numpy может обрабатывать большие массивы по сравнению с чистым Python. Используя numpy, можно значительно ускорить операции, такие как суммирование, благодаря оптимизированным C-расширениям. Это особенно полезно в научных и финансовых приложениях, где важна работа с большими объемами данных и высокая производительность.
Подпишись 👉🏻 @KodduuPython 🤖
import numpy as np
import timeit
# Создание больших массивов чисел
array_size = 10**6
python_list = list(range(array_size))
numpy_array = np.arange(array_size)
# Функция для суммирования элементов с использованием чистого Python
def sum_python(lst):
return sum(lst)
# Функция для суммирования элементов с использованием numpy
def sum_numpy(arr):
return np.sum(arr)
# Измерение времени выполнения для обеих функций
python_duration = timeit.timeit(lambda: sum_python(python_list), number=1)
numpy_duration = timeit.timeit(lambda: sum_numpy(numpy_array), number=1)
print(f"Python: {python_duration:.5f} seconds")
print(f"Numpy: {numpy_duration:.5f} seconds")
📌 Этот код демонстрирует, насколько быстрее numpy может обрабатывать большие массивы по сравнению с чистым Python. Используя numpy, можно значительно ускорить операции, такие как суммирование, благодаря оптимизированным C-расширениям. Это особенно полезно в научных и финансовых приложениях, где важна работа с большими объемами данных и высокая производительность.
Подпишись 👉🏻 @KodduuPython 🤖
⚡1
🔮 Устойчивая обработка ошибок с монадами Maybe
📌 Монада Maybe помогает избежать ошибок, связанных с использованием
Подпишись 👉🏻 @KodduuPython 🤖
from typing import Generic, TypeVar, Union, Callable
T = TypeVar('T')
class Maybe(Generic[T]):
def __init__(self, value: Union[T, None]):
self.value = value
def is_nothing(self) -> bool:
return self.value is None
def map(self, func: Callable[[T], T]) -> 'Maybe[T]':
if self.is_nothing():
return self
try:
return Maybe(func(self.value))
except Exception:
return Maybe(None)
def get_or_else(self, default: T) -> T:
return self.value if not self.is_nothing() else default
# Пример использования
def safe_divide(x: float, y: float) -> Maybe[float]:
if y == 0:
return Maybe(None)
return Maybe(x / y)
result = safe_divide(10, 2).map(lambda x: x * 100).get_or_else(0)
print(result) # Вывод: 500.0
result = safe_divide(10, 0).map(lambda x: x * 100).get_or_else(0)
print(result) # Вывод: 0
📌 Монада Maybe помогает избежать ошибок, связанных с использованием
None или делением на ноль. Вместо выбрасывания исключений, функция возвращает объект Maybe, который можно безопасно обрабатывать с помощью методов map и get_or_else. Это делает код более предсказуемым и устойчивым к ошибкам, особенно полезно в API-интерфейсах и при работе с неполными данными.Подпишись 👉🏻 @KodduuPython 🤖
⚡3
💡 Управление светодиодами с MicroPython
📌 Этот код демонстрирует, как использовать MicroPython для управления светодиодом на микроконтроллере. Функция
MicroPython предоставляет более высокий уровень абстракции над аппаратными компонентами по сравнению с C/C++, что упрощает процесс разработки и прототипирования. Данный пример может быть полезен для создания простых устройств, таких как индикаторы состояния или световые эффекты в проектах с микроконтроллерами.
Подпишись 👉🏻 @KodduuPython 🤖
from machine import Pin
from time import sleep
def blink_led(pin_number: int, delay: float = 0.5):
try:
led = Pin(pin_number, Pin.OUT)
while True:
led.on()
sleep(delay)
led.off()
sleep(delay)
except KeyboardInterrupt:
print("Программа остановлена пользователем")
except Exception as e:
print(f"Произошла ошибка: {e}")
# Пример использования
if __name__ == "__main__":
blink_led(pin_number=2)
📌 Этот код демонстрирует, как использовать MicroPython для управления светодиодом на микроконтроллере. Функция
blink_led включает и выключает светодиод с заданной задержкой, которую можно настроить. Это базовый пример, который показывает простоту использования Python для быстрого прототипирования встраиваемых систем.MicroPython предоставляет более высокий уровень абстракции над аппаратными компонентами по сравнению с C/C++, что упрощает процесс разработки и прототипирования. Данный пример может быть полезен для создания простых устройств, таких как индикаторы состояния или световые эффекты в проектах с микроконтроллерами.
Подпишись 👉🏻 @KodduuPython 🤖
Создаем AI для игровых персонажей на Python 🎮
📌 В этом коде реализован простой AI для NPC в играх. Он может патрулировать, преследовать игрока и уклоняться от угроз. Такой подход позволяет сделать игровой процесс более интерактивным и интересным.
Подпишись 👉🏻 @KodduuPython 🤖
import random
class NPC:
def __init__(self, name: str):
self.name = name
self.position = (0, 0)
def patrol(self):
# Перемещение в случайном направлении
self.position = (self.position[0] + random.choice([-1, 1]),
self.position[1] + random.choice([-1, 1]))
print(f"{self.name} патрулирует: {self.position}")
def chase(self, target_position):
# Перемещение в сторону цели
direction_x = 0 if target_position[0] == self.position[0] else (1 if target_position[0] > self.position[0] else -1)
direction_y = 0 if target_position[1] == self.position[1] else (1 if target_position[1] > self.position[1] else -1)
self.position = (self.position[0] + direction_x,
self.position[1] + direction_y)
print(f"{self.name} преследует игрока: {self.position}")
def evade(self, threat_position):
# Уклонение от угрозы
direction_x = 0 if threat_position[0] == self.position[0] else (-1 if threat_position[0] > self.position[0] else 1)
direction_y = 0 if threat_position[1] == self.position[1] else (-1 if threat_position[1] > self.position[1] else 1)
self.position = (self.position[0] + direction_x,
self.position[1] + direction_y)
print(f"{self.name} уклоняется от угрозы: {self.position}")
# Пример использования
npc = NPC("Гоблин")
npc.patrol()
npc.chase((5, 5))
npc.evade((3, 3))
📌 В этом коде реализован простой AI для NPC в играх. Он может патрулировать, преследовать игрока и уклоняться от угроз. Такой подход позволяет сделать игровой процесс более интерактивным и интересным.
Подпишись 👉🏻 @KodduuPython 🤖
💡 Lazy Evaluation и генераторы для обработки больших файлов
📌 Этот код демонстрирует использование генераторов для отложенной обработки больших файлов журнала. Вместо загрузки всего файла в память, каждая строка обрабатывается по мере необходимости, что экономит ресурсы. Полезно для анализа логов серверов или других больших файлов данных, когда память ограничена или файл слишком велик для однократной загрузки.
Подпишись 👉🏻 @KodduuPython 🤖
import logging
def read_large_file(file_path):
try:
with open(file_path, 'r') as file:
yield from (line.strip() for line in file)
except FileNotFoundError:
logging.error(f"Файл {file_path} не найден.")
except Exception as e:
logging.error(f"Произошла ошибка: {e}")
def process_log_file(file_path):
total_lines = 0
error_count = 0
for line in read_large_file(file_path):
total_lines += 1
if "ERROR" in line:
error_count += 1
return total_lines, error_count
# Пример использования
file_path = "server.log"
total_lines, error_count = process_log_file(file_path)
print(f"Всего строк: {total_lines}, ошибок: {error_count}")
📌 Этот код демонстрирует использование генераторов для отложенной обработки больших файлов журнала. Вместо загрузки всего файла в память, каждая строка обрабатывается по мере необходимости, что экономит ресурсы. Полезно для анализа логов серверов или других больших файлов данных, когда память ограничена или файл слишком велик для однократной загрузки.
Подпишись 👉🏻 @KodduuPython 🤖
👍3
🔄 Функциональные потоки данных с использованием itertools
📌 Библиотека
Подпишись 👉🏻 @KodduuPython 🤖
from itertools import islice, chain
def process_large_data(data):
# Пример использования itertools.chain для объединения нескольких списков
combined_data = chain(data, ['дополнительно', 'данные'])
# Пример использования itertools.islice для ленивой выборки данных
sliced_data = islice(combined_data, 5) # Возьмем только первые 5 элементов
return list(sliced_data)
# Пример использования
data = ['элемент1', 'элемент2', 'элемент3']
print(process_large_data(data))
📌 Библиотека
itertools предоставляет мощные инструменты для работы с итераторами, что помогает обрабатывать большие объемы данных эффективно. В этом примере chain объединяет списки, а islice позволяет отбирать элементы без использования циклов, что экономит память и улучшает производительность.Подпишись 👉🏻 @KodduuPython 🤖
🎮 Разработка многопользовательской игры на Python с WebSockets
📌 Этот код демонстрирует создание простого WebSocket сервера на Python с использованием библиотеки
Подпишись 👉🏻 @KodduuPython 🤖
import asyncio
import websockets
connected_clients = set()
async def handle_connection(websocket, path):
# Добавляем нового клиента в список подключений
connected_clients.add(websocket)
try:
async for message in websocket:
# Рассылаем сообщение всем подключенным клиентам
await asyncio.gather(*[client.send(message) for client in connected_clients])
except websockets.exceptions.ConnectionClosed:
print(f"Клиент отключился: {websocket}")
finally:
# Удаляем клиента из списка подключений при отключении
connected_clients.remove(websocket)
async def start_server():
# Запуск WebSocket сервера на указанном порту
server = await websockets.serve(handle_connection, "localhost", 8765)
print("Сервер запущен на ws://localhost:8765")
await server.wait_closed()
# Запуск сервера
asyncio.run(start_server())
📌 Этот код демонстрирует создание простого WebSocket сервера на Python с использованием библиотеки
websockets. Сервер обрабатывает подключения клиентов и пересылает сообщения между ними, позволяя создать базовую многопользовательскую систему обмена данными в реальном времени. Подобная архитектура может быть использована для создания игр в реальном времени, чатов и других приложений, требующих мгновенной синхронизации между клиентами.Подпишись 👉🏻 @KodduuPython 🤖
⚡3
🚀 Оптимизация производительности с помощью Cython
📌 Этот код показывает, как Cython может ускорить выполнение циклов через простой пример суммирования чисел. Мы создаем Cython модуль, компилируем его и получаем значительное ускорение по сравнению с Python. Это полезно для оптимизации критически важных участков в сложных вычислительных приложениях.
🛠 pip install Cython
Подпишись 👉🏻 @KodduuPython 🤖
# example_cython.pyx
def cython_function(int n):
cdef int i, total = 0
for i in range(n):
total += i
return total
# Python-функция для сравнения
def python_function(n: int) -> int:
total = 0
for i in range(n):
total += i
return total
# Пример использования
if __name__ == "__main__":
from time import time
import pyximport
pyximport.install()
# Импортируем скомпилированный модуль
import example_cython
n = 10000000
start = time()
result_python = python_function(n)
end = time()
print(f"Python: {end - start:.2f} seconds, Result: {result_python}")
start = time()
result_cython = example_cython.cython_function(n)
end = time()
print(f"Cython: {end - start:.2f} seconds, Result: {result_cython}")
📌 Этот код показывает, как Cython может ускорить выполнение циклов через простой пример суммирования чисел. Мы создаем Cython модуль, компилируем его и получаем значительное ускорение по сравнению с Python. Это полезно для оптимизации критически важных участков в сложных вычислительных приложениях.
🛠 pip install Cython
Подпишись 👉🏻 @KodduuPython 🤖
❤1
🔌 Автоматизация сетевого администрирования с Paramiko
📌 Этот код демонстрирует, как использовать библиотеку Paramiko для автоматизации SSH-подключений и выполнения команд на удалённых устройствах. Функция
Подпишись 👉🏻 @KodduuPython 🤖
import paramiko
from paramiko.ssh_exception import SSHException, NoValidConnectionsError, AuthenticationException
def execute_ssh_command(host: str, username: str, password: str, command: str) -> str:
try:
with paramiko.SSHClient() as client:
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
client.connect(hostname=host, username=username, password=password)
stdin, stdout, stderr = client.exec_command(command)
result = stdout.read().decode().strip()
return result
except AuthenticationException:
return "Authentication failed, please verify your credentials."
except (SSHException, NoValidConnectionsError) as e:
return f"SSH Error: {e}"
except Exception as e:
return f"Unexpected Error: {e}"
# Пример использования
output = execute_ssh_command('192.168.1.1', 'admin', 'password123', 'show ip interface brief')
print(output)
📌 Этот код демонстрирует, как использовать библиотеку Paramiko для автоматизации SSH-подключений и выполнения команд на удалённых устройствах. Функция
execute_ssh_command подключается к сетевому устройству по SSH, выполняет указанную команду и возвращает её результат. Это полезно для автоматизации задач сетевого администрирования, таких как мониторинг и управление устройствами.Подпишись 👉🏻 @KodduuPython 🤖
💡 Управляем освещением с помощью Raspberry Pi и Python
📌 Этот скрипт управляет светодиодом, подключенным к Raspberry Pi, включая и выключая его на заданное время. Используется библиотека RPi.GPIO для настройки пина в режиме вывода. Это может быть полезно для создания простых систем управления освещением в умном доме.
Подпишись 👉🏻 @KodduuPython 🤖
import RPi.GPIO as GPIO
import time
def setup_gpio(pin: int):
"""Настройка GPIO-пина для вывода"""
GPIO.setmode(GPIO.BCM)
GPIO.setup(pin, GPIO.OUT)
GPIO.output(pin, GPIO.LOW)
def toggle_light(pin: int, duration: int):
"""Включает свет на заданное время, затем выключает"""
try:
setup_gpio(pin)
GPIO.output(pin, GPIO.HIGH)
print("Свет включен.")
time.sleep(duration)
GPIO.output(pin, GPIO.LOW)
print("Свет выключен.")
except Exception as e:
print(f"Ошибка: {e}")
finally:
GPIO.cleanup(pin) # Очищаем только используемый пин
# Пример использования
if __name__ == "__main__":
toggle_light(pin=18, duration=5)
📌 Этот скрипт управляет светодиодом, подключенным к Raspberry Pi, включая и выключая его на заданное время. Используется библиотека RPi.GPIO для настройки пина в режиме вывода. Это может быть полезно для создания простых систем управления освещением в умном доме.
Подпишись 👉🏻 @KodduuPython 🤖
❤4
🌍 Создание процедурно-генерируемого мира с Perlin Noise
📌 Этот код показывает, как использовать Perlin Noise для создания процедурно-генерируемой карты высот. Функция
🛠 pip install noise matplotlib
Используя данную технику, разработчики могут генерировать разнообразные игровые миры, легко адаптируемые под различные жанры.
Подпишись 👉🏻 @KodduuPython 🤖
import numpy as np
from noise import pnoise2
import matplotlib.pyplot as plt
def generate_height_map(width: int, height: int, scale: float) -> np.ndarray:
height_map = np.zeros((width, height))
for x in range(width):
for y in range(height):
height_map[x][y] = pnoise2(x / scale, y / scale, octaves=6, persistence=0.5, lacunarity=2.0)
return height_map
def display_height_map(height_map: np.ndarray):
plt.imshow(height_map, cmap='terrain')
plt.colorbar()
plt.title("Procedurally Generated World")
plt.show()
# Пример использования
width, height, scale = 100, 100, 50.0
height_map = generate_height_map(width, height, scale)
display_height_map(height_map)
📌 Этот код показывает, как использовать Perlin Noise для создания процедурно-генерируемой карты высот. Функция
generate_height_map создает двумерный массив высот, применяя параметры шума, такие как октавы и масштаб. Это полезно для игр, требующих уникального ландшафта, например, roguelike или sandbox.🛠 pip install noise matplotlib
Используя данную технику, разработчики могут генерировать разнообразные игровые миры, легко адаптируемые под различные жанры.
Подпишись 👉🏻 @KodduuPython 🤖
👍2
Forwarded from AIGENTTO
WP умер, перевёл все сайты на DarwinClaw (аналог OpenClaw)
Раньше часто использовал WordPress с кучей плагинов для создания сайтов и лендингов. Сейчас поставил DarwinClaw на чистую машину Ubuntu и сказал ему сделать всю инфраструктуру для сайтов. Он поставил nginx + node.js.
Потом попросил перенести сайты, указав ссылки на них. Он создал все сайты.
Попросил добавить git и репозиторий для хранения этих сайтов. То есть всегда можно быстро откатить изменения.
Теперь полностью управляю этим зоопарком лендингов через TG команды типа:
- Убери с сайта X вот это
- Добавь на сайт Y вот то
- Поменяй стиль сайта Z на ...
- Откати последние изменения на сайте XM
- Замени икноку на сайте XS на более современную
- Собери статистику по заполнению форм с сайтов X, Y, Z в формате
Раньше это либо делали люди за деньги, либо WP с его тяжёлой инфраструктурой, теперь всё делает бот на DarwinClaw. Примерно то же самое можно сделать на OpenClaw.
#КЕЙС
Подпишись 👉🏻 @aigentto 🤖
Раньше часто использовал WordPress с кучей плагинов для создания сайтов и лендингов. Сейчас поставил DarwinClaw на чистую машину Ubuntu и сказал ему сделать всю инфраструктуру для сайтов. Он поставил nginx + node.js.
Потом попросил перенести сайты, указав ссылки на них. Он создал все сайты.
Попросил добавить git и репозиторий для хранения этих сайтов. То есть всегда можно быстро откатить изменения.
Теперь полностью управляю этим зоопарком лендингов через TG команды типа:
- Убери с сайта X вот это
- Добавь на сайт Y вот то
- Поменяй стиль сайта Z на ...
- Откати последние изменения на сайте XM
- Замени икноку на сайте XS на более современную
- Собери статистику по заполнению форм с сайтов X, Y, Z в формате
Раньше это либо делали люди за деньги, либо WP с его тяжёлой инфраструктурой, теперь всё делает бот на DarwinClaw. Примерно то же самое можно сделать на OpenClaw.
#КЕЙС
Подпишись 👉🏻 @aigentto 🤖
⚡️ Эффективная работа с большими данными через Dask
📌 Этот код демонстрирует, как использовать Dask для обработки больших объемов данных. Мы создаем массив случайных чисел и вычисляем их сумму, применяя Dask для распределённых вычислений. Такой подход позволяет работать с массивами, которые не помещаются в оперативную память, и ускоряет вычисления.
Подпишись 👉🏻 @KodduuPython 🤖
import dask.array as da
def compute_large_array_sum(array_size: int) -> float:
try:
# Создаем большой Dask массив
large_array = da.random.random((array_size, array_size), chunks=(1000, 1000))
# Вычисляем сумму всех элементов
total_sum = large_array.sum().compute()
return total_sum
except Exception as e:
print(f"Ошибка при вычислении: {e}")
return 0.0
# Пример использования
if __name__ == "__main__":
size = 10000 # Размер массива 10000x10000
total = compute_large_array_sum(size)
print(f"Сумма элементов массива: {total}")
📌 Этот код демонстрирует, как использовать Dask для обработки больших объемов данных. Мы создаем массив случайных чисел и вычисляем их сумму, применяя Dask для распределённых вычислений. Такой подход позволяет работать с массивами, которые не помещаются в оперативную память, и ускоряет вычисления.
Подпишись 👉🏻 @KodduuPython 🤖
⚡1
Forwarded from AIGENTTO
WP умер, перевёл все сайты на DarwinClaw (аналог OpenClaw)
Раньше часто использовал WordPress с кучей плагинов для создания сайтов и лендингов. Сейчас поставил DarwinClaw на чистую машину Ubuntu и сказал ему сделать всю инфраструктуру для сайтов. Он поставил nginx + node.js.
Потом попросил перенести сайты, указав ссылки на них. Он создал все сайты.
Попросил добавить git и репозиторий для хранения этих сайтов. То есть всегда можно быстро откатить изменения.
Теперь полностью управляю этим зоопарком лендингов через TG команды типа:
- Убери с сайта X вот это
- Добавь на сайт Y вот то
- Поменяй стиль сайта Z на ...
- Откати последние изменения на сайте XM
- Замени икноку на сайте XS на более современную
- Собери статистику по заполнению форм с сайтов X, Y, Z в формате
Раньше это либо делали люди за деньги, либо WP с его тяжёлой инфраструктурой, теперь всё делает бот на DarwinClaw. Примерно то же самое можно сделать на OpenClaw.
#КЕЙС
Подпишись 👉🏻 @aigentto 🤖
Раньше часто использовал WordPress с кучей плагинов для создания сайтов и лендингов. Сейчас поставил DarwinClaw на чистую машину Ubuntu и сказал ему сделать всю инфраструктуру для сайтов. Он поставил nginx + node.js.
Потом попросил перенести сайты, указав ссылки на них. Он создал все сайты.
Попросил добавить git и репозиторий для хранения этих сайтов. То есть всегда можно быстро откатить изменения.
Теперь полностью управляю этим зоопарком лендингов через TG команды типа:
- Убери с сайта X вот это
- Добавь на сайт Y вот то
- Поменяй стиль сайта Z на ...
- Откати последние изменения на сайте XM
- Замени икноку на сайте XS на более современную
- Собери статистику по заполнению форм с сайтов X, Y, Z в формате
Раньше это либо делали люди за деньги, либо WP с его тяжёлой инфраструктурой, теперь всё делает бот на DarwinClaw. Примерно то же самое можно сделать на OpenClaw.
#КЕЙС
Подпишись 👉🏻 @aigentto 🤖
🎮 Мультиплеерные игры на Python: создание сетевого взаимодействия
📌 Этот код демонстрирует создание простого сервера на Python для обработки клиентских подключений в асинхронном режиме. Используя
Подпишись 👉🏻 @KodduuPython 🤖
import asyncio
async def handle_client(reader, writer):
try:
while True:
data = await reader.read(100)
if not data:
break
message = data.decode()
print(f"Received: {message}")
writer.write(data)
await writer.drain()
except Exception as e:
print(f"Error: {e}")
finally:
writer.close()
await writer.wait_closed()
async def main(host='127.0.0.1', port=8888):
server = await asyncio.start_server(handle_client, host, port)
async with server:
print(f"Server listening on {host}:{port}")
await server.serve_forever()
if __name__ == "__main__":
try:
asyncio.run(main())
except KeyboardInterrupt:
print("Server stopped")
📌 Этот код демонстрирует создание простого сервера на Python для обработки клиентских подключений в асинхронном режиме. Используя
asyncio, сервер может одновременно обрабатывать несколько клиентов для обмена данными. Код устойчив к ошибкам, поддерживает многопользовательское взаимодействие и легко расширяем для создания сложных многопользовательских игр.Подпишись 👉🏻 @KodduuPython 🤖
⚡1
📸 Эффективная обработка изображений с OpenCV и multiprocessing
📌 Этот код демонстрирует, как можно ускорить обработку изображений с помощью OpenCV и библиотеки multiprocessing. Он параллельно изменяет размер изображений из указанной директории, что значительно повышает производительность при работе с большими наборами данных. Это полезно для приложений компьютерного зрения, где обработка изображений является частой задачей.
Подпишись 👉🏻 @KodduuPython 🤖
import cv2
import os
from multiprocessing import Pool
def process_image(image_path: str, output_dir: str):
try:
image = cv2.imread(image_path)
if image is None:
raise ValueError(f"Cannot open image {image_path}")
# Пример обработки: изменение размера изображения
resized_image = cv2.resize(image, (100, 100))
_, filename = os.path.split(image_path)
output_path = os.path.join(output_dir, filename)
cv2.imwrite(output_path, resized_image)
print(f"Processed {filename}")
except Exception as e:
print(f"Error processing {image_path}: {e}")
def parallel_image_processing(input_dir: str, output_dir: str, num_workers: int = 4):
if not os.path.exists(output_dir):
os.makedirs(output_dir)
image_files = [os.path.join(input_dir, f) for f in os.listdir(input_dir) if f.endswith(('.png', '.jpg', '.jpeg'))]
with Pool(processes=num_workers) as pool:
pool.starmap(process_image, [(image_path, output_dir) for image_path in image_files])
# Пример использования
input_directory = 'images/input'
output_directory = 'images/output'
parallel_image_processing(input_directory, output_directory)
📌 Этот код демонстрирует, как можно ускорить обработку изображений с помощью OpenCV и библиотеки multiprocessing. Он параллельно изменяет размер изображений из указанной директории, что значительно повышает производительность при работе с большими наборами данных. Это полезно для приложений компьютерного зрения, где обработка изображений является частой задачей.
Подпишись 👉🏻 @KodduuPython 🤖
⚡1
Forwarded from AIGENTTO
Как применить OpenClaw в вашей компании?
Многие сейчас просят перейти на технологию OpenClaw. Но для начала важно понимать, что это и как оно может функционировать в B2B/Enterprise.
Во-первых, OpenClaw - это индивидуальный бот, собранный из уже существующих технологий. Ключевое здесь - это индивидуальный бот, то есть он будет отвечать и делать так, как его научит/сконфигурирует конкретный человек. Поэтому, если к одному боту подключить всех сотрудников, то каждый будет его переобучать при каждом запросе, и любая информация будет ходить между всеми сотрудниками.
Во-вторых, OpenClaw умеет делать почти все, но это достигается за счёт полного доступа к компьютеру, на котором он запускается. Это требует много ресурсов (от ~1000 руб с функцией браузинга в месяц за виртуальную машину для каждого сотрудника).
Поэтому нужен новый OpenClaw для B2B/Enterprise, который сможет быть индивидуальным ботом для каждого сотрудника при неизменности общих знаний и без возможности получить чужие персональные данные и знания.
Также эта версия должна быть оптимальнее по потреблению ресурсов, чтобы не выделять целую виртуальную машину под каждого сотрудника.
И конечно, эта версия должна быть безопаснее - защита от prompt injections и доступы к данным компании по ролям через single sign-on компании.
К сожалению, таких версий нет в open source. Поэтому мы начали разрабатывать свою версию OpenClaw (DarwinClaw.RU) как раз для B2B/Enterprise.
На тестирование новой версии записывайтесь здесь. Мы уже начинаем внедрение в двух очень больших компаниях.
Подпишись 👉🏻 @aigentto 🤖
Многие сейчас просят перейти на технологию OpenClaw. Но для начала важно понимать, что это и как оно может функционировать в B2B/Enterprise.
Во-первых, OpenClaw - это индивидуальный бот, собранный из уже существующих технологий. Ключевое здесь - это индивидуальный бот, то есть он будет отвечать и делать так, как его научит/сконфигурирует конкретный человек. Поэтому, если к одному боту подключить всех сотрудников, то каждый будет его переобучать при каждом запросе, и любая информация будет ходить между всеми сотрудниками.
Во-вторых, OpenClaw умеет делать почти все, но это достигается за счёт полного доступа к компьютеру, на котором он запускается. Это требует много ресурсов (от ~1000 руб с функцией браузинга в месяц за виртуальную машину для каждого сотрудника).
Поэтому нужен новый OpenClaw для B2B/Enterprise, который сможет быть индивидуальным ботом для каждого сотрудника при неизменности общих знаний и без возможности получить чужие персональные данные и знания.
Также эта версия должна быть оптимальнее по потреблению ресурсов, чтобы не выделять целую виртуальную машину под каждого сотрудника.
И конечно, эта версия должна быть безопаснее - защита от prompt injections и доступы к данным компании по ролям через single sign-on компании.
К сожалению, таких версий нет в open source. Поэтому мы начали разрабатывать свою версию OpenClaw (DarwinClaw.RU) как раз для B2B/Enterprise.
На тестирование новой версии записывайтесь здесь. Мы уже начинаем внедрение в двух очень больших компаниях.
Подпишись 👉🏻 @aigentto 🤖
🔄 Реактивное программирование с RxPy
📌 Этот код демонстрирует использование реактивного программирования с библиотекой RxPy для обработки потоков данных. Мы создаем поток событий, фильтруем его для получения только четных чисел, масштабируем их и обрабатываем. Такой подход особенно полезен в приложениях, где нужно реагировать на события в реальном времени, например, в системах мониторинга или интерактивных интерфейсах.
🛠 pip install rx
Этот пример показывает, как с помощью RxPy можно легко управлять потоками данных и обрабатывать события, используя функциональные подходы для создания более выразительного и гибкого кода.
Подпишись 👉🏻 @KodduuPython 🤖
from rx import create
from rx.operators import filter, map
from rx.core.typing import Observer, Scheduler
def event_stream(observer: Observer, scheduler: Scheduler):
# Список событий для обработки
events = [1, 2, 3, 4, 5, 6]
for event in events:
observer.on_next(event) # Передаем событие в поток
observer.on_completed() # Завершаем поток
def handle_event(event):
print(f"Обработано событие: {event}")
source = create(event_stream)
source.pipe(
filter(lambda x: x % 2 == 0), # Фильтруем только четные числа
map(lambda x: x * 10) # Масштабируем числа
).subscribe(
on_next=handle_event,
on_error=lambda e: print(f"Ошибка: {e}"), # Обработка ошибок
on_completed=lambda: print("Обработка завершена") # Завершение обработки
)
📌 Этот код демонстрирует использование реактивного программирования с библиотекой RxPy для обработки потоков данных. Мы создаем поток событий, фильтруем его для получения только четных чисел, масштабируем их и обрабатываем. Такой подход особенно полезен в приложениях, где нужно реагировать на события в реальном времени, например, в системах мониторинга или интерактивных интерфейсах.
🛠 pip install rx
Этот пример показывает, как с помощью RxPy можно легко управлять потоками данных и обрабатывать события, используя функциональные подходы для создания более выразительного и гибкого кода.
Подпишись 👉🏻 @KodduuPython 🤖
⚡1
🏠 Умный дом с Raspberry Pi и Python
📌 Этот код управляет светодиодом на Raspberry Pi с помощью Python и библиотеки RPi.GPIO. Он включает и выключает светодиод с заданной периодичностью, что полезно для простых проектов умного дома, таких как управление освещением. Перед использованием убедитесь, что светодиод правильно подключен к пину GPIO 18.
Подпишись 👉🏻 @KodduuPython 🤖
import RPi.GPIO as GPIO
import time
# Настройка GPIO
GPIO.setmode(GPIO.BCM)
GPIO.setwarnings(False) # Отключить предупреждения
LED_PIN = 18
GPIO.setup(LED_PIN, GPIO.OUT)
def toggle_light(state: bool):
"""Включает или выключает светодиод."""
GPIO.output(LED_PIN, state)
try:
# Пример использования: мигание светодиода
for _ in range(5):
toggle_light(True)
time.sleep(1)
toggle_light(False)
time.sleep(1)
finally:
GPIO.cleanup()
📌 Этот код управляет светодиодом на Raspberry Pi с помощью Python и библиотеки RPi.GPIO. Он включает и выключает светодиод с заданной периодичностью, что полезно для простых проектов умного дома, таких как управление освещением. Перед использованием убедитесь, что светодиод правильно подключен к пину GPIO 18.
Подпишись 👉🏻 @KodduuPython 🤖
⚡1