1 硬件组成：

K230+雨量检测传感器+mq135检测传感器+蜂鸣器+led灯

| ​​核心主控​​ K230-CanMV等官方开发板 | 1 |

| ​​传感器模块​​ | ​​雨量检测传感器​​ | ​​数字开关量模块​​：简单检测有无雨滴（输出0/1）。
​​模拟量模块​​：输出模拟电压值，可判断雨量大小（推荐）。 | 1 |

| | ​​空气质量传感器​​ | ​​MQ-135模块​​：已带板载电路，输出模拟量和数字量（TTL电平）。 | 1 |

| ​​执行器（告警）​​ | ​​蜂鸣器​​ | ​​有源蜂鸣器​​：给定电平就响，操作简单。
​​无源蜂鸣器​​：需要 PWM 驱动，可播放不同音调（推荐，更灵活）。 | 1 |

| | ​​LED灯​​ | 普通发光二极管（可多色，如RGB LED） | 至少1个 |

| ​​基础电路​​ | ​​限流电阻​​ | 220Ω - 1kΩ，用于连接LED。 | 若干 |

| | ​​三极管/MOS管​​ | 如S8050，如果驱动功率较大的蜂鸣器（>20mA），需要用它来放大电流。 | 1

| ​​连接件​​ | ​​杜邦线​​ | 公对公、母对母、公对母，用于连接各个模块。 | 1套 |

| ​​电源​​ | ​​电源适配器​​ | 根据K230开发板要求选择，通常是5V/2A或12V/1A。 | 1 |

| | ​​面包板​​ | 用于免焊接搭建电路，方便调试。 | 1 |

2 功能简介：

1. ​​环境空气质量监测​​：通过MQ135传感器检测空气中的有害气体（如CO2、酒精、苯、烟雾等）浓度，评估空气质量。

2. ​​降雨检测​​：通过雨滴/降雨传感器检测是否下雨以及雨量大小。

3. ​​智能告警​​：通过蜂鸣器（声音告警）和LED灯（光告警）组合，在不同条件下触发不同模式的告警。

4. ​​核心处理与智能分析​​：K230作为核心AI芯片，不仅可以读取传感器数据，还能运行更复杂的算法，

   基于yolov8算法实现输电线路6种异常，并转成kpu模型。

3 效果演示：

4 核心代码展示：

from libs.PipeLine import PipeLine
from libs.YOLO import YOLOv8
from libs.Utils import *
import os, sys, gc
import ulab.numpy as np
from machine import Pin, ADC
import time
import image
from media.sensor import *
from media.display import *
from media.media import *


# 图像保存函数
def save_img(img, chn, save_directory, prefix="image"):
    try:
        # 根据图像格式确定文件后缀
        if img.format() == image.YUV420:
            suffix = "yuv420"
        elif img.format() == image.RGB888:
            suffix = "rgb888"
        elif img.format() == image.RGBP888:
            suffix = "rgb888p"
        else:
            suffix = "jpg"  # 默认使用jpg

        # 生成带时间戳的文件名
        timestamp = time.localtime()
        filename = f"{save_directory}/{prefix}_{timestamp[0]:04d}{timestamp[1]:02d}{timestamp[2]:02d}_{timestamp[3]:02d}{timestamp[4]:02d}_{timestamp[5]:02d}_chn{chn}.{suffix}"

        # 保存图像
        success = img.save(filename)
        if success:
            print(f"保存成功: {filename}")
            return True
        else:
            print(f"保存失败: {filename}")
            return False
    except Exception as e:
        print(f"保存异常: {e}")
        return False


# 文本保存函数
def save_text_to_file(text, save_directory, prefix="log"):
    try:

        # 生成带日期的文件名
        timestamp = time.localtime()
        date_str = f"{timestamp[0]:04d}{timestamp[1]:02d}{timestamp[2]:02d}"
        filename = f"{save_directory}/{prefix}_{date_str}.txt"

        # 生成带时间戳的文本行
        time_str = f"{timestamp[3]:02d}:{timestamp[4]:02d}:{timestamp[5]:02d}"
        log_line = f"[{time_str}] {text}\n"

        # 追加写入文件
        with open(filename, 'a') as f:
            f.write(log_line)

        print(f"日志保存: {filename}")
        return True

    except Exception as e:
        print(f"日志保存异常: {e}")
        return False


