Сделал мини игру в честь полета на Луну. Попробуй NASA Artemis Lunar FlyBy сам. Сможешь сделать лунный маневр?

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

🚀 Улучшение производительности с помощью многопоточности и многопроцессорности

from concurrent.futures import ProcessPoolExecutor

def process_image(image_path: str) -> str:
    # Замените на реальную логику обработки изображения
    return f"Обработано изображение: {image_path}"

def parallel_processing(file_paths: list):
    if not file_paths:
        return []
    with ProcessPoolExecutor() as executor:
        results = list(executor.map(process_image, file_paths))
    return results

# Пример использования
if __name__ == "__main__":
    image_files = ["image1.jpg", "image2.jpg", "image3.jpg"]
    image_results = parallel_processing(image_files)
    print("Результаты обработки изображений:", image_results)

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

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

🌱 Симуляция игры "Жизнь" с NumPy

import numpy as np

def initialize_grid(size: int) -> np.ndarray:
    return np.random.choice([0, 1], size * size, p=[0.8, 0.2]).reshape(size, size)

def count_neighbors(grid: np.ndarray, x: int, y: int) -> int:
    neighbors = [
        (i, j)
        for i in range(x - 1, x + 2)
        for j in range(y - 1, y + 2)
        if (i != x or j != y) and (0 <= i < grid.shape[0]) and (0 <= j < grid.shape[1])
    ]
    return sum(grid[i, j] for i, j in neighbors)

def update_grid(grid: np.ndarray) -> np.ndarray:
    new_grid = grid.copy()
    for x in range(grid.shape[0]):
        for y in range(grid.shape[1]):
            neighbors_count = count_neighbors(grid, x, y)
            if grid[x, y] == 1 and neighbors_count not in (2, 3):
                new_grid[x, y] = 0
            elif grid[x, y] == 0 and neighbors_count == 3:
                new_grid[x, y] = 1
    return new_grid

# Пример использования
grid_size = 10
grid = initialize_grid(grid_size)

for _ in range(5):
    print(grid)
    grid = update_grid(grid)

📌 Этот код реализует "Игру в жизнь" Джона Конуэя, где клетки на сетке живут, умирают или рождаются в зависимости от числа соседей. Используется библиотека NumPy для управления состоянием сетки. Это полезно для изучения клеточных автоматов и моделирования процессов.

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

🎮 Разработка многопользовательской игры на Flask и WebSockets

from flask import Flask, render_template, request
from flask_socketio import SocketIO, emit

app = Flask(__name__)
app.config['SECRET_KEY'] = 'your_secret_key'
socketio = SocketIO(app)

players = set()

@app.route('/')
def index():
    return render_template('index.html')

@socketio.on('connect')
def handle_connect():
    players.add(request.sid)
    emit('player_count', {'count': len(players)}, broadcast=True)

@socketio.on('disconnect')
def handle_disconnect():
    players.discard(request.sid)
    emit('player_count', {'count': len(players)}, broadcast=True)

@socketio.on('move')
def handle_move(data):
    emit('move', data, broadcast=True, include_self=False)

if __name__ == '__main__':
    socketio.run(app, debug=True)

# Пример использования: создайте файл index.html с минимальной логикой подключения

📌 Этот код создаёт базовую структуру для многопользовательской игры, используя Flask и WebSockets. Сервер обрабатывает подключение игроков и синхронизирует их действия в реальном времени. Подобный подход можно использовать для создания простых интерактивных игр и приложений с обменом данных между клиентами.

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

📊 Визуализация данных IoT в реальном времени с Plotly

import random
import time
import datetime
import plotly.graph_objects as go
from plotly.subplots import make_subplots

def get_sensor_data() -> float:
    """Имитирует получение данных с датчика."""
    return random.uniform(20.0, 25.0)

def update_plot(fig, x_data, y_data):
    """Обновляет график новыми данными."""
    fig.data[0].x = x_data
    fig.data[0].y = y_data
    fig.update_layout(transition_duration=500)
    fig.show()

def visualize_iot_data():
    """Визуализирует данные с IoT-устройств в реальном времени."""
    fig = make_subplots(rows=1, cols=1)
    fig.add_trace(go.Scatter(x=[], y=[], mode='lines+markers', name='Температура'))

    x_data, y_data = [], []

    try:
        while True:
            x_data.append(datetime.datetime.now().strftime('%H:%M:%S'))
            y_data.append(get_sensor_data())
            update_plot(fig, x_data, y_data)
            time.sleep(1)
    except KeyboardInterrupt:
        print("Визуализация остановлена")
    except Exception as e:
        print(f"Ошибка: {e}")

