基于esp32的视频监控系统

项目简介

使用esp32作为主控，ov2640采集图像，将视频或图像保存至sd卡，也可通过wifi在线实时查看。

项目功能

• 按下boot键打开或关闭补光灯，
• 按下侧键拍照并保存，
• 通过udp向外实时传输视频，通过python上位机查看画面，可作为视频监控。

项目参数

• esp32 S3 N16R2提供wifi功能，超大的falsh即使没安装sd卡，也能存储较多照片
• ov2640 200w像素 24p
• RGB彩色补光灯

硬件说明

• 注意部分esp32S3型号某些引脚用于连接外部falsh不可用作其他用途

软件代码

固件micropython

hardware.py

from machine import Pin,PWM
from machine import Timer
import os
import json
import camera
import socket
import network
import time
class SteeringEngine:
    def __init__(self):
        self.PWN_pin=PWM(8,freq=50,duty_ns=500)#0.5ms-2.5ms
    def setangle(self,angle):
        duty_ns=int(2000000/180*(angle+90)+500000)
        self.PWN_pin.duty_ns(duty_ns)
        print(self.PWN_pin.duty_ns(),1/self.PWN_pin.freq())
class LED:
    def __init__(self):
        self.bottom_pin=Pin(0,Pin.IN,Pin.PULL_UP)
        self.bottom_pin.irq(trigger=Pin.IRQ_FALLING,handler=self.press_down_isr)
        
        self.red_pin=Pin(41,Pin.OUT,value=1)
        self.green_pin=Pin(48,Pin.OUT,value=1)
        self.blue_pin=Pin(42,Pin.OUT,value=1)
        
        self.state = 0
    def press_down_isr(self,pin):
        self.timer=Timer(-1,mode=Timer.ONE_SHOT,period =10,callback= self.button_check_isr)
    def button_check_isr(self,t):
        if self.bottom_pin.value()==0:
            if self.state==0 :
                self.red_pin.value(0)
                self.green_pin.value(0)
                self.blue_pin.value(0)
                self.state = 1
            else:
                self.red_pin.value(1)
                self.green_pin.value(1)
                self.blue_pin.value(1)
                self.state = 0
class WIFI:
    def __init__(self,WLAN_name='',password='Q'):
        self.ssid=WLAN_name
        self.key=password
        self.sta_if = network.WLAN(network.STA_IF)
    def connect(self):
        if not self.sta_if.isconnected():
            start=time.ticks_ms()
            print('connecting to network...')
            self.sta_if.active(True)
            self.sta_if.connect(self.ssid, self.key)
            while not self.sta_if.isconnected():
                now=time.ticks_ms()
                if time.ticks_diff(now, start)>10000:
                    print('connect  WiFi error')
                    
        print('connect  WiFi ok')
        print(self.sta_if.ifconfig()[0])
    def disconnect(self):
        self.sta_if.disconnect()
        print('disconnect  WiFi ok')
class Camera:
    def __init__(self):
        try:
            camera.init(0, format=camera.JPEG)
        except Exception as e:
            camera.deinit()
            camera.init(0, format=camera.JPEG)

        camera.flip(1)
        camera.mirror(1)
        camera.framesize(camera.FRAME_HVGA)
        camera.speffect(camera.EFFECT_NONE)
        camera.saturation(0)
        camera.brightness(0)
        camera.contrast(0)
        camera.quality(10)
        #10-63数字越小质量越高
    def get(self):
        return camera.capture()
    def retry(self):
        try:
            camera.deinit()
            camera.init(0, format=camera.JPEG)
        except :
            camera.deinit()
            camera.init(0, format=camera.JPEG)
    def deinit(self):
        camera.deinit()
class UDPsever:
    def __init__(self,ip="ipconfig查看电脑ip",port=9090):
        self.s = socket.socket(socket.AF_INET,socket.SOCK_DGRAM,0)
        self.ip=ip
        self.port=port
    def send(self,data):
        self.s.sendto(data, (self.ip, self.port))
class Shutter:
    def __init__(self,func=lambda : print('')):
        self.bottom_pin=Pin(15,Pin.IN,Pin.PULL_DOWN)
        self.bottom_pin.irq(trigger=Pin.IRQ_RISING  ,handler=self.press_down_isr)
        self.func=func
    def press_down_isr(self,pin):
        self.timer=Timer(-1,mode=Timer.ONE_SHOT,period =10,callback= self.button_check_isr)
    def button_check_isr(self,t):
        if self.bottom_pin.value()==0:
            self.func()
class Hotspot:
    def __init__(self):

        self.ap_if = network.WLAN(network.AP_IF)
        
    def on(self):
        self.ap_if.active(True)
        self.ap_if.config(ssid="night-light")
        print(self.ap_if.ifconfig()[0])
    def off(self):
        self.ap_if.active(False)
class Config:
    def __init__(self):
        
        if 'config.json' not in os.listdir():
            data={
                'img_index' : 0,
                  'data':'123'
                }
            with open('config.json', 'wb') as file:
                json.dump(data, file)
        with open('config.json', 'rb') as file:
             self.config_dict = json.load(file)
    def save(self):
        with open('config.json', 'wb') as file:
            json.dump(self.config_dict, file)
    def __getitem__(self, key):
        return self.config_dict[key]
    def __setitem__(self, key, value):
        self.config_dict[key] = value
        self.save()
if __name__=='__main__':
    pass
    

• 注意将自己家wifi名称和密码填入
• 通过ipconfig查看自己电脑ip地址

main.py

from hardware import WIFI,LED,UDPsever,Camera,Config,Shutter,SteeringEngine
import time
import machine
import gc
machine.freq(240000000)
config=Config()
LED()
wifi=WIFI()
wifi.connect()
udp=UDPsever()
camera=Camera()
def take_pic():
    file_name=f'img_{config['img_index']+1}.png'
    
    try:
        with open(file_name, "wb") as f:
            f.write(camera.get())
    except Exception as e:
        print(e)
    else:
        config['img_index'] += 1
        print('拍一张')
    
Shutter(func=take_pic)
s=SteeringEngine()
s.setangle(-90)
while True:
#     try:
        udp.send(camera.get())
        time.sleep(0.2)
        print(gc.mem_free()/1024)
#     except Exception as e:
#         print('err:',e)
#         gc.collect() 
#         camera.retry()
#     finally :
#         camera.deinit()
    
    

上位机代码

import socket
import cv2
import io
from PIL import Image
import numpy as np

s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, 0)
s.bind(("0.0.0.0", 9090))
while True:
    data, IP = s.recvfrom(100000)
    bytes_stream = io.BytesIO(data)
    image = Image.open(bytes_stream)
    img = np.asarray(image)
    img = cv2.cvtColor(img, cv2.COLOR_RGB2BGR)  
    cv2.imshow("ESP32 Capture Image", img)
    if cv2.waitKey(1) == ord("q"):
        break

实物图