if __name__ == "__main__":
    # === 传感器初始化 ===
    mq135 = ADC(0)  # 空气质量传感器 (ADC0)
    rain_sensor = ADC(1)  # 雨量传感器 (ADC1)
    buzzer = Pin(20, Pin.OUT)  # 有源蜂鸣器 (GPIO20)
    led = Pin(42, Pin.OUT)  # 使用GPIO42

    # === 文件保存初始化 ===
    image_counter = 0
    text_counter = 0
    save_interval = 30  # 每30帧保存一次
    image_save_directory = "/data/videos"
    text_save_directory = "/data/logs"

    # === 模型和显示初始化 ===
    kmodel_path = "/data/best.kmodel"
    labels = ["Insulator string", "Broken shell", "nest", "Flashover damage shell", "kite", "trash"]
    model_input_size = [320, 320]

    pl = PipeLine(rgb888p_size=[640, 360], display_mode="lcd")
    pl.create()
    yolo = YOLOv8(
        task_type="detect", mode="video",
        kmodel_path=kmodel_path, labels=labels,
        rgb888p_size=[640, 360], model_input_size=model_input_size,
        display_size=pl.get_display_size(), conf_thresh=0.5,
        nms_thresh=0.45, max_boxes_num=50, debug_mode=0
    )
    yolo.config_preprocess()

    # === 雨量校准参数 ===
    DRY_VALUE = 3000
    WET_VALUE = 800

    # === 报警参数 ===
    buzzer_interval = 2000

    # === 主循环 ===
    frame_count = 0
    try:
        while True:
            with ScopedTiming("total", 1):
                # 获取图像帧
                img = pl.get_frame()

                # 目标检测
                res = yolo.run(img)

                # === 图像保存逻辑 ===
                try:
                    # 定期保存
                    if frame_count % save_interval == 0:
                        # 从不同通道获取图像并保存
                        for chn_id in [1]:  # 尝试所有通道
                            try:
                                save_frame = pl.sensor.snapshot(chn=chn_id)
                                if save_frame:
                                    prefix = "periodic"
                                    if res is not None and len(res[0]) > 0:
                                        prefix = "target"
                                    save_img(save_frame, chn_id, image_save_directory, prefix)
                                    break  # 成功保存一个通道后就退出
                            except Exception as e:
                                print(f"通道{chn_id}保存失败: {e}")
                                continue

                    # 检测到目标时立即保存
                    if res is not None and len(res[0]) > 0:
                        for chn_id in [1]:
                            try:
                                save_frame = pl.sensor.snapshot(chn=chn_id)
                                if save_frame:
                                    save_img(save_frame, chn_id, image_save_directory, "target")
                                    break
                            except Exception as e:
                                print(f"目标保存通道{chn_id}失败: {e}")
                                continue

                except Exception as e:
                    print(f"图像保存错误: {e}")

                # === 传感器数据读取 ===
                air_quality = mq135.read_u16()
                if air_quality < 1000:
                    quality = "优"
                    air_alarm = False
                elif air_quality < 2000:
                    quality = "良"
                    air_alarm = False
                else:
                    quality = "污染"
                    air_alarm = True

                rain_value = rain_sensor.read_u16()
                if rain_value < 100:
                    rain_status = "无雨"
                    rain_alarm = False
                else:
                    intensity = (1 - (rain_value - WET_VALUE) / (DRY_VALUE - WET_VALUE)) * 100
                    rain_status = f"有雨 {min(intensity, 100):.1f}%"
                    rain_alarm = intensity > 50

                # === 蜂鸣器控制 ===
                current_time = time.ticks_ms()
                if air_alarm or rain_alarm:
                    if (current_time // buzzer_interval) % 2 == 0:
                        buzzer.value(1)
                        alarm_text = " !报警!"
                    else:

                        buzzer.value(0)
                        alarm_text = " !注意!"
                else:

                    buzzer.value(0)
                    alarm_text = ""
                # 检测结果信息
                detection_info = ""
                if res is not None and len(res[0]) > 0:
                    detected_objects = []
                    for i in range(len(res[0])):
                        label = res[1][i]
                        confidence = res[2][i]
                        detected_objects.append(f"{label}({confidence:.2f})")
                    detection_info = f" | 检测到: {', '.join(detected_objects)}"
                else:
                    detection_info = " | 未检测到目标"

                # === 显示信息 ===
                alarm_text = ""
                text = f"空气质量:{quality} | 雨量:{rain_status}{alarm_text} | {detection_info}"
                print(text)

                # === 文本日志保存 ===
                try:
                    # 定期保存日志（每10帧或检测到目标时）
                    if frame_count % 10 == 0 or (res is not None and len(res[0]) > 0):
                        save_text_to_file(text, text_save_directory, "detection_log")
                        text_counter += 1
                except Exception as e:
                    print(f"日志保存错误: {e}")

                # 显示检测结果
                yolo.draw_result(res, pl.osd_img)

                # LED控制
                if res is not None and len(res[0]) > 0:
                    led.value(1)
                    time.sleep(0.1)
                    led.value(0)
                    time.sleep(0.1)

                # 绘制信息
                pl.osd_img.draw_rectangle(0, 0, 640, 30, color=(0, 0, 0), thickness=-1, alpha=0.7)
                pl.osd_img.draw_string_advanced(10, 5, 30, text, color=(255, 255, 0, 255))

                pl.show_image()

                frame_count += 1
                image_counter += 1

                # 垃圾回收
                if frame_count % 10 == 0:
                    gc.collect()

    except KeyboardInterrupt:
        print("用户停止程序")
    except Exception as e:
        print(f"程序异常: {e}")
    finally:
        # 保存最终状态
        try:
            final_text = f"程序结束 | 总帧数: {frame_count} | 保存图像: {image_counter} | 保存日志: {text_counter}"
            save_text_to_file(final_text, text_save_directory, "system_log")
        except:
            pass

        # 清理资源
        try:
            yolo.deinit()
        except:
            pass

        try:
            pl.destroy()
        except:
            pass

        try:
            # 释放媒体资源
            MediaManager.deinit()
            Display.deinit()
        except:
            pass

        print("程序结束")