# Пример использования
visualize_iot_data()

📌 Этот код демонстрирует, как собирать и визуализировать данные с IoT-устройств в реальном времени. Он использует Plotly для создания динамического графика, обновляющегося каждую секунду с новыми данными датчика. Такой подход удобен для мониторинга параметров в реальном времени.

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

Отказ от Cursor 🙅‍♂️

Луну обогнули 🌔, вернёмся на землю.

Я писал про кейс перевода всех сайтов на OpenClaw. Теперь переводим разработку больших продуктов.

Было так: Cursor пишет код как ассистент разработчика. Говоришь ему "построй план", корректируешь, и с нуля быстро можешь написать довольно масштабную систему.

После этого начинается просто корректировка, новые фичи и фиксы багов. И как правило, в большой системе, обросшей фичами, внедрение нового занимает время, особенно если команда не Full Stack (отдельно Front и отдельно Back разработка).

Но я попробовал и вынес дальнейшую разработку двух больших систем в OpenClaw.

По сути, на VPS с OpenClaw (полный sudo доступ) подтягиваем нужные репозитории. И в чате ему говорим: в репо A добавь X, почини Y, найди баги. Вроде все то же самое, как и в Cursor, но нет, все мгновенно оказывается в Prod, и в цикле там уже включено UI тестирование, потому что есть playwright + browser.

Dev Pipeline теперь это не 3 отдельные среды (prod + test + dev), а Front+Back Dev + QA + DevOps со средним time-2-market в 3 недели, а просто три ветки prod/test/dev, все три постоянно подняты, и time-2-market падает до часов.

Этап дизайна в Figma тоже уходит. Зачем просить дизайнера, который все время занят, сделать новый дизайн, если можно просто сказать боту: "Накидай три версии дизайна новой фичи сразу с самой фичей прямо в dev-ветке", и уже через 5 минут смотреть, выбирать и корректировать.

Написание больших систем с нуля все еще удобнее в Cursor, но дальнейшее развитие работает на OpenClaw на 4+.

Напомню, мы используем свою оптимизированную версию OpenClaw - DarwinClaw, которая скоро должна стать доступной для всех.

#КЕЙС

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

Создание AI для NPC с алгоритмом A*

from heapq import heappush, heappop
from typing import List, Tuple

def heuristic(a: Tuple[int, int], b: Tuple[int, int]) -> float:
    # Вычисляем манхэттенское расстояние
    return abs(a[0] - b[0]) + abs(a[1] - b[1])

def a_star_search(start: Tuple[int, int], goal: Tuple[int, int], grid: List[List[int]]) -> List[Tuple[int, int]]:
    open_set = []
    heappush(open_set, (0, start))
    came_from = {}
    g_score = {start: 0}
    f_score = {start: heuristic(start, goal)}

    while open_set:
        _, current = heappop(open_set)
        
        if current == goal:
            path = []
            while current in came_from:
                path.append(current)
                current = came_from[current]
            path.append(start)  # Добавляем начальную позицию в путь
            return path[::-1]
        
        for dx, dy in [(-1, 0), (1, 0), (0, -1), (0, 1)]:
            neighbor = (current[0] + dx, current[1] + dy)
            tentative_g_score = g_score[current] + 1

            if 0 <= neighbor[0] < len(grid) and 0 <= neighbor[1] < len(grid[0]) and grid[neighbor[0]][neighbor[1]] == 0:
                if neighbor not in g_score or tentative_g_score < g_score[neighbor]:
                    came_from[neighbor] = current
                    g_score[neighbor] = tentative_g_score
                    f_score[neighbor] = tentative_g_score + heuristic(neighbor, goal)
                    if neighbor not in [i[1] for i in open_set]:
                        heappush(open_set, (f_score[neighbor], neighbor))
    
    return []

# Пример использования
grid = [
    [0, 1, 0, 0, 0],
    [0, 1, 0, 1, 0],
    [0, 0, 0, 1, 0],
    [0, 1, 1, 0, 0],
    [0, 0, 0, 0, 0]
]
start = (0, 0)
goal = (4, 4)
path = a_star_search(start, goal, grid)
print("Путь:", path)

