Использование Apache Storm для обработки потоков данных с Python.

Использование Apache Storm для обработки потоков данных с Python.

🔄 Обработка потоков данных с Python и Apache Storm: первая встреча

Всем привет! Сегодня поговорим о чем-то по-настоящему живом — потоковых данных. Представьте Twitter-ленту, клики пользователей на сайте или датчики в IoT — всё это нескончаемые потоки событий. Для их реального времени обработки часто используют Apache Storm. Он мощный, масштабируемый и работает под лозунгом: “боевой зубастый молниеносный фермер потоков”.

Но у Storm есть коварная особенность — родной язык для написания логики обработки данных в нём это Java. К счастью, есть библиотека streamparse — она позволяет писать topologies на Python и запускать их в Apache Storm. Сегодня покажу, как это работает.

📦 Установка

Для начала убедитесь, что у вас установлен Apache Storm (версии 1.x). Далее установим streamparse:

pip install streamparse

Создаём проект:

sparse quickstart storm_example
cd storm_example

Внутри проекта — готовая структура: папки для Spout и Bolt’ов, настройки, зависимости.

🌪 Что такое Spout и Bolt?

Spout — источник данных. Получает события "снаружи": из файла, очереди или API.

Bolt — обрабатывает данные. Например, парсит JSON, считает статистику, отправляет в БД.

⚡ Пример: считаем слова в потоке

Создадим Spout, который выдаёт предложения:

from streamparse import Spout
import random

class SentenceSpout(Spout):
    def initialize(self, stormconf, context):
        self.sentences = ["hello world", "python storm integration", "stream processing rocks"]

    def next_tuple(self):
        sentence = random.choice(self.sentences)
        self.emit([sentence])

Теперь Bolt, который разбивает предложения на слова:

from streamparse import Bolt

class SplitSentenceBolt(Bolt):
    def process(self, tup):
        sentence = tup.values[0]
        for word in sentence.split():
            self.emit([word])

И ещё один Bolt, который считает слова:

from streamparse import Bolt
from collections import defaultdict

class WordCountBolt(Bolt):
    def initialize(self, conf, ctx):
        self.counts = defaultdict(int)

    def process(self, tup):
        word = tup.values[0]
        self.counts[word] += 1
        self.log(f"{word}: {self.counts[word]}")

🛠 Запускаем локально

Добавим в файл topologies/wordcount.py описание топологии:

from streamparse import Grouping, Topology
from bolts.word_count import WordCountBolt
from bolts.split_sentence import SplitSentenceBolt
from spouts.sentence import SentenceSpout

class WordCountTopology(Topology):
    sentence_spout = SentenceSpout.spec()
    splitter_bolt = SplitSentenceBolt.spec(inputs=[sentence_spout])
    counter_bolt = WordCountBolt.spec(inputs=[splitter_bolt], par=2)

Теперь можно запустить локально:

sparse run

Вы увидите, как Apache Storm обрабатывает поток: предложения → слова → счётчики.

🎯 Зачем это нужно?

Apache Storm — не просто блажь ради распределённости. Он умеет:

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

А с помощью streamparse вы сохраняете привычный Python, не углубляясь в джунгли Java.

📌 Вывод

Apache Storm и streamparse позволяют собирать мощные системы потоковой обработки на Python. Это отличный шаг, если вы работаете с реальным временем: логи, метрики, события. Да, он требует установки Java и самого Storm, но в обмен вы получаете надёжный и быстрый фреймворк.

Следующий раз разберём, как подключать Kafka и обрабатывать поток реальных сообщений. А пока — пусть молнии потекут по Python-контактам! ⚡🐍

— Иван.

Создание виртуальных агентов с использованием библиотеки SIMBA.

Привет, друзья! Сегодня копнем чуть глубже в захватывающий мир Python и искусственного интеллекта. Рассмотрим, как создавать виртуальных агентов с помощью необычной, но очень мощной библиотеки — SIMBA.

SIMBA (Simple Intelligent Multiagent-Based Architecture) — это легковесный инструмент на Python для моделирования поведения агентов в симулированной среде. Проще говоря, он позволяет вам создавать умных ботов, которые могут принимать решения, взаимодействовать с окружением и друг с другом.

