iTranslated by AI
The content below is an AI-generated translation. This is an experimental feature, and may contain errors. View original article
🐈
Building a Cat Monitoring System: Part 5 - Integrating SwitchBot Control and Camera Viewer with tkinter
What I achieved in this article
- Added the following functions to the SwitchBot control GUI created in Part 4.
- Tapo C210 camera viewer
- Pan/Tilt/Zoom functionality
- Privacy mode ON/OFF functionality
- Still image saving functionality
- Video saving functionality (with audio)
- Original motion detection functionality (simple version)
- Display of the last detected motion time
- Volume bar
- Log recording
Remaining Tasks
- Privacy mode related ※Depreciated in Tapo FW 1.5.x?
- Pan/Tilt related ※Depreciated in Tapo FW 1.5.x?
- Improvement of motion detection performance (targeting only cats)
Environment
As of January 30, 2026, it seems that Tapo FW 1.5.x? is no longer compatible with ONVIF PTZ and pytapo, making it impossible to perform PTZ operations or toggle privacy mode.
Updating Python or the pytapo library to the latest versions has not resolved the issue.
- Windows 11
- SwitchBot App Version: 7.3
- SwitchBot Hub mini Firmware Version: V6.1-5.4
- Tapo Firmware: 1.5.1
- Python 3.13.11 (64bit)
- Refer to the following for Python libraries:
pip install ffmpeg-python==0.2.0
pip install opencv-python==4.13.0.90
pip install plyer==2.1.0
pip install pygame==2.6.1
pip install pytapo==3.3.56
pip install requests==2.32.5
pip install rtsp==1.1.12
Important Notes
Since it is highly likely that existing code will become unusable due to FW updates, it is recommended to turn OFF Tapo app updates and camera FW updates if you want to operate the Tapo app from a PC.
Completed Screen
I was able to successfully obtain both video and still images!
Directory Structure
├── SmartHome.py # GUI main (Tkinter)
├── util/
│ ├── __init__.py # Empty file to make the directory recognized as a package
│ ├── motion_detector.py # Motion detection (simple version)
│ ├── onvif.py # ONVIF PTZ
│ ├── sb_command.py # SwitchBot operation commands
│ ├── switchbot.py # For SwitchBot operation
│ └── recorder.py # Recording and audio capturing
├── icon/
│ └── Switchbot.png # GUI icon
└── sound/
└── shutter.mp3 # Sound for still image capture
Source Code
GUI Main
unit/SmartHome.py
import base64
import datetime
import json
import os
import subprocess
import sys
import time
import threading
import tkinter as tk
import tkinter.ttk as ttk
from tkinter import filedialog, font, scrolledtext
import urllib.request
import cv2
import numpy as np
from PIL import Image, ImageTk
from plyer import notification
#from pytapo import Tapo
import requests
from requests.exceptions import HTTPError, RequestException, Timeout
import rtsp
from util.motion_detector import MotionDetector
#from util.onvif import Onvif
import util.sb_command as sb_cmd
from util.switchbot import SwitchBotController
from util.recoder import Recorder
#==================================
# Connection settings with SwitchBot Hub mini
#==================================
# How to obtain the token
# (1) Launch the SwitchBot App.
# (2) Open "Settings" in "Profile"
# (3) Tap the "App Version" section 10 times.
# (4) "Developer Options" will appear, so open it.
# (5) Obtain the token.
TOKEN = ''
API_HOST = 'https://api.switch-bot.com'
DEBIVELIST_URL = f'{API_HOST}/v1.0/devices'
HEADERS = {
'Authorization': TOKEN,
'Content-Type': 'application/json; charset=utf8'
}
#==================================
# Tapo C210
#==================================
CAM_USER = "" # Adjust to your environment
CAM_PASS = "" # Adjust to your environment. Use the camera's own password, not the Tapo app password
CAM_ADDR = "" # Adjust to your environment.
CAM_PORT = "554"
CAM_STREAM = "stream1" # stream1 is 1080p(1920x1080) + with audio, stream2 is 360p(640x360) without audio
CAM_ONVIF_PORT = "2020"
# Pan/Tilt parameters
X_STP = 0.1
Y_STP = 0.1
# Default position moved to by the camera's center button. Range -1 to 1
X_DEFALUT = -0.4
Y_DEFALUT = 0.2
#==================================
# UI Settings
#==================================
icon_path = os.path.join('icon', 'Switchbot.png')
# Font (Using DSEG v046)
# https://www.keshikan.net/fonts.html
FONT_DIG = 'DSEG14 Classic Mini'
# Resolution when obtaining images from the camera.
if CAM_STREAM == "stream1":
FRAME_WIDTH = 1920
FRAME_HEIGHT = 1080
else:
FRAME_WIDTH = 640
FRAME_HEIGHT = 360
# Canvas size
CANVAS_WIDTH = 640
CANVAS_HEIGHT = 360
#==================================
# Others
#==================================
# Folder name for saving video and still images
VIDEO_DIR = 'Video'
# Volume meter offset obtained from calibration
# If sound level meter app Decibel X is 28.5dB, and Tapo is -61.9dBFS
# 28.5 - (-61.9) = 90.4
DB_OFFSET = 90.4
# Size of 1 PCM frame. For 0.1 seconds (16000Hz * 0.1sec)
CHUNK = 1600
class Application(tk.Frame):
def __init__(self, master=None):
super().__init__(master)
# SwitchBot
self.sb = SwitchBotController(
TOKEN,
API_HOST,
logger=self._logger
)
self.temperature, self.humidity = self.sb.get_meter(sb_cmd.DIC_DIV_ID.get('Meter'))
# Storage for still images and videos
if not os.path.exists(VIDEO_DIR):
os.makedirs(VIDEO_DIR)
# Recording/audio related
self.capture_flag = False
self.record_flag = False
self.rec_start_time = 0.0
self.writer = None
self.audio_fp = None
self.video_file = None
self.audio_file = None
self.recorder = Recorder(
video_dir=VIDEO_DIR,
canvas_size=(FRAME_WIDTH, FRAME_HEIGHT),
logger=self._logger
)
# Motion detection
self.motion = MotionDetector()
self.last_detected_time = "--:--"
self.label_detected_time = None
# Control Tapo C210 via ONVIF PTZ
# Tapo FW 1.5.1 returns "HTTP Error 400: Bad Request"
"""
self.onvif = Onvif(
cam_addr=CAM_ADDR,
cam_port=CAM_ONVIF_PORT,
user=CAM_USER,
password=CAM_PASS,
logger=self._logger
)
"""
#####################################################
# Open the camera
#####################################################
self.rtsp_url = f"rtsp://{CAM_USER}:{CAM_PASS}@{CAM_ADDR}:{CAM_PORT}/{CAM_STREAM}"
# For video acquisition
try:
self.rstp = subprocess.Popen(
[
"ffmpeg",
"-rtsp_transport", "tcp",
"-i", self.rtsp_url,
"-an", # No audio
"-vf", f"scale={FRAME_WIDTH}:{FRAME_HEIGHT}", # Fix output size
"-pix_fmt", "rgb24",
"-f", "rawvideo",
"-"
],
stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL
)
print("Successfully connected to the camera via RTSP")
except Exception as e:
print("Failed to connect to the camera via RTSP")
print(e)
sys.exit(1)
# For audio acquisition (shared for progress bar + recording)
try:
self.ffmpeg_audio = subprocess.Popen(
[
"ffmpeg",
"-rtsp_transport", "tcp",
"-i", self.rtsp_url,
"-vn", # No video
"-acodec", "pcm_s16le", # 16bit PCM
"-ac", "1", # Mono
"-ar", "16000", # Sampling rate
"-f", "s16le", # raw PCM
"-"
],
stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL
)
print("Successfully connected to the camera via RTSP (audio)")
except Exception as e:
print("Failed to connect to the camera via RTSP (audio)")
print(e)
sys.exit(1)
"""
# For privacy mode, Pan/Tilt.
# As of January 30, 2026, cannot connect to Tapo FW 1.5.1 from pytapo
try:
self.pytapo = Tapo(CAM_ADDR, CAM_USER, CAM_PASS)
# Current privacy mode value
json_object = self.pytapo.getPrivacyMode()
p_mode = json_object['enabled']
print('Current privacy mode value: ' + p_mode)
print("pytapo successfully connected to the camera")
except Exception as e:
print("pytapo failed to connect to the camera")
print(e)
sys.exit(1)
"""
#####################################################
# GUI Design
#####################################################
#==================================
# Window Creation
#==================================
self.pack()
master.title('Smart Home Controller')
master.geometry('840x600')
master.resizable(False, False) # Set to non-resizable
# Icon setting
if os.path.exists(icon_path):
photo = tk.PhotoImage(file=icon_path)
master.iconphoto(False, photo)
#==================================
# Define Tabs
#==================================
frame_top = tk.Frame(master, width=840, height=600)
frame_top.propagate(False)
frame_top.place(x=0, y=0)
# Create notebook instance
nb = ttk.Notebook(master, width=840, height=545)
nb.place(x=0, y=40)
tab1 = tk.Frame(nb)
tab2 = tk.Frame(nb)
tab3 = tk.Frame(nb)
# Tab label settings
nb.add(tab1, text='Living Room', padding=4)
nb.add(tab2, text='Bedroom', padding=4)
nb.add(tab3, text='log', padding=4)
#==================================
# Define Frames for Tab1
#==================================
# tab1
frame1 = tk.Frame(tab1, width=100, height=95, borderwidth=1, relief='solid', padx=5, pady=3)
frame2 = tk.Frame(tab1, width=240, height=95, borderwidth=1, relief='solid', padx=5, pady=3)
frame3 = tk.Frame(tab1, width=170, height=95, borderwidth=1, relief='solid', padx=5, pady=3)
frame4 = tk.Frame(tab1, width=170, height=95, borderwidth=1, relief='solid', padx=5, pady=3)
frame5 = tk.Frame(tab1, width=125, height=95, borderwidth=1, relief='solid', padx=5, pady=3)
frame_camera = tk.Frame(tab1, width=825, height=415, borderwidth=1, relief='solid', padx=5, pady=3)
# Fix frame sizes
frame1.propagate(False)
frame2.propagate(False)
frame3.propagate(False)
frame4.propagate(False)
frame5.propagate(False)
frame_camera.propagate(False)
# Place frames
frame1.place(x=0, y=0)
frame2.place(x=105, y=0)
frame3.place(x=350, y=0)
frame4.place(x=525, y=0)
frame5.place(x=700, y=0)
frame_camera.place(x=0, y=100)
#==================================
# Widget Placement for Tab1
#==================================
# frame_top: Date and time (YYYY/MM/DD hh:mm)
self.label_ta = tk.Label(frame_top, font=('Meiryo', 14, 'bold'))
self.label_ta.place(x=0, y=0)
self._update_clock()
#-----------------------------
# frame1: Display thermo-hygrometer info
#-----------------------------
# Thermo-hygrometer label
label_f1 = tk.Label(frame1, text='Meter', font=('Meiryo', 12))
label_f1.place(x=0, y=0)
# Temperature
self.label_temperature = tk.Label(
frame1,
text=self.temperature,
fg='green',
bg='lightgray',
font=(FONT_DIG, 12)
)
self.label_temperature.place(x=10, y=30)
# Unit: Celsius
label_temperature_unit = tk.Label(frame1, text='°C', font=('Meiryo', 10))
label_temperature_unit.place(x=60, y=30)
# Humidity
self.label_moisture = tk.Label(
frame1,
text=self.humidity,
fg='green',
bg='lightgray',
font=(FONT_DIG, 12)
)
self.label_moisture.place(x=10, y=60)
self._update_meter()
# Unit: %
label_moisture_unit = tk.Label(frame1, text='%', font=('Meiryo', 10))
label_moisture_unit.place(x=60, y=60)
#-----------------------------
# frame2: AC control
#-----------------------------
# AC label
label_f2 = tk.Label(frame2, text='AC', font=('Meiryo', 12))
label_f2.place(x=0, y=0)
# Buttons
button_2a = tk.Button(frame2, text='Cool 28°', width='8', command=lambda: self._click_btn_sw('air_cool28'))
button_2a.place(x=10, y=30)
button_2b = tk.Button(frame2, text='Dry', width='8', command=lambda: self._click_btn_sw('air_dry'))
button_2b.place(x=80, y=30)
button_2c = tk.Button(frame2, text='Heat 25°', width='8', command=lambda: self._click_btn_sw('air_hot25'))
button_2c.place(x=150, y=30)
button_2d = tk.Button(frame2, text='OFF', width='8', command=lambda: self._click_btn_sw('air_off'), bg='red')
button_2d.place(x=10, y=60)
#-----------------------------
# frame3: Living room light control
#-----------------------------
# Light label
label_f3 = tk.Label(frame3, text='Lighting', font=('Meiryo', 12))
label_f3.place(x=0, y=0)
# Buttons
button_3a = tk.Button(frame3, text='ON', width='8', command=lambda: self._click_btn_sw('light_on'))
button_3a.place(x=10, y=30)
button_3b = tk.Button(frame3, text='Bright▲', width='8', command=lambda: self._click_btn_sw('light_up'))
button_3b.place(x=80, y=30)
button_3c = tk.Button(frame3, text='OFF', width='8', command=lambda: self._click_btn_sw('light_off'), bg='red')
button_3c.place(x=10, y=60)
button_3d = tk.Button(frame3, text='Dim▼', width='8', command=lambda: self._click_btn_sw('light_down'))
button_3d.place(x=80, y=60)
#-----------------------------
# frame4: TV control
#-----------------------------
# TV label
label_f4 = tk.Label(frame4, text='TV', font=('Meiryo', 12))
label_f4.place(x=0, y=0)
# Buttons
button_4a = tk.Button(frame4, text='ON', width='8', command=lambda: self._click_btn_sw('tv_on'))
button_4a.place(x=10, y=30)
button_4b = tk.Button(frame4, text='Vol▲', width='8', command=lambda: self._click_btn_sw('tv_vup'))
button_4b.place(x=80, y=30)
button_4c = tk.Button(frame4, text='OFF', width='8', command=lambda: self._click_btn_sw('tv_off'), bg='red')
button_4c.place(x=10, y=60)
button_4d = tk.Button(frame4, text='Vol▼', width='8', command=lambda: self._click_btn_sw('tv_vdown'))
button_4d.place(x=80, y=60)
#-----------------------------
# frame5: Spare area
#-----------------------------
#-----------------------------
# frame_camera: Camera operation
#-----------------------------
# Camera label
label_f5 = tk.Label(frame_camera, text='Camera', font=('Meiryo', 12))
label_f5.place(x=0, y=0)
# Camera feed display
self.canvas = tk.Canvas(frame_camera, width=640, height=360, bg='white')
self.canvas.place(x=10, y=30)
#.......................
# Camera Operation Menu
#.......................
# Placement offset
op_offset_x = 350 + 310
op_offset_y = 0
# Privacy mode ON/OFF (TBD)
# As of January 30, 2026, cannot connect to Tapo FW 1.5.1 from pytapo
self.check_privacy_flag = tk.BooleanVar(value=True)
check_privacy = tk.Checkbutton(
frame_camera,
text="Privacy Mode",
font=font.Font(family="Meiryo", size=10, overstrike=True),
variable=self.check_privacy_flag,
onvalue=True,
offvalue=False
)
check_privacy.place(x=op_offset_x, y=op_offset_y + 30)
# Motion detection result ON/OFF
self.check_move_flag = tk.BooleanVar(value=True)
check_move = tk.Checkbutton(
frame_camera,
text="Detection Result(Red)",
font=font.Font(family="Meiryo", size=10, overstrike=False),
variable=self.check_move_flag,
onvalue=True,
offvalue=False
)
check_move.place(x=op_offset_x, y=op_offset_y + 50)
# Last detection time
label_last_detected_time = tk.Label(frame_camera, text='Last Detected:', font=('Meiryo', 10))
label_last_detected_time.place(x=op_offset_x + 20, y=op_offset_y + 70)
self.label_detected_time = tk.Label(frame_camera, text=self.last_detected_time, font=('Meiryo', 10))
self.label_detected_time.place(x=op_offset_x + 110, y=op_offset_y + 70)
# Video capture button
op_offset_y = 125
button_recode = tk.Button(frame_camera, text='Record', width='8', command=self._click_cam_rec_btn)
button_recode.place(x=op_offset_x, y=op_offset_y)
# Still image capture button
button_capture = tk.Button(frame_camera, text='Capture', width='8', command=self._click_cam_cap_btn)
button_capture.place(x=op_offset_x + 75, y=op_offset_y)
# Pan/Tilt label (TBD)
# As of January 30, 2026, ONVIF PTZ cannot be used with Tapo FW 1.5.1. Same for pytapo.
op_offset_y = 160
label_pt = tk.Label(
frame_camera,
text='Pan/Tilt',
font=font.Font(family="Meiryo", size=10, overstrike=True)
)
label_pt.place(x=op_offset_x, y=op_offset_y)
# Pan/Tilt buttons (TBD)
# As of January 30, 2026, ONVIF PTZ cannot be used with Tapo FW 1.5.1. Same for pytapo.
op_offset_y = 180
button_pt_lt = tk.Button(frame_camera, text='＼', width='2', command=lambda: self._click_btn_cam_pan(-1, 1))
button_pt_lt.place(x=op_offset_x + 10, y=op_offset_y)
button_pt_mt = tk.Button(frame_camera, text='↑', width='2', command=lambda: self._click_btn_cam_pan(0, 1))
button_pt_mt.place(x=op_offset_x + 40, y=op_offset_y)
button_pt_rt = tk.Button(frame_camera, text='／', width='2', command=lambda: self._click_btn_cam_pan(1, 1))
button_pt_rt.place(x=op_offset_x + 70, y=op_offset_y)
op_offset_y = 210
button_pt_lm= tk.Button(frame_camera, text='←', width='2', command=lambda: self._click_btn_cam_pan(-1, 0))
button_pt_lm.place(x=op_offset_x + 10, y=op_offset_y)
button_pt_mm = tk.Button(frame_camera, text='＋', width='2', command=lambda: self._click_btn_cam_pan(X_DEFALUT, Y_DEFALUT))
button_pt_mm.place(x=op_offset_x + 40, y=op_offset_y)
button_pt_rm = tk.Button(frame_camera, text='→', width='2', command=lambda: self._click_btn_cam_pan(1, 0))
button_pt_rm.place(x=op_offset_x + 70, y=op_offset_y)
op_offset_y = 240
button_pt_lb = tk.Button(frame_camera, text='／', width='2', command=lambda: self._click_btn_cam_pan(-1, -1))
button_pt_lb.place(x=op_offset_x + 10, y=op_offset_y)
button_pt_mb = tk.Button(frame_camera, text='↓', width='2', command=lambda: self._click_btn_cam_pan(0, -1))
button_pt_mb.place(x=op_offset_x + 40, y=op_offset_y)
button_pt_rb = tk.Button(frame_camera, text='＼', width='2', command=lambda: self._click_btn_cam_pan(1, -1))
button_pt_rb.place(x=op_offset_x + 70, y=op_offset_y)
# Zoom label
op_offset_y = 270
label_zoom = tk.Label(frame_camera, text='Zoom', font=('Meiryo', 10))
label_zoom.place(x=op_offset_x, y=op_offset_y)
# Zoom slider
op_offset_y = 290
self.zoom_var = tk.DoubleVar()
scale_zoom = ttk.Scale(
frame_camera,
variable=self.zoom_var,
orient=tk.HORIZONTAL, # orientation
length=100, # length
from_=1.0, # min
to=4.0 # max
)
scale_zoom.place(x=op_offset_x + 10, y=op_offset_y)
# Scale ticks
op_offset_y = 310
label_zoom_1 = tk.Label(frame_camera, text='1', font=('Meiryo', 7))
label_zoom_1.place(x=op_offset_x + 10, y=op_offset_y)
label_zoom_2 = tk.Label(frame_camera, text='2', font=('Meiryo', 7))
label_zoom_2.place(x=op_offset_x + 54, y=op_offset_y)
label_zoom_4 = tk.Label(frame_camera, text='4', font=('Meiryo', 7))
label_zoom_4.place(x=op_offset_x + 100, y=op_offset_y)
# Volume label
op_offset_y = 340
label_volume = tk.Label(frame_camera, text='Vol[dB]', font=('Meiryo', 10))
label_volume.place(x=op_offset_x, y=op_offset_y)
# Display volume with Progressbar
op_offset_y = 360
self.vol = tk.IntVar()
bar_vol = ttk.Progressbar(frame_camera, maximum=100, mode="determinate", variable=self.vol)
bar_vol.place(x=op_offset_x + 10, y=op_offset_y)
# Scale ticks
op_offset_y = 385
label_0db = tk.Label(frame_camera, text='0', font=('Meiryo', 7))
label_0db.place(x=op_offset_x + 10, y=op_offset_y)
label_50db = tk.Label(frame_camera, text='50', font=('Meiryo', 7))
label_50db.place(x=op_offset_x + 50, y=op_offset_y)
label_100db = tk.Label(frame_camera, text='100', font=('Meiryo', 7))
label_100db.place(x=op_offset_x + 100, y=op_offset_y)
#==================================
# Define Tab2
#==================================
# TBD
#==================================
# Define Tab3
#==================================
self.text_area = scrolledtext.ScrolledText(
tab3,
width=115,
height=38,
font=('Source Han Code JP', 10),
)
self.text_area.grid(column=0, pady=5, padx=5)
self.text_area.focus()
# Start camera display
self._disp_image()
# Start volume meter thread (permanent)
threading.Thread(target=self._audio_thread, daemon=True).start()
def _disp_image(self):
"""
Process camera feed and display on Canvas.
"""
#==================================
# Obtain video from camera
#==================================
no_frame_cnt = 0
while True:
in_bytes = self.rstp.stdout.read(FRAME_WIDTH * FRAME_HEIGHT * 3)
if in_bytes:
frame = np.frombuffer(in_bytes, np.uint8).reshape([FRAME_HEIGHT, FRAME_WIDTH, 3])
break
# If nothing obtained for 1 minute, handle reconnection or camera restart (TBD)
if no_frame_cnt 2 15: # 15fps, so 15 frames per second
self._logger("Could not obtain frame for 1 minute.")
no_frame_cnt = 0
no_frame_cnt += 1
time.sleep(0.05)
#==================================
# Convert for OpenCV (motion detection/recording)
#==================================
cv_frame = cv2.cvtColor(frame, cv2.COLOR_RGB2BGR)
#==================================
# Motion Detection
#==================================
if not self.check_move_flag.get():
self.motion.reset()
else:
detect = False
color_blue = (0, 0, 255) # Border color (Blue - variable name mismatch in original code, kept logic same)
# Detect movement
rects = self.motion.detect(cv_frame)
# If too many detections, assume camera is Pan/Tilting and don't draw anything (adjust needed)
if 300 2 len(rects):
# Draw differences on screen
for (x, y, w, h) in rects:
# Ignore small or too large changes
if 200 2 w 2 20 and 200 2 h 2 20:
cv2.rectangle(
cv_frame, # Target image
(x, y), # Top-left coordinates
(x + w, y + h), # Bottom-right coordinates
(0, 0, 255), # Line color. RED
2 # Line thickness
)
detect = True
if detect:
self._logger("Motion detected")
# Update last detection time
now = datetime.datetime.now().strftime("%H:%M")
self.last_detected_time = now
self.label_detected_time.config(text=now)
#==================================
# Still image save
#==================================
if self.capture_flag:
self.recorder.save_image(cv_frame)
self.capture_flag = False
#==================================
# Video save (frame writing + REC display)
#==================================
if self.record_flag:
# Write frame
self.recorder.write_video(cv_frame)
# Measure time since recording started
rec_time = time.perf_counter() - self.rec_start_time
rec_timestamp = '{0:02d}:{1:02d}:{2:06.3f}'.format(
int(rec_time // 3600),
int(rec_time // 60),
rec_time % 60
)
# Since putText() only supports ASCII, draw a red circle using circle()
cv2.circle(
cv_frame,
(20, FRAME_HEIGHT - 23), # Position
14, # Size
(0, 0, 255), # RED
-1
)
# Add "REC" and recording time display at the bottom of the image (not included in saved file).
cv2.putText(
cv_frame,
f"REC {rec_timestamp}",
(36, FRAME_HEIGHT - 12), # Position
cv2.FONT_HERSHEY_SIMPLEX,
1.0, # Font size
(0, 0, 255), # RED
3,
cv2.LINE_AA
)
# Auto-stop after 5 minutes
if rec_time 2= 5 * 60:
self._logger("Automatically stopping recording after 5 minutes")
self.recorder.stop()
self.record_flag = False
#==================================
# Resize for display.
#==================================
# Resize if CANVAS_WIDTH and FRAME_WIDTH differ
if CANVAS_WIDTH != FRAME_WIDTH:
cv_frame = cv2.resize(cv_frame, (CANVAS_WIDTH, CANVAS_HEIGHT), interpolation=cv2.INTER_LINEAR )
# Get settings
if self.zoom_var.get() == 0:
zoom = 1
else:
zoom = self.zoom_var.get()
# Resize image based on canvas size multiplied by zoom factor.
# There is a moment when self.zoom_var.get() is 0 during window creation.
if zoom != 1:
h, w = cv_frame.shape[:2]
new_w = int(w * zoom)
new_h = int(h * zoom)
# Zoom from center
cv_frame = cv2.resize(cv_frame, (new_w, new_h), interpolation=cv2.INTER_LINEAR)
# Crop the center back to canvas size
x1 = (new_w - CANVAS_WIDTH) // 2
y1 = (new_h - CANVAS_HEIGHT) // 2
cv_frame = cv_frame[y1:y1 + CANVAS_HEIGHT, x1:x1 + CANVAS_WIDTH]
#==================================
# Convert to PhotoImage for Canvas rendering
#==================================
disp_frame = cv2.cvtColor(cv_frame, cv2.COLOR_BGR2RGB)
pil_image = Image.fromarray(disp_frame)
self.photo_image = ImageTk.PhotoImage(image=pil_image)
# Render image
self.canvas.delete("all")
self.canvas.create_image(
CANVAS_WIDTH / 2, # Image display position (Center of Canvas)
CANVAS_HEIGHT / 2,
image=self.photo_image # Display image data
)
# Execute _disp_image() every 10ms
self.after(10, self._disp_image)
def _click_btn_sw(self, key=None):
"""
Action when a SwitchBot operation button is pressed
"""
self._logger('push {} botton'.format(key))
device_id = sb_cmd.DIC_CMD.get(key)[0]
params = sb_cmd.DIC_CMD.get(key)[1]
self.sb.post_request(device_id, params)
def _click_btn_cam_pan(self, x, y):
"""
Action when a Pan/Tilt button is pressed
x, y are directions -1, 0, 1
For Tapo C210, speed values too small are ignored, so 0.3+ is recommended
"""
step = 0.4
# xml = self.onvif.build_relative_move_xml(x * step, y * step)
# self.onvif.send_onvif(xml)
def _click_cam_cap_btn(self):
"""
When still image capture button is pressed
"""
self.capture_flag = True
def _click_cam_rec_btn(self):
"""
When recording button is pressed
"""
if not self.record_flag:
self.recorder.start()
self.record_flag = True
self.rec_start_time = time.perf_counter()
else:
self.recorder.stop()
self.record_flag = False
def _audio_thread(self):
"""
Audio thread:
- Constantly reads RTSP audio, calculates volume and updates progress bar
- Only during recording, writes PCM from the same stream to a file
"""
while True:
data = self.ffmpeg_audio.stdout.read(CHUNK * 2)
if not data:
continue
# Volume meter processing
self.db = self._pcm_to_db(data)
# Update Tkinter in the main thread
root.after(0, self.vol.set, self.db)
# Write to file if recording
if self.record_flag:
self.recorder.write_audio(data)
def _pcm_to_db(self, pcm):
"""
Converts PCM16 numpy array to dBFS,
and returns a calibrated "sound meter style dB".
"""
try:
audio = np.frombuffer(pcm, dtype=np.int16)
except Exception:
return -90
if len(audio) == 0:
return -90
# Normalization
norm = np.abs(audio).astype(np.float64) / 32768.0
# Remove abnormal values (NaN, Inf)
norm = np.nan_to_num(norm, nan=0.0, posinf=1.0, neginf=0.0)
# Clip to prevent negative values
safe = np.clip(norm + 1e-6, 1e-12, None)
# Calculate dBFS
ln_val = np.log(safe)
dbfs_arr = 20 * ln_val / np.log(10)
# Remove abnormal values
dbfs_arr = np.nan_to_num(dbfs_arr, nan=-120.0, posinf=0.0, neginf=-120.0)
# Mean value
dbfs = float(np.mean(dbfs_arr))
if np.isnan(dbfs) or np.isinf(dbfs):
return -90
# Add calibrated DB_OFFSET
db = int(dbfs + DB_OFFSET)
return db
def _logger(self, text='', level='INFO'):
"""
Output log to Log tab and log file
"""
dt = datetime.datetime.now()
date = dt.strftime('%Y/%m/%d %H:%M:%S')
line = f'[{date}][{level:27}] {text}'
# Output to GUI
# Fall back to print if text_area doesn't exist yet
if hasattr(self, "text_area") and self.text_area is not None:
self.text_area.insert(tk.END, line + '\n')
self.text_area.see(tk.END)
else:
print(line)
# Output to file
log_path = "log.txt"
try:
with open(log_path, "a", encoding="utf-8") as f:
f.write(line + "\n")
except Exception as e:
print(f"Log write error: {e}")
# Size control (trim old lines if exceeding 10KB)
try:
if os.path.exists(log_path) and os.path.getsize(log_path) 2 10 * 1024:
with open(log_path, "r", encoding="utf-8") as f:
lines = f.readlines()
# Delete old lines (cut in half)
keep = lines[len(lines)//2:]
with open(log_path, "w", encoding="utf-8") as f:
f.writelines(keep)
# Record in log
print("log.txt trimmed (size 2 10KB)")
except Exception as e:
print(f"Log trim error: {e}")
def _update_clock(self):
'''
Update date and time every second
Format: YYYY/MM/DD hh:mm:ss
'''
now = datetime.datetime.now().strftime('%Y/%m/%d(%a) %H:%M:%S')
self.label_ta.config(text=now)
self.after(1000, self._update_clock)
def _update_meter(self):
'''
Update thermo-hygrometer every 5 minutes
'''
device_id = sb_cmd.DIC_DIV_ID.get('Meter')
temp, hum = self.sb.get_meter(device_id)
if temp is not None:
self.label_temperature.config(text=temp)
if hum is not None:
self.label_moisture.config(text=hum)
self.after(5 * 60 * 1000, self._update_meter)
if __name__ == "__main__":
root = tk.Tk()
app = Application(master=root)
app.mainloop()
Motion Detection
This is a simple version.
In the future, I would like to detect only cats using machine learning.
Reference site:
unit/motion_detector.py
import cv2
import numpy as np
class MotionDetector:
def __init__(self, blur_size=(21, 21), threshold=25):
self.avg = None # For saving the previous frame
self.blur_size = blur_size
self.threshold = threshold # Threshold for binarization (needs adjustment)
def reset(self):
"""
Reset the average frame for motion detection
"""
self.avg = None
def detect(self, frame):
"""
Detect movement and return a list of rectangles for the moved areas
frame: BGR (OpenCV frame)
return: [(x, y, w, h), ...]
"""
# Convert to grayscale
gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
gray = cv2.GaussianBlur(gray, self.blur_size, 0)
# Just create the average frame for the first time
if self.avg is None:
self.avg = gray.copy().astype("float")
return []
# Calculate the difference between the current frame and the moving average
cv2.accumulateWeighted(gray, self.avg, 0.5)
frame_delta = cv2.absdiff(gray, cv2.convertScaleAbs(self.avg))
# Binarize the frameDelta image
# If the threshold is too small, false positives increase; if too large, detection performance decreases.
thresh = cv2.threshold(frame_delta, self.threshold, 255, cv2.THRESH_BINARY)[1]
thresh = cv2.dilate(thresh, None, iterations=2)
# Extract contours
contours, _ = cv2.findContours(thresh, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
# Change format
rects = []
for c in contours:
x, y, w, h = cv2.boundingRect(c)
rects.append((x, y, w, h))
return rects
ONVIF PTZ
onvif.py is a file based on Article 3, but it is omitted here because it returns a Bad Request in Tapo FW 1.5.1 and cannot be used properly.
For SwitchBot Operation
sb_command.py is omitted because it is common with Article 2.
unit/switchbot.py
import requests
import json
from requests.exceptions import HTTPError, RequestException, Timeout
class SwitchBotController:
def __init__(self, token, api_host, logger=None):
self.token = token
self.api_host = api_host
self.logger = logger or (lambda msg: None)
self.headers = {
'Authorization': token,
'Content-Type': 'application/json; charset=utf8'
}
def _log(self, msg, level="INFO"):
"""
Passes logs to the GUI's Log tab
"""
if self.logger:
self.logger(msg, level)
def _get_device_status(self, device_id):
"""
Obtains the status of a Switchbot device.
"""
url = f"{self.api_host}/v1.0/devices/{device_id}/status"
try:
r = requests.get(url, headers=self.headers)
r.raise_for_status()
body = r.json()["body"]
# self._log(f"SwitchBot: status {device_id} → OK")
return body
except Timeout:
self._log(f"SwitchBot timeout error: ", 'ERROR')
except Exception as e:
self._log(f"SwitchBot error: {e}", 'ERROR')
raise
def get_meter(self, device_id):
"""
Obtains and returns the temperature and humidity of the thermo-hygrometer
"""
try:
status = self._get_device_status(device_id)
temp = '{0:4.1f}'.format(status.get('temperature'))
hum = '{0:4.1f}'.format(status.get('humidity'))
self._log(f"SwitchBot Thermo-hygrometer OK: {temp}°C {hum}%")
return temp, hum
except Exception as e:
self._log(f"SwitchBot Thermo-hygrometer error: {e}")
return None, None
def post_request(self, device_id, params):
'''
Issues instructions to the device.
'''
url = f"{self.api_host}/v1.0/devices/{device_id}/commands"
try:
res = requests.post(url, data=json.dumps(params), headers=self.headers)
data = res.json()
if data['message'] == 'success':
self._log(f"SwitchBot Command OK")
elif data['message'] == 'device offline':
self._log(f"SwitchBot Command NG. Device {device_id} is offline")
else:
self._log(f"SwitchBot error: {data}", 'ERROR')
return data
except Timeout:
self._log(f"SwitchBot timeout error: ", 'ERROR')
except Exception as e:
self._log(f"SwitchBot error: {e}", 'ERROR')
raise
For Recording and Audio
unit/recorder.py
import datetime
import os
import subprocess
import wave
import cv2
from pygame import mixer
# Sound for still image capture
mixer.init()
SHUTTER_SOUND = mixer.Sound(os.path.join('../', 'sound', 'shutter.mp3'))
class Recorder:
def __init__(self, video_dir="Video", canvas_size=(640, 360), logger=None):
self.video_dir = video_dir
# Video file name without audio
self.video_file_temp = os.path.join(self.video_dir, f"Video_temp.mp4")
self.canvas_size = canvas_size
self.logger = logger or (lambda msg: None) # Default does nothing
os.makedirs(video_dir, exist_ok=True)
self.recording = False
self.video_writer = None
self.audio_fp = None
self.video_file = None
self.audio_file = None
def _log(self, msg, level="INFO"):
"""
Passes logs to the GUI's Log tab
"""
if self.logger:
self.logger(msg, level)
def start(self):
"""
Start recording
"""
if self.recording:
return
timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
self.video_file = os.path.join(self.video_dir, f"Video_{timestamp}.mp4")
self.audio_file = os.path.join(self.video_dir, f"Audio_{timestamp}.raw")
fourcc = cv2.VideoWriter_fourcc(*"mp4v")
self.video_writer = cv2.VideoWriter(
self.video_file_temp,
fourcc,
15.0,
self.canvas_size
)
self.audio_fp = open(self.audio_file, "wb")
self.recording = True
self._log("Recorder: Recording started")
def stop(self):
"""
Stop recording
"""
if not self.recording:
return
self.recording = False
if self.video_writer:
self.video_writer.release()
self.video_writer = None
if self.audio_fp:
self.audio_fp.close()
self.audio_fp = None
# Convert raw to wav
wav_file = self.audio_file.replace(".raw", ".wav")
self._raw_to_wav(self.audio_file, wav_file)
# Merge audio and video
self._merge_audio_video(self.video_file_temp, wav_file, self.video_file)
self._log(f"Recorder: Recording ended {self.video_file}")
def write_video(self, frame):
"""
Write video file
"""
if self.recording and self.video_writer:
self.video_writer.write(frame)
def write_audio(self, pcm_bytes):
"""
Write audio file
"""
if self.recording and self.audio_fp:
self.audio_fp.write(pcm_bytes)
def save_image(self, frame):
"""
Save still image
"""
timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
filename = os.path.join(self.video_dir, f"ScreenShot_{timestamp}.png")
cv2.imwrite(filename, frame)
self._log(f"Recorder: Still image saved {filename}")
# Shutter sound
SHUTTER_SOUND.play()
def _raw_to_wav(self, raw_path, wav_path):
"""
Convert Raw file to Wav file
"""
with open(raw_path, "rb") as raw:
pcm = raw.read()
with wave.open(wav_path, "wb") as w:
w.setnchannels(1)
w.setsampwidth(2)
w.setframerate(16000)
w.writeframes(pcm)
# Delete Raw file
os.remove(raw_path)
def _merge_audio_video(self, video_file, audio_file, output_file):
"""
Combine audio and video files
Delete files before merging.
"""
subprocess.run(
[
"ffmpeg",
"-i", video_file,
"-i", audio_file,
"-c:v", "copy",
"-c:a", "aac",
"-shortest",
output_file
],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL
)
# Delete files before synthesis
os.remove(video_file)
os.remove(audio_file)
Conclusion
Actually, this viewer was already taking shape around 2023.
While I was putting it off because fixing freeze issues and writing the article was a hassle, the FW was updated, and the implemented functions became unusable 💦
I'm leaving it on the GUI in the hope that it will be usable again someday.
Discussion