📌 Алгоритм A* используется для поиска оптимального пути на карте, что делает его идеальным для реализации AI в играх.

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

☁️ Микросервис для прогноза погоды на Flask и OpenWeatherMap API

from flask import Flask, jsonify, request
import requests
import os

app = Flask(__name__)

API_KEY = os.getenv('OPENWEATHERMAP_API_KEY')  # Получение ключа API из переменных окружения
BASE_URL = 'http://api.openweathermap.org/data/2.5/weather'

def get_weather_data(city: str) -> dict:
    try:
        response = requests.get(BASE_URL, params={'q': city, 'appid': API_KEY, 'units': 'metric'})
        response.raise_for_status()
        return response.json()
    except requests.HTTPError as http_err:
        return {'error': f'HTTP error occurred: {http_err}'}
    except Exception as err:
        return {'error': f'Other error occurred: {err}'}

@app.route('/weather', methods=['GET'])
def weather():
    city = request.args.get('city')
    if not city:
        return jsonify({'error': 'Город не указан'}), 400

    weather_data = get_weather_data(city)
    if 'error' in weather_data:
        return jsonify(weather_data), 500
    
    return jsonify({
        'city': city,
        'temperature': weather_data['main']['temp'],
        'description': weather_data['weather'][0]['description']
    })

if __name__ == '__main__':
    app.run(debug=True)

📌 Этот код создает микросервис на Flask, который предоставляет прогноз погоды по заданному городу, используя API OpenWeatherMap. Запросы к API обрабатываются с учетом ошибок, а данные возвращаются в виде JSON. Используйте переменные окружения для хранения ключа API, чтобы повысить безопасность.

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

🔗 Композирование функций для сложных операций

from typing import Callable

def compose(*functions: Callable) -> Callable:
    def composed_function(x):
        for func in reversed(functions):  # Применение функций в обратном порядке
            x = func(x)
        return x
    return composed_function

# Пример простых функций
def double(x: int) -> int:
    return x * 2

def increment(x: int) -> int:
    return x + 1

# Композирование функций
process = compose(double, increment)

# Пример использования
result = process(3)  # Результат: (3 + 1) * 2 = 8
print(result)

📌 Композирование позволяет объединять простые функции в более сложные, улучшая читаемость и поддержку кода. В данном примере функции double и increment комбинируются в цепочку, применяемую к значению 3. Это особенно полезно в обработке данных, где часто требуется выполнить серию преобразований.

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

⚡️ Ускоряем Python с Cython

# Пример Cython кода для ускорения вычислений
def sum_of_squares(int n):
    cdef int i
    cdef long total = 0
    for i in range(n):
        total += i * i
    return total

# Сохраните этот код в файле с расширением .pyx и скомпилируйте с помощью setup.py
# Пример setup.py для компиляции

from setuptools import setup
from Cython.Build import cythonize

setup(
    ext_modules=cythonize("example.pyx")
)

# Выполните команду: python setup.py build_ext --inplace
# Пример использования
import example

result = example.sum_of_squares(1000000)
print(result)

📌 Cython позволяет преобразовывать Python-код в C-код, что значительно ускоряет выполнение ресурсоемких операций. Этот пример показывает, как использовать Cython для оптимизации функции, суммирующей квадраты чисел. Преобразование и компиляция кода выполняются легко, предоставляя значительный прирост производительности для вычислительно интенсивных задач.

🛠 pip install cython

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

⚡️ Ускорение Python-кода с помощью Cython

# Пример использования Cython для ускорения вычислений
# Сначала создайте файл example.pyx с этим содержимым:

def calculate_sum(n: int) -> int:
    cdef int i, total = 0
    for i in range(n):
        total += i
    return total

# Затем создайте setup.py для компиляции:

from setuptools import setup
from Cython.Build import cythonize

setup(
    ext_modules=cythonize("example.pyx"),
)

# Для компиляции выполните:
# python setup.py build_ext --inplace

# Пример использования скомпилированного модуля:
import example

result = example.calculate_sum(1000000)
print(result)

📌 Cython позволяет компилировать Python-код в C, значительно увеличивая его производительность. В этом примере мы используем Cython для ускорения функции суммирования чисел. Подобный подход пригодится в проектах, где критически важна скорость выполнения, например, в научных вычислениях или обработке больших объемов данных.

🛠 pip install Cython

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

📊 Оптимизация базы данных: ORM vs. прямые SQL-запросы

