⚡️ Оптимизация с многопоточностью и асинхронностью

import asyncio
import concurrent.futures
import time

def cpu_bound_task(n: int) -> int:
    print(f"Выполняю тяжёлую задачу для {n}...")
    time.sleep(2)  # Имитация длительного вычисления
    return n * n

async def async_main(nums: list):
    if not nums:
        print("Список чисел пуст.")
        return
    loop = asyncio.get_running_loop()
    with concurrent.futures.ProcessPoolExecutor() as pool:
        results = await asyncio.gather(
            *[loop.run_in_executor(pool, cpu_bound_task, num) for num in nums]
        )
    return results

# Пример использования
if __name__ == "__main__":
    numbers = [1, 2, 3, 4, 5]
    start = time.time()
    results = asyncio.run(async_main(numbers))
    end = time.time()
    print(f"Результаты: {results}")
    print(f"Время выполнения: {end - start:.2f} секунд")

📌 Этот код демонстрирует использование многопоточности и асинхронности для оптимизации выполнения CPU-интенсивных задач в Python. Мы используем concurrent.futures.ProcessPoolExecutor в сочетании с asyncio для параллельной обработки задач, что позволяет значительно ускорить выполнение по сравнению с последовательным вариантом. Это полезно при работе с задачами, требующими длительных вычислений, такими как обработка данных или сложные вычисления. Убедитесь, что список чисел не пустой, чтобы избежать ненужных вычислений.

Подпишись 👉🏻 @KodduuPython 🤖

🚀 Последний шанс купить следующие программы курсов:

👉 FullStack Developer and Data Scientist (Python+JS+Data+CookBook)🔥🔥🔥

👉 Junior FullStack Developer: Python+JavaScript (быстрый курс) 🔥

⚡️⚡️⚡️ Только до 28 февраля ⚡️⚡️⚡️

🧬 Симуляция эволюции популяции с DEAP

import random
from deap import base, creator, tools, algorithms

# Создание классов для максимизации функции
creator.create("FitnessMax", base.Fitness, weights=(1.0,))
creator.create("Individual", list, fitness=creator.FitnessMax)

def eval_function(individual):
    return sum(individual),

toolbox = base.Toolbox()
toolbox.register("attr_float", random.random)
toolbox.register("individual", tools.initRepeat, creator.Individual, toolbox.attr_float, n=5)
toolbox.register("population", tools.initRepeat, list, toolbox.individual)

toolbox.register("mate", tools.cxBlend, alpha=0.5)
toolbox.register("mutate", tools.mutGaussian, mu=0, sigma=1, indpb=0.2)
toolbox.register("select", tools.selTournament, tournsize=3)
toolbox.register("evaluate", eval_function)

def main():
    random.seed(42)
    pop = toolbox.population(n=10)
    algorithms.eaSimple(pop, toolbox, cxpb=0.5, mutpb=0.2, ngen=20, verbose=True)

if __name__ == "__main__":
    main()

📌 Этот код демонстрирует создание и выполнение простой эволюционной модели с использованием библиотеки DEAP. Мы определяем популяцию, состоящую из индивидуумов с генами, представляющими случайные числа, и применяем процессы кроссовера и мутации, чтобы максимизировать сумму генов. Это может быть полезно для поиска оптимальных решений в задачах, где традиционные методы неэффективны.

Подпишись 👉🏻 @KodduuPython 🤖

⚡️⚡️⚡️ Доступ к продаже следующих программ курсов закрывается сегодня ⚡️⚡️⚡️

👉 FullStack Developer and Data Scientist (Python+JS+Data+CookBook)🔥🔥🔥

👉 Junior FullStack Developer: Python+JavaScript (быстрый курс) 🔥

🏃🏃‍♀️🏃‍♂️

⚡ Асинхронные сетевые операции с asyncio

import asyncio
import aiohttp

async def fetch_url(session, url):
    try:
        async with session.get(url) as response:
            return await response.text()
    except aiohttp.ClientError as e:
        print(f"Ошибка при запросе {url}: {e}")
        return None