Что особенно круто — библиотека проста в освоении, но при этом дает возможность строить сложные сценарии поведения, как в видеоиграх или симуляциях.

Начнем с простого примера: допустим, у нас есть мир, где несколько агентов ищут еду. Каждый агент может видеть окружение, двигаться и принимать решения.

Установка SIMBA:

pip install simba-agents

Теперь создаем базового агента:

from simba.agents import BaseAgent

class ForagerAgent(BaseAgent):
    def __init__(self, name):
        super().__init__(name)
        self.energy = 10

    def decide(self, perception):
        if 'food' in perception:
            return 'eat'
        return 'move_randomly'

    def act(self, action):
        if action == 'eat':
            self.energy += 5
        elif action == 'move_randomly':
            self.energy -= 1

В этом примере у каждого агента есть "энергия". Он смотрит на мир, решает, что делать (если рядом еда — ест, иначе — двигается случайно), и выполняет действие.

Теперь создаем простую среду:

from simba.environment import GridWorld

env = GridWorld(width=10, height=10)
agent = ForagerAgent("Alex")
env.add_agent(agent, position=(5, 5))
env.add_object('food', position=(6, 5))

Запускаем симуляцию:

for _ in range(10):
    perception = env.get_perception(agent)
    action = agent.decide(perception)
    agent.act(action)
    env.step()

Вуаля! У нас есть первый виртуальный агент. Конечно, это ещё не уровень GTA или The Sims, но вы уже можете создавать более сложные поведения: задания, эмоции, даже память!

SIMBA поддерживает и более продвинутые вещи — коммуникацию между агентами, настройку целей и ограничений, сценарии с многими участниками. Всё это можно комбинировать, чтобы строить обучающиеся модели, игровые NPC, симуляции толпы и многое другое.

Почему стоит попробовать?

- Прост в использовании: код читаемый, API понятный.
- Расширяемость: легко создавать свои типы агентов и объектов.
- Визуализация: можно подключать простую графику для отслеживания симуляции.

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

В следующий раз придумаем что-то более масштабное — например, колонию роботов на Марсе 😉

На этом всё, с вами был Иван. Программируйте интересно!

Интеграция инженерных моделей в Python-приложения.

Интеграция инженерных моделей в Python-приложения: от сырой формулы до готового сервиса

Когда инженеры говорят про "модель", они чаще всего имеют в виду набор уравнений или алгоритм, описывающий поведение системы. Численное моделирование, теплообмен, механика, электрика, химия — практически любая физическая дисциплина опирается на такие модели. А если нужно сделать из них что-то полезное? Например, веб-сервис расчёта теплообмена для проектировщиков? Здесь в игру вступает Python.

Основной плюс Python в этой задаче — богатство библиотек. Инженеры описывают модель на бумаге, а Python даёт возможность быстро превратить её в работающий код. Сегодня мы посмотрим, как можно встроить инженерную модель в реальное приложение — с минимальными усилиями и максимальной пользой.

Допустим, у нас есть простая модель расчёта теплопередачи через стену:

q = U A (Tinside - Toutside)

Где:
- U — коэффициент теплопередачи (Вт/м²·K),
- A — площадь стены (м²),
- Tinside и Toutside — температуры внутри и снаружи (°C).

Первым делом оформим это в отдельную функцию:

def heat_transfer(U, A, T_inside, T_outside):
    return U * A * (T_inside - T_outside)

Теперь представим, что мы хотим предоставить к этой модели простой API-доступ — например, чтобы инженер мог подставить свои значения и получить результат через браузер. Для этого воспользуемся FastAPI:

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class HeatRequest(BaseModel):
    U: float
    A: float
    T_inside: float
    T_outside: float

@app.post("/calculate_heat/")
def calculate_heat(data: HeatRequest):
    q = heat_transfer(data.U, data.A, data.T_inside, data.T_outside)
    return {"heat_loss": q}

Запускается это всё одной командой:

uvicorn main:app --reload

