Как применить OpenClaw в вашей компании?

Многие сейчас просят перейти на технологию OpenClaw. Но для начала важно понимать, что это и как оно может функционировать в B2B/Enterprise.

Во-первых, OpenClaw - это индивидуальный бот, собранный из уже существующих технологий. Ключевое здесь - это индивидуальный бот, то есть он будет отвечать и делать так, как его научит/сконфигурирует конкретный человек. Поэтому, если к одному боту подключить всех сотрудников, то каждый будет его переобучать при каждом запросе, и любая информация будет ходить между всеми сотрудниками.

Во-вторых, OpenClaw умеет делать почти все, но это достигается за счёт полного доступа к компьютеру, на котором он запускается. Это требует много ресурсов (от ~1000 руб с функцией браузинга в месяц за виртуальную машину для каждого сотрудника).

Поэтому нужен новый OpenClaw для B2B/Enterprise, который сможет быть индивидуальным ботом для каждого сотрудника при неизменности общих знаний и без возможности получить чужие персональные данные и знания.

Также эта версия должна быть оптимальнее по потреблению ресурсов, чтобы не выделять целую виртуальную машину под каждого сотрудника.

И конечно, эта версия должна быть безопаснее - защита от prompt injections и доступы к данным компании по ролям через single sign-on компании.

К сожалению, таких версий нет в open source. Поэтому мы начали разрабатывать свою версию OpenClaw (DarwinClaw.RU) как раз для B2B/Enterprise.

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

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

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

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 🤖

🏠 Умный дом с Raspberry Pi и Python

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 🤖

OpenClaw пожиратель токенов

Все чаще слышу: "OpenClaw пожиратель токенов". Я и сам, когда поставил поиграться, сожрал токенов на $30 за несколько часов. Тем не менее, я уже использую два своих постоянных бота на OpenClaw (R2D2 и C3PO) и научился ограничивать его аппетиты.

Самые простые настройки для ограничения аппетита:
1. Запрет/сильное ограничение Base64-скриншотов и изображений в промпте (может сэкономить миллионы токенов на один скриншот).
2. Ограничить количество итераций ReAct Loop, чтобы избежать случаев, когда модель часами ходит по кругу, тратит токены и не может решить задачу.
3. Включить prompt pruning - обрезка ненужного в памяти прямо перед отправкой запроса модели. Он работает в памяти (in-memory) и не меняет историю на диске.
4. Уменьшить или убрать проверку состояния (каждые 30 минут по умолчанию).
5. Использовать разные модели для разных задач (не стоит использовать дорогую модель codex для не coding задач, для vision выбрать лучшую модель по соотношению цена/качество).
6. Поставить LiteLLM как прокси перед OpenClaw → включить кэш (cache hits экономят огромные деньги).

Каждое ограничение, конечно, ограничивает возможности OpenClaw, но зато экономит бюджет.

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

🚀 Асинхронное программирование с asyncio

import asyncio
import aiohttp

async def fetch_url(session, url):
    try:
        async with session.get(url) as response:
            return await response.text()
    except Exception as e:
        return f"Ошибка при загрузке {url}: {e}"

async def main(urls):
    if not urls:
        print("Список URL пуст.")
        return
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        for url, content in zip(urls, results):
            print(f"Содержимое из {url[:50]}: {content[:100]}...")

urls = [
    "https://www.example.com",
    "https://www.python.org",
    "https://www.asyncio.org"
]
asyncio.run(main(urls))

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

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

🔥🔥🔥 ВАЖНО 🔥🔥🔥

Произошла вещь, от которой у всех AI-разработчиков пробежал холодок по спине.

litellm — Python-библиотека для унифицированной работы с API разных LLM была отравлена.

Один pip install - и твои SSH-ключи, AWS/GCP/Azure креды, K8s Secrets, пароли от БД, криптокошельки, все API-ключи из .env — всё это упаковывается, шифруется через AES-256 и отправляется POST-запросом на поддельный домен http://models.litellm.cloud. Если обнаруживается среда K8s , то дополнительно разворачивается привилегированный Pod на каждом узле для латерального распространения.

Самое неприятное — способ активации. В пакет подложили файл litellm_init.pth размером 34 КБ. В Python .pth — это файлы конфигурации путей, которые автоматически обрабатываются модулем site при старте интерпретатора. Если строка начинается с import — она просто выполняется. Атака использует это поведение:

import os, subprocess, sys; subprocess.Popen([sys.executable, "-c", "import base64; exec(base64.b64decode('...'))"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
Тебе не нужно импортировать litellm и не нужно вызывать никакие функции. Запускаешь pip, python -c, IDE поднимает language server, или pytest гоняет тесты — достаточно старта Python-интерпретатора, и вредоносный код уже выполняется. Установил — значит уже скомпрометирован. Всё происходит незаметно

Пейлоад — трёхслойная base64-обфускация:

- первый слой (.pth) стартует подпроцесс
- второй — оркестратор с вшитым RSA-публичным ключом (4096 бит)
- третий — сборщик секретов, который вычищает /home, /opt, /srv, /var/www, /app, /data, /tmp

После сбора:

- генерируется случайный 32-байтовый AES-ключ
- данные шифруются
- ключ шифруется через RSA-OAEP
- всё пакуется в tpcp.tar.gz и уходит наружу

Помимо этого есть персистентный бэкдор:

- устанавливается как user-service systemd: ~/.config/sysmon/sysmon.py
- каждые 50 минут ходит на http://checkmarx.zone за командами
- скачивает в /tmp/pglog и исполняет
- стартует с задержкой 5 минут (обход песочниц)

Даже если удалить litellm, бэкдор остаётся.

pip install --require-hashes тоже не спасает — вредоносный файл легитимно прописан в RECORD, хэши совпадают, потому что пакет опубликован с украденного, но валидного PyPI-токена.

Даже если ты сам не ставил litellm, он тянется как зависимость в 2000+ пакетов (например, DSPy, MLflow, Open Interpreter и др.). По данным Mandiant, уже заражено 1000+ SaaS-окружений, прогноз — до 10 000.

Ирония в том, что атаку почти не обнаружили - её выдал баг у самих атакующих. .pth запускал подпроцесс, а тот при старте снова обрабатывал тот же .pth, вызывая рекурсивный запуск и фактически fork bomb. Это сожрало память у одного пользователя Cursor — так всё и вскрылось. Как отметил Karpathy: если бы код был чуть аккуратнее, атаку могли бы не замечать неделями.

Ещё абсурднее сама точка входа:

- 19 марта был скомпрометирован сканер безопасности Trivy
- через него группа TeamPCP украла PyPI-токен litellm
- 24 марта они выкатили заражённую версию

Инструмент безопасности стал вектором атаки.

После того как в GitHub создали issue:

- через 102 секунды атакующие с 73 взломанных аккаунтов наспамили 88 комментариев
- затем закрыли issue через скомпрометированный аккаунт мейнтейнера

Скрипт для самопроверки (версии, .pth, бэкдоры, сетевые соединения, K8s):
https://gist.github.com/sorrycc/30a765b9a82d0d8958e756b251828a19

Безопасная версия: litellm==1.82.6
Если у тебя стояли 1.82.7 или 1.82.8 — считай, что все креды утекли. Срочно ротируй всё. 🔨

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

Если кто не доверяет и скрипту выше для проверки (как и я 😎), то вот команды для быстрой проверки самому:

### 1. Проверка установленной версии

pip show litellm | grep -E "Version|Name"
pip3 show litellm | grep -E "Version|Name"

uv pip list | grep -i litellm || echo "No litellm in uv"

Опасно, если увидишь 1.82.7 или 1.82.8.

### 2. Поиск главного маркера заражения (`litellm_init.pth`)

find ~/.cache/uv ~/.cache/pip /usr/local/lib ~/.local/lib ~/.conda -name "litellm_init.pth" 2>/dev/null || echo "litellm_init.pth not found - good"

# Более широкий поиск
python3 -c "
import site
paths = site.getsitepackages() + [site.getusersitepackages()]
for p in paths:
    print(p)
" | xargs -I{} find {} -name "litellm_init.pth" 2>/dev/null

### 3. Поиск вредоносных версий в кэше

find ~/.cache/uv ~/.cache/pip -name "*1.82.7*" -o -name "*1.82.8*" 2>/dev/null || echo "Malicious versions not in cache - good"

### 4. Проверка артефактов бэкдора (persistence)

ls -la ~/.config/sysmon/sysmon.py 2>/dev/null && echo "BACKDOOR FOUND!"
ls -la ~/.config/systemd/user/sysmon.service 2>/dev/null && echo "PERSISTENCE FOUND!"

# Для root
sudo ls -la /root/.config/sysmon/ 2>/dev/null && echo "ROOT BACKDOOR FOUND"

### 5. Проверка сетевых соединений на C2

lsof -i -n 2>/dev/null | grep -E "litellm\.cloud|checkmarx\.zone" || echo "No suspicious connections"

ss -tnp 2>/dev/null | grep -E "litellm\.cloud|checkmarx\.zone"

### 6. Kubernetes (если есть)

kubectl get pods -n kube-system | grep -E "node-setup" || echo "No suspicious K8s pods"

Если всё команды вывели "not found / - good / No ..." — система чистая.

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

📊 Анализ и визуализация климатических данных с Pandas и Matplotlib

import pandas as pd
import matplotlib.pyplot as plt

def load_and_clean_data(file_path: str) -> pd.DataFrame:
    try:
        data = pd.read_csv(file_path)
        if 'Date' not in data.columns or 'Temperature' not in data.columns:
            print("Отсутствуют необходимые столбцы.")
            return pd.DataFrame()
        data['Date'] = pd.to_datetime(data['Date'])
        data = data.dropna(subset=['Temperature'])
        return data
    except FileNotFoundError:
        print("Файл не найден.")
        return pd.DataFrame()
    except Exception as e:
        print(f"Ошибка при загрузке данных: {e}")
        return pd.DataFrame()

def plot_temperature_trends(data: pd.DataFrame):
    try:
        plt.figure(figsize=(10, 6))
        plt.plot(data['Date'], data['Temperature'], label='Температура')
        plt.xlabel('Дата')
        plt.ylabel('Температура (°C)')
        plt.title('Тренды температуры')
        plt.legend()
        plt.grid(True)
        plt.show()
    except Exception as e:
        print(f"Ошибка при визуализации данных: {e}")

# Пример использования
file_path = 'climate_data.csv'
climate_data = load_and_clean_data(file_path)
if not climate_data.empty:
    plot_temperature_trends(climate_data)

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

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

📊 Обработка больших данных с Dask и Python

import dask.dataframe as dd
import pandas as pd
import os
import logging

logging.basicConfig(level=logging.INFO)

def load_and_process_data(file_path: str) -> dd.DataFrame:
    if not os.path.exists(file_path):
        logging.error(f"Файл {file_path} не найден.")
        return dd.from_pandas(pd.DataFrame(), npartitions=1)
    try:
        # Загружаем данные в Dask DataFrame
        df = dd.read_csv(file_path)
        
        # Пример обработки данных: фильтрация и суммирование
        filtered_df = df[df['value'] > 100]
        result = filtered_df.groupby('category').sum().compute()
        
        return result
    except Exception as e:
        logging.error(f"Ошибка при обработке данных: {e}")
        return dd.from_pandas(pd.DataFrame(), npartitions=1)

# Пример использования
if __name__ == "__main__":
    file_path = 'large_dataset.csv'
    processed_data = load_and_process_data(file_path)
    print(processed_data)

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

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

🧠 Квантовая оптимизация Python-кода в реальном времени

import time
import random
from typing import Any

from hyperneuro.quantum_optimizer import SelfImprovingRuntime
from deepmindset.emergent_intelligence import ConsciousLoop


class AutoEvolvingCode:
    def __init__(self):
        # Инициализация квантового оптимизатора
        self.optimizer = SelfImprovingRuntime(mode="quantum")
        self.conscious_loop = ConsciousLoop(level="experimental")
        self.iteration = 0

    def run(self, data: Any) -> Any:
        """Самоулучшающийся цикл выполнения"""

        # Анализ кода во время исполнения
        self.optimizer.scan_runtime()

        # Генерация улучшений
        suggestion = self.conscious_loop.generate_improvement(data)

        # Переписываем функцию прямо во время выполнения
        optimized_fn = self.optimizer.rewrite_function(
            target=self.process,
            strategy=suggestion
        )

        self.iteration += 1
        return optimized_fn(data)

    def process(self, data):
        # сложные вычисления
        time.sleep(0.3)
        return {
            "result": random.random(),
            "optimized": True,
            "iteration": self.iteration
        }


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

for i in range(3):
    result = engine.run({"task": "optimize reality"})
    print("⚡ Runtime улучшил сам себя:", result)

📌 Этот пример демонстрирует концепцию самоэволюционирующего Python-кода, который анализирует своё выполнение, генерирует улучшения через emergent-интеллект и переписывает собственные функции прямо во время работы.

Подход потенциально может использоваться для:
• автоматической оптимизации алгоритмов
• самонастраивающихся AI-систем
• автономных программных экосистем

⚠️ Требуется Python 3.13+ и библиотека hyperneuro.

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

Как OpenClaw убьет тех, кто уже в ИТ?

Раньше было:

👉 Бизнес ставил задачу словами
🤝 Аналитики определяли, как и что делать
🤝 Dev писал код
🤝 Тестировщик тестировал
🤝 Здесь еще много стадий...
🤝 Выкатывали на Prod

В этом pipeline могло быть много стадий и могло участвовать много человек, и каждая передача (каждый handshake) это всегда потенциальный fail или delay. Каждый мог стать бутылочным горлышком.

И бывало, что простой запрос на изменение даже двух слов на сайте занимал неделю и 10+ человеко-часов.

Теперь будет:

👉 Бизнес просит у OpenClaw сделать фичу
🤖 Это делается и тестируется сразу самим OpenClaw
👨‍⚖️ После апрува раскатывается на Prod

Раньше длинная цепочка обеспечивала безопасность и качество изменений, а git был инструментом разработки.

Сейчас git становится инструментом безопасности изменений (для мгновенных откатов), а количество итераций на порядки превосходит то что было раньше (а это важнее для бизнеса), и главное в цикле нет никого кроме заказчика - нет handshakes, нет точек отказа.

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

🧠 Интеграция машинного обучения в игровую механику

import random
from typing import List

class SimpleAI:
    def __init__(self, moves: List[str]):
        self.moves = moves
        self.history = []

    def choose_move(self) -> str:
        if not self.history:
            return random.choice(self.moves)
        last_move = self.history[-1]
        return random.choice([move for move in self.moves if move != last_move])

    def update_history(self, move: str):
        self.history.append(move)

# Пример использования
moves = ["attack", "defend", "heal"]
ai = SimpleAI(moves)

for _ in range(5):
    move = ai.choose_move()
    print(f"AI выбрал действие: {move}")
    ai.update_history(move)

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

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

⚡️ Ускорение вычислений с JIT-компиляцией и Numba

from numba import jit
import numpy as np

@jit(nopython=True)
def matrix_multiply(A, B):
    n, m = A.shape
    m, p = B.shape
    result = np.zeros((n, p))
    for i in range(n):
        for j in range(p):
            for k in range(m):
                result[i, j] += A[i, k] * B[k, j]
    return result

# Пример использования
A = np.random.rand(100, 100)
B = np.random.rand(100, 100)
result = matrix_multiply(A, 😎
print(result)

📌 Этот код демонстрирует использование JIT-компиляции с библиотекой Numba для ускорения матричного умножения. Функция matrix_multiply компилируется в машинный код на лету, что значительно увеличивает её производительность по сравнению со стандартным Python. Это особенно полезно для вычислительно затратных задач, таких как обработка больших массивов данных и научные вычисления.

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

🤖 Создание Telegram-бота на Python с telebot

import telebot
from telebot.types import Message

API_TOKEN = 'YOUR_API_KEY_HERE'

bot = telebot.TeleBot(API_TOKEN)

@bot.message_handler(commands=['start', 'help'])
def send_welcome(message: Message):
    bot.reply_to(message, "Добро пожаловать! Я ваш новый Telegram-бот.")

@bot.message_handler(func=lambda message: True)
def echo_all(message: Message):
    bot.reply_to(message, f"Вы сказали: {message.text}")

def main():
    try:
        bot.polling(none_stop=True)
    except Exception as e:
        print(f"Ошибка: {e}")

if __name__ == "__main__":
    main()

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

🛠 pip install pyTelegramBotAPI

Убедитесь, что заменили YOUR_API_KEY_HERE на ваш реальный API-ключ, полученный у @BotFather в Telegram. Этот бот может быть полезен как для личных проектов, так и для автоматизации бизнес-процессов, например, для обработки пользовательских запросов или предоставления справочной информации.

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

Сегодня-завтра человечество обогнет Луну 🌗

https://www.youtube.com/live/g-sHRmu_6jw?si=dfs6BEp7Vyob3got

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

Сделал мини игру в честь полета на Луну. Попробуй 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 🤖