import sqlite3
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import declarative_base, sessionmaker

# Настройка для прямых SQL-запросов
conn = sqlite3.connect(':memory:')
cursor = conn.cursor()
cursor.execute('CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT)')

# Настройка для ORM с использованием SQLAlchemy
Base = declarative_base()

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String)

engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()

def add_user_sql(name: str):
    try:
        cursor.execute('INSERT INTO users (name) VALUES (?)', (name,))
        conn.commit()
    except sqlite3.Error as e:
        print(f"SQL Error: {e}")

def add_user_orm(name: str):
    try:
        user = User(name=name)
        session.add(user)
        session.commit()
    except Exception as e:
        session.rollback()
        print(f"ORM Error: {e}")

# Пример использования
add_user_sql('Alice')
add_user_orm('Bob')

# Закрытие соединений
cursor.close()
conn.close()
session.close()

📌 В этом примере показано использование прямых SQL-запросов и ORM для добавления записей в базу данных SQLite. Прямые SQL-запросы дают больше контроля и могут быть быстрее, но менее удобны. ORM, как SQLAlchemy, упрощает работу с базой данных, автоматизируя многие процессы, но может добавить некоторую задержку из-за абстракции.

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

Полезный тул от NVIDIA для penetration тестирования вашего ИИ бота

Гоняет всякие промты, чтобы найти дыры (персональные данные и прочее) в вашем AI боте:
https://github.com/NVIDIA/garak/

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

🔐 Замыкания: Управление состоянием без классов

def create_counter(start: int = 0):
    count = start

    def increment():
        nonlocal count
        count += 1
        return count

    return increment

# Пример использования
counter = create_counter()
print(counter())  # Выведет: 1
print(counter())  # Выведет: 2
print(counter())  # Выведет: 3

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

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

🔒 Автоматизация сетевых операций с Paramiko

import paramiko
from paramiko import SSHException, AuthenticationException, BadHostKeyException

def execute_remote_command(hostname: str, username: str, password: str, command: str) -> str:
    try:
        with paramiko.SSHClient() as client:
            client.set_missing_host_key_policy(paramiko.WarningPolicy())
            client.connect(hostname, username=username, password=password)

            stdin, stdout, stderr = client.exec_command(command)
            result = stdout.read().decode()
            return result
    except AuthenticationException:
        return "Ошибка аутентификации"
    except BadHostKeyException:
        return "Неправильный ключ хоста"
    except SSHException as e:
        return f"SSH ошибка: {e}"
    except Exception as e:
        return f"Ошибка: {e}"

# Пример использования
if __name__ == "__main__":
    response = execute_remote_command(
        hostname='192.168.1.100',
        username='admin',
        password='password',
        command='ls'
    )
    print(response)

📌 Этот скрипт демонстрирует, как выполнять команды на удаленном сервере через SSH с помощью Paramiko. Он автоматически уведомляет о ключах хоста и обрабатывает ошибки подключения. Это полезно для автоматизации задач администрирования и управления серверами.

🛠 pip install paramiko

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

🧠 Быстрый старт с машинным обучением в Scikit-learn

from sklearn.datasets import load_iris
from sklearn.model_selection import train_test_split
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score

# Загрузка датасета
data = load_iris()
X, y = data.data, data.target

# Разделение данных на обучающую и тестовую выборки
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

# Создание и обучение модели
model = RandomForestClassifier(n_estimators=100, random_state=42)
model.fit(X_train, y_train)

# Предсказание и оценка модели
y_pred = model.predict(X_test)
accuracy = accuracy_score(y_test, y_pred)

print(f"Точность модели: {accuracy:.2f}")

# Обработка ошибок
try:
    # Пример использования модели на новых данных
    new_data = [[5.1, 3.5, 1.4, 0.2]]
    prediction = model.predict(new_data)
    print(f"Предсказанный класс: {prediction[0]}")
except Exception as e:
    print(f"Ошибка при предсказании: {e}")

📌 Этот код демонстрирует, как быстро создать и обучить модель машинного обучения с использованием Scikit-learn. Мы используем датасет ирисов для классификации, обучая модель случайного леса и оцениваем её точность. Такой подход позволяет быстро прототипировать решения и интегрировать их в приложения, что делает Python предпочтительным инструментом для задач машинного обучения по сравнению с C++.

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

🧪 Быстрое тестирование кода с Pytest

# Установка Pytest
# 🛠 pip install pytest