И вот у нас есть полноценный API, который принимает параметры инженерной модели и возвращает результат.

Далее по вкусу:
- можно добавить сохранение в базе (SQLAlchemy),
- логгирование и валидацию (Pydantic, logging),
- построение графиков (matplotlib, plotly)
- даже экспорт в Excel (openpyxl)

Если модель сложнее и реализована в специализированных пакетах вроде SciPy, pyomo (оптимизационные задачи) или sympy (символьная математика) — всё равно можно завернуть её в API на FastAPI, Flask или даже прямую CLI-утилиту.

Python, по сути, превращает инженерные наброски в продукт: автоматизированный инструмент, доступный из любого интерфейса — от командной строки до мобильного приложения.

Заключение? Простая инженерная формула может стать основой для серьёзного приложения, если завернуть её во что-то, что удобно использовать. И Python — прекрасный инструмент для такого превращения.

Настройка и использование приложений для анализа медицинских изображений.

🔥 Медицинские изображения и Python: сила технологий в ваших руках

Привет! Это Иван. Сегодня мы перенесемся в мир медицины — не для постановки диагноза, а чтобы посмотреть, как Python помогает врачам и исследователям анализировать медицинские снимки. Нет, это не искусственный интеллект, заменяющий докторов, а реальные инструменты, которые извлекают информацию из снимков МРТ, КТ и рентгена — быстро и эффективно.

Давайте разберем, как можно построить базовое приложение для анализа медицинских изображений с помощью Python. Начнем с библиотек, которые рекомендую для этого:

- pydicom — работа с DICOM-файлами (медицинский стандарт хранения изображений),
- matplotlib и opencv-python — визуализация и первичная обработка,
- numpy — численные операции,
- scikit-image — продвинутые методы обработки изображений.

📦 Установка:

pip install pydicom opencv-python matplotlib numpy scikit-image

👌 Взгляд на DICOM

Медицинские изображения обычно хранятся в формате DICOM. Давайте откроем одно такое изображение и визуализируем его.

import pydicom
import matplotlib.pyplot as plt

dicom_file = pydicom.dcmread("example.dcm")
image_data = dicom_file.pixel_array

plt.imshow(image_data, cmap="gray")
plt.title("Original DICOM Image")
plt.axis("off")
plt.show()

Просто и понятно. Выгружаем пиксельные данные и визуализируем. Но это только начало.

✂️ Немного обработки

Теперь применим фильтр контуров, чтобы выделить аномалии или границы органов.

import cv2
import numpy as np

# Преобразуем данные в формат uint8 (от 0 до 255)
image_uint8 = cv2.convertScaleAbs(image_data)

# Применяем оператор Canny
edges = cv2.Canny(image_uint8, threshold1=50, threshold2=150)

plt.imshow(edges, cmap='gray')
plt.title("Edges Detected")
plt.axis("off")
plt.show()

Такой простой инструмент может быстро подсветить интересующие области радиологу или разработчику ИИ.

🔬 Сегментация

Предположим, нужно выделить определённую область — например, легкие на снимке грудной клетки. Хотите магии? Давайте поработаем с порогами.

from skimage import filters

threshold = filters.threshold_otsu(image_uint8)
binary_mask = image_uint8 > threshold

plt.imshow(binary_mask, cmap='gray')
plt.title("Otsu Segmentation")
plt.axis("off")
plt.show()

Алгоритм Отсу — это способ найти “разумный” порог разделения. В результате получаем бинарное изображение, где нужная область выделяется автоматически.

🙌 Что дальше?

Дальше можно добавлять нейросеть, делать 3D-реконструкции, сохранять результаты в виде отчётов... Но самое важное: инструменты уже в ваших руках. Даже базовая визуализация может помочь исследователю сократить время анализа, а врачу — сфокусироваться на главном.

Не бойтесь сложных тем — с Python всё становится доступным. Увидимся в следующем посте ✌️

— Иван, программист.

Создание приложений управления освещением с использованием Raspberry Pi и Python.