async def main(urls):
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url) for url in urls]
        responses = await asyncio.gather(*tasks)
        return responses

# Пример использования
urls = [
    "https://www.example.com",
    "https://www.python.org",
    "https://www.asyncio.org"
]
results = asyncio.run(main(urls))
for url, content in zip(urls, results):
    print(f"Содержимое {url[:30]}: {content[:100]}...")

📌 Этот код демонстрирует, как использовать модуль asyncio и библиотеку aiohttp для выполнения асинхронных HTTP-запросов. Асинхронное программирование позволяет запускать несколько сетевых операций одновременно, что значительно улучшает производительность приложений, работающих с сетью. Такой подход полезен для веб-скрейпинга, API-запросов и других задач, где важна скорость обработки сетевых данных.

Подпишись 👉🏻 @KodduuPython 🤖

🧪 Автоматизация тестирования с помощью PyTest

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 🤖

🎮 Реализация мультиплеера с WebSockets

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 🤖

Ускорение работы с данными с помощью numpy

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 🤖

🔮 Устойчивая обработка ошибок с монадами Maybe

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 🤖

💡 Управление светодиодами с MicroPython

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 🎮

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 и генераторы для обработки больших файлов

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 🤖

🔄 Функциональные потоки данных с использованием itertools

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

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 🤖

🚀 Оптимизация производительности с помощью Cython

# 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 🤖

🔌 Автоматизация сетевого администрирования с Paramiko

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

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 🤖

🌍 Создание процедурно-генерируемого мира с Perlin Noise

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 🤖

WP умер, перевёл все сайты на DarwinClaw (аналог OpenClaw)

Раньше часто использовал WordPress с кучей плагинов для создания сайтов и лендингов. Сейчас поставил DarwinClaw на чистую машину Ubuntu и сказал ему сделать всю инфраструктуру для сайтов. Он поставил nginx + node.js.

Потом попросил перенести сайты, указав ссылки на них. Он создал все сайты.

Попросил добавить git и репозиторий для хранения этих сайтов. То есть всегда можно быстро откатить изменения.

Теперь полностью управляю этим зоопарком лендингов через TG команды типа:
- Убери с сайта X вот это
- Добавь на сайт Y вот то
- Поменяй стиль сайта Z на ...
- Откати последние изменения на сайте XM
- Замени икноку на сайте XS на более современную
- Собери статистику по заполнению форм с сайтов X, Y, Z в формате

Раньше это либо делали люди за деньги, либо WP с его тяжёлой инфраструктурой, теперь всё делает бот на DarwinClaw. Примерно то же самое можно сделать на OpenClaw.

#КЕЙС

Подпишись 👉🏻 @aigentto 🤖

⚡️ Эффективная работа с большими данными через Dask

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 🤖

WP умер, перевёл все сайты на DarwinClaw (аналог OpenClaw)

Раньше часто использовал WordPress с кучей плагинов для создания сайтов и лендингов. Сейчас поставил DarwinClaw на чистую машину Ubuntu и сказал ему сделать всю инфраструктуру для сайтов. Он поставил nginx + node.js.

Потом попросил перенести сайты, указав ссылки на них. Он создал все сайты.

Попросил добавить git и репозиторий для хранения этих сайтов. То есть всегда можно быстро откатить изменения.

Теперь полностью управляю этим зоопарком лендингов через TG команды типа:
- Убери с сайта X вот это
- Добавь на сайт Y вот то
- Поменяй стиль сайта Z на ...
- Откати последние изменения на сайте XM
- Замени икноку на сайте XS на более современную
- Собери статистику по заполнению форм с сайтов X, Y, Z в формате

Раньше это либо делали люди за деньги, либо WP с его тяжёлой инфраструктурой, теперь всё делает бот на DarwinClaw. Примерно то же самое можно сделать на OpenClaw.

#КЕЙС

Подпишись 👉🏻 @aigentto 🤖