# Функция для тестирования
def add(a: int, b: int) -> int:
    return a + b

# Простой тест для функции add
def test_add():
    assert add(2, 3) == 5
    assert add(-1, 1) == 0
    assert add(0, 0) == 0

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

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

Мониторинг логов с автоматическим исправлением проблем

И так, раньше надо было использовать различные отдельные инструменты для мониторинга логов (Zabbix и прочее), чтобы в крайнем случае сделать reboot отдельных компонентов или всей системы.

Но оказалось, можно отдать задачу ReAct боту типа OpenClaw (мы используем версию B2B DarwinClaw). Он может проверять логи хоть каждые 2 минуты. Тонны токенов здесь не тратятся, он также просто делает поиск по ключевым словам. И важно, что он сам может подобрать ключевые слова и обновлять этот набор исходя из типа логов.

Дальше больше - он может не просто сделать reboot, он может исправить проблему прямо в коде, или сделать откат в git, если поймет, что это проблема новой версии 🔥

Ну то есть, по сути это полноценный support+разраб 24x7, работающий за миску токенов. Да, важно прописать ему некоторые правила (когда то-то делай так, когда это делай сяк), по сути, вашу политику работы с Prod.

#КЕЙС

Прочие полезные кейсы:
👉 WP умер, перевёл все сайты на DarwinClaw
👉 Сбор любой публичной статистики
👉 Отказ от Cursor IDE

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

### Чат-бот для психологической поддержки с использованием NLP 🤖

import random
import spacy

nlp = spacy.load("en_core_web_sm")

def get_response(user_input: str) -> str:
    doc = nlp(user_input)

    # Проверка на наличие ключевых лексем
    for token in doc:
        if token.lemma_ == "sad":
            return random.choice([
                "Мне жаль это слышать. Хочешь поговорить об этом?",
                "Грустно слышать. Может, расскажешь подробнее?",
            ])
        elif token.lemma_ == "happy":
            return random.choice([
                "Это замечательно! Что делает тебя счастливым?",
                "Рад слышать, что ты счастлив!",
            ])
    
    return "Расскажи, как ты себя чувствуешь?"

# Пример использования
def chat():
    print("Чат-бот: Привет! Как ты себя чувствуешь сегодня?")
    while True:
        user_input = input("Вы: ")
        if user_input.lower() in ["выход", "пока"]:
            print("Чат-бот: Пока! Заботься о себе.")
            break
        print(f"Чат-бот: {get_response(user_input)}")

chat()

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

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

🔒 Декораторы для улучшения безопасности API

from functools import wraps

def require_authentication(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        user_authenticated = kwargs.get('authenticated', False)
        if not user_authenticated:
            raise PermissionError("Требуется аутентификация для доступа к этой функции.")
        return func(*args, **kwargs)
    return wrapper

# Пример использования
@require_authentication
def get_user_data(user_id: int, authenticated=False) -> dict:
    # Имитация получения данных пользователя
    return {"user_id": user_id, "name": "Alice"}

try:
    print(get_user_data(user_id=1, authenticated=True))  # Успешный доступ
    print(get_user_data(user_id=2, authenticated=False)) # Ошибка доступа
except PermissionError as e:
    print(e)

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

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

⚡️ Реактивное программирование с RxPy

import rx
from rx import operators as ops
from rx.scheduler import ThreadPoolScheduler
import requests

def fetch_url(url: str) -> str:
    try:
        response = requests.get(url)
        response.raise_for_status()
        return f"Fetched {url} with status {response.status_code}"
    except requests.RequestException as e:
        return f"Error fetching {url}: {str(e)}"

urls = [
    "http://example.com",
    "http://nonexistent.url",
    "http://example.org"
]

# Настройка многопоточного планировщика
scheduler = ThreadPoolScheduler(max_workers=2)

# Создание реактивного потока
rx.from_(urls).pipe(
    ops.map(lambda url: fetch_url(url)),
    ops.do_action(lambda result: print(result))
).subscribe(
    on_error=lambda e: print(f"Error occurred: {e}"),
    on_completed=lambda: print("Fetching completed!")
)

📌 Этот код демонстрирует использование RxPy для асинхронного веб-скрапинга. Он создает поток URL-адресов, которые обрабатываются параллельно с помощью многопоточного планировщика. Каждый URL запрашивается, и результат выводится в консоль. Такой подход полезен для обработки большого количества запросов или выполнения асинхронных задач.

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