Привет! Сегодня в нашем блоге — прикладная магия Python, а именно: управление освещением с помощью Raspberry Pi. Это не просто теоретическое упражнение, это настоящий шаг в мир умного дома. Готовы превратить свою лампочку в IoT-устройство? Погнали!

🛠 Что понадобится?

- Raspberry Pi (подойдет даже Zero W)
- Светодиод или реле (если хотите управлять настоящими лампами)
- Резистор на 220 Ом (если светодиод)
- Несколько проводов и макетная плата
- Немного терпения и Python

Подключим светодиод к GPIO-контакту, например, к пину 17. Не забудьте резистор между GPIO и анодом диода, чтобы не сжечь вывод.

Теперь к коду.

📦 Используем библиотеку gpiozero — она значительно упрощает работу с GPIO.

Простейшая версия программы для включения лампочки выглядит так:

from gpiozero import LED
from time import sleep

led = LED(17)

while True:
    led.on()
    sleep(1)
    led.off()
    sleep(1)

Это базовая моргалка, идеальная для проверки подключения.

⚡ А теперь сделаем чуть умнее: управление светом по времени суток. Например, включать освещение после заката.

Для этого используем библиотеку astral, которая дает информацию о времени восхода и заката солнца по геолокации.

Установка:

pip install astral

Пример:

from gpiozero import LED
from time import sleep
from datetime import datetime
from astral import LocationInfo
from astral.sun import sun

led = LED(17)
city = LocationInfo("Moscow", "Russia", "Europe/Moscow", 55.7558, 37.6173)

def is_dark():
    s = sun(city.observer, date=datetime.now().date())
    now = datetime.now(city.timezone)
    return now < s['sunrise'] or now > s['sunset']

while True:
    if is_dark():
        led.on()
    else:
        led.off()
    sleep(60)

Теперь лампа сама включается вечером и выключается утром — как у настоящих хайтек-грибов 🍄

🎛 Хотите больше интерактива? Подключите веб-интерфейс или контролируйте свет через Telegram-бота. А можно добавить датчик движения — и получить "умный" ночник.

💡 В чем польза?

- Изучаете работу с GPIO
- Используете сторонние модули Python (gpiozero, astral)
- Получаете практическое понимание реального применения Python в IoT

На этом всё! Пусть ваш код будет светлым — во всех смыслах этого слова :)

Как использовать Ansible и Python для автоматизации системных задач.

Как использовать Ansible и Python для автоматизации системных задач.

Привет! Сегодня копнем в сторону автоматики — поговорим о том, как использовать Ansible и Python для автоматизации системных задач. Ansible сам по себе мощный инструмент, но иногда хочется чуть больше гибкости — тут на помощь и приходит Python. Совместно они превращаются в универсальный швейцарский нож DevOps'а.

🎯 Зачем это вообще нужно?

Представим типичный сценарий: у вас есть десяток серверов, на которых нужно установить nginx, обновить пакеты, разложить конфиги. Можно, конечно, вручную, но это скучно и долго. Ansible позволяет описать всё это в виде playbook'ов — вы просто нажимаете "старт", и магия происходит сама. Но если нужна логика посложнее — Python готов помочь.

📦 Установка Ansible и библиотек

Начнем с подготовки:

pip install ansible
pip install ansible-runner

ansible-runner — это модуль, который позволяет управлять Ansible-скриптами напрямую из Python.

🛠 Простой playbook

Давайте создадим playbook, который обновляет систему и устанавливает htop.

Файл: update_install.yml

- hosts: all
  become: true
  tasks:
    - name: Update apt cache
      apt:
        update_cache: yes

    - name: Install htop
      apt:
        name: htop
        state: present

⚙ Запуск playbook из Python

Теперь напишем скрипт, который запустит этот playbook прямо из Python.

import ansible_runner

def run_playbook():
    result = ansible_runner.run(private_data_dir='.', playbook='update_install.yml')
    print("Status:", result.status)
    print("RC:", result.rc)

    for e in result.events:
        print(e.get('stdout', ''))

run_playbook()

Важно, чтобы рядом лежал файл inventory с описанием хостов.

Файл: inventory

[servers]
192.168.1.10 ansible_user=ubuntu ansible_ssh_private_key_file=~/.ssh/id_rsa

🔁 Генерация динамического инвентори на Python

Вместо статического файла можно сгенерировать список хостов прямо из Python:

import json

inventory = {
    "all": {
        "hosts": ["192.168.1.10", "192.168.1.11"],
        "vars": {
            "ansible_user": "ubuntu",
            "ansible_ssh_private_key_file": "~/.ssh/id_rsa"
        }
    }
}

with open("inventory.json", "w") as f:
    json.dump(inventory, f)

Запуск с таким инвентори:

result = ansible_runner.run(
    private_data_dir='.',
    playbook='update_install.yml',
    inventory='inventory.json'
)

🚀 Бонус: передача переменных из Python

Нужно передать параметры в playbook? Не проблема:

extra_vars = {
    "package_name": "nginx"
}

ansible_runner.run(
    private_data_dir='.',
    playbook='install_pkg.yml',
    extravars=extra_vars
)

А в самом playbook:

- name: Install package
  hosts: all
  become: true
  tasks:
    - name: Install {{ package_name }}
      apt:
        name: "{{ package_name }}"
        state: present

🔥 Вывод

Вместе Ansible и Python открывают ворота в мир мощной и гибкой автоматизации. Хотите запустить плейбук после триггера в Django или по расписанию через Celery? Просто запускайте нужный скрипт. Это отличная пара для тех, кто предпочитает программировать инфраструктуру, а не кликать по консоли.

С вами был Иван. Пиши код и автоматизируй всё вокруг!

Работа с протоколом SNMP для управления сетевыми устройствами.

Привет! Сегодня мы окунемся в мир сетей — разберемся, как с помощью Python управлять сетевыми устройствами через протокол SNMP. Если тебе кажется, что это сложно — спешу успокоить: всё на удивление просто, особенно если использовать правильные библиотеки.

SNMP (Simple Network Management Protocol) — это протокол, который позволяет считывать информацию о состоянии устройств в сети и даже управлять ими. Чаще всего его используют для мониторинга роутеров, свитчей, принтеров и серверов.

Для работы с SNMP в Python существует отличная библиотека — pysnmp. Она поддерживает все версии протокола и предоставляет простой интерфейс для запросов.

Установим её:

pip install pysnmp

Разберем простой пример: считаем системное имя (sysName) устройства.

from pysnmp.hlapi import *

iterator = getCmd(
    SnmpEngine(),
    CommunityData('public', mpModel=0),  # SNMPv1/v2c
    UdpTransportTarget(('192.168.1.1', 161)),
    ContextData(),
    ObjectType(ObjectIdentity('SNMPv2-MIB', 'sysName', 0))
)

errorIndication, errorStatus, errorIndex, varBinds = next(iterator)

if errorIndication:
    print(f"Error: {errorIndication}")
elif errorStatus:
    print(f"{errorStatus.prettyPrint()} at {errorIndex}")
else:
    for varBind in varBinds:
        print(f"{varBind[0].prettyPrint()} = {varBind[1].prettyPrint()}")

В этом скрипте мы запрашиваем объект sysName.0, который представляет имя хоста. Обрати внимание: указывается IP-адрес устройства и SNMP-комьюнити. Здесь мы используем community public, распространённое значение по умолчанию — но лучше использовать свою для безопасности.

Теперь пример посложнее: соберем список интерфейсов.

from pysnmp.hlapi import *

for (errorIndication,
     errorStatus,
     errorIndex,
     varBinds) in nextCmd(SnmpEngine(),
                          CommunityData('public'),
                          UdpTransportTarget(('192.168.1.1', 161)),
                          ContextData(),
                          ObjectType(ObjectIdentity('IF-MIB', 'ifDescr')),
                          lexicographicMode=False):
    if errorIndication:
        print(f"Error: {errorIndication}")
        break
    elif errorStatus:
        print(f"{errorStatus.prettyPrint()} at {errorIndex}")
        break
    else:
        for varBind in varBinds:
            print(f"{varBind[0]} = {varBind[1]}")

Функция nextCmd выполняет последовательный опрос — это удобно, когда нужно пройти по дереву OID'ов, например, чтобы получить список интерфейсов.

А теперь — изюминка! Через SNMP можно не только читать данные, но и изменять параметры. Например, выключить интерфейс. Но для этого нужно использовать права на запись, а значит подбирать правильное community и понимать последствия.

Вот пример с setCmd:

from pysnmp.hlapi import *

errorIndication, errorStatus, errorIndex, varBinds = next(
    setCmd(SnmpEngine(),
           CommunityData('private'),  # community с правами записи
           UdpTransportTarget(('192.168.1.1', 161)),
           ContextData(),
           ObjectType(ObjectIdentity('IF-MIB', 'ifAdminStatus', 2), Integer(2)))  # 2 = down
)

if errorIndication:
    print(f"Error: {errorIndication}")
elif errorStatus:
    print(f"{errorStatus.prettyPrint()} at {errorIndex}")
else:
    print("Interface disabled successfully")

Осторожно: этим можно на самом деле отключить интерфейс маршрутизатора! Так что всегда проверяй документацию и тестируй в безопасной среде.

На этом всё. SNMP — мощный инструмент в арсенале Python-программиста, особенно если ты работаешь с сетевым оборудованием. Надеюсь, теперь ты смотришь на сетевые протоколы без страха — скорее с азартом исследователя.

Увидимся в следующих постах!
— Иван

Работа с текстом: использование модуля textwrap для форматирования строк

📦 Python для начинающих: форматируем текст как профи с помощью textwrap

Привет! С вами Иван, и сегодня я расскажу про один из тех модулей стандартной библиотеки Python, который редко упоминают, но который может прилично упростить жизнь — особенно, если вы работаете с текстами. Речь пойдет о модуле textwrap. Он помогает красиво разносить текст по строкам, контролировать ширину текста, оформлять вывод — всего лишь парой строк кода.

Представьте, что вы делаете консольный словарь, новостной бот или пишете отчет в текстовый файл. Текст должен хорошо читаться, особенно в терминале. Вот тут в игру вступает textwrap.

Начнем с простого:

import textwrap

sample_text = "Python is a powerful high-level programming language known for its readability and versatility."

wrapped = textwrap.fill(sample_text, width=40)
print(wrapped)

Результат будет такой:

Python is a powerful high-level
programming language known for its
readability and versatility.

Функция fill возвращает строку, где длинный текст красиво "перенесен" на несколько строк, каждая — не длиннее 40 символов. Если больше — строки будут обрезаться и переноситься.

Теперь представим, что вы хотите получить список строк, чтобы, например, написать их в файл построчно. Тогда подойдет метод wrap:

lines = textwrap.wrap(sample_text, width=30)
for line in lines:
    print(line)

Результат:

Python is a powerful high-
level programming language
known for its readability and
versatility.

Но это еще не всё! Бывали ли у вас случаи, когда нужно было красиво оформить блок текста с отступом или добавить префикс к каждой строке, например, "> " для цитаты? Вот как это делается:

quote = textwrap.fill(sample_text, width=50, initial_indent="> ", subsequent_indent="> ")
print(quote)

Получим:

> Python is a powerful high-level programming
> language known for its readability and
> versatility.

А ещё есть текстовая "обрезка". Не всегда хочется показывать пользователю весь текст. Иногда достаточно одного абзаца. Вот как можно аккуратно "усечь" строку до нужного количества символов с добавлением троеточия:

short = textwrap.shorten(sample_text, width=50, placeholder="...")
print(short)

Вывод:

Python is a powerful high-level programming...

textwrap — это простой, но мощный способ сделать текст более читаемым и привлекательным прямо из консоли. Особенно он полезен, если вы оформляете сообщения, вывод или простые отчеты в текстовые файлы. Не стоит недооценивать, насколько визуальная чистота и читаемость важны даже в командной строке.

На этом всё. Форматируйте с умом — и пиши красиво, даже в консоли! 👨‍💻

– Иван.

Как извлекать метаданные из изображений с помощью PIL

Как извлекать метаданные из изображений с помощью PIL