4 posts tagged with "smarthome"

def calculate_pan_position_bytes(pan_pos_value):
    HEX_VALUES = [4096, 256, 16, 1]  # 定义常量
    pan_pos_ints = []
    for i, value in enumerate(HEX_VALUES):
        pan_pos_ints.append(pan_pos_value // value)
        pan_pos_value %= value
    # 转换为2位16进制字符串
    pan_pos_strs = [f"{i:02X}" for i in pan_pos_ints]
    return "".join(pan_pos_strs)
# 将17转化为16进制，应该是11，拆分加0，应该是0101
# 前方补0到总长度为8位，结果与我们预期一致
print(calculate_pan_position_bytes(17))
# 00000101

def calculate_pan_position_bytes(pan_pos_value, axis_type):
    """
    计算轴（旋转）的位置字节。

    参数:
    pan_pos_value (int): 位置值，
    axis_type (str): 轴的类型 ('y' or 'Y' for Y-axis, others for Z-axis)

    返回:
    pan_step_str (str): 计算得到的平移位置字节，格式为十六进制字符串。
    """
    if axis_type.lower() == "y":
        pan_pos_value = max(-177.5, min(pan_pos_value, 177.5))  # 限制取值范围
    else:
        pan_pos_value = max(-21, min(pan_pos_value, 21))  # 限制取值范围

    pan_pos_value = int(pan_pos_value * 25)  # 将角度转换为步长
    pan_direction = "-" if pan_pos_value < 0 else "+"  # 设定旋转方向
    pan_pos_value = abs(pan_pos_value)  # 取绝对值

    HEX_VALUES = [4096, 256, 16, 1]  # 定义常量

    pan_pos_ints = []
    for i, value in enumerate(HEX_VALUES):
        if pan_direction == "+":
            pan_pos_ints.append(pan_pos_value // value)
        else: # 异或操作
            pan_pos_ints.append((pan_pos_value // value)^ 0xF)
            if i == 3 : # 最后一个数字，取反后加1
                pan_pos_ints[-1] = pan_pos_ints[-1]+1
        pan_pos_value %= value

    # 转换为2位16进制字符串
    pan_pos_strs = [f"{i:02X}" for i in pan_pos_ints]
    return "".join(pan_pos_strs)

import keyboard
from usbcamera import  *
from usbcamera import move_to_absolute_position
"""
设备 "/dev/ttyUSB0" 的云台旋转至绝对定位：
Y轴转到180度，速度为9.
Z轴转到30度，速度为10
"""
move_to_absolute_position(vv=9, ww=10, Y=180, Z=30, device="COM16")

# 关联键盘事件和控制函数
keyboard.on_press_key("up", lambda _: turn_up(device="COM16"))
keyboard.on_press_key("down", lambda _: turn_down(device="COM16"))
keyboard.on_press_key("left", lambda _: turn_left(device="COM16"))
keyboard.on_press_key("right", lambda _: turn_right(device="COM16"))
keyboard.on_press_key("enter", lambda _: move_home(device="COM16"))
keyboard.on_press_key("space", lambda _: turn_stop(device="COM16"))
# 按下数字1则转动到水平最左，垂直最下，可以根据自己需要多预设几个目标角度。
keyboard.on_press_key("1", lambda _: move_to_absolute_position(vv=10, ww=10, Y=180, Z=-30, device="COM16"))

# 让脚本保持运行状态以捕获事件
keyboard.wait("esc")  # 按 'esc' 键退出

#!/usr/bin/env python3
# coding:utf-8

import serial
import serial.tools.list_ports
import time

# VISCA命令集
commands = {
    "stop": "81010601{vv}{ww}0303FF",
    "left": "81010601{vv}{ww}0103FF",
    "right": "81010601{vv}{ww}0203FF",
    "up": "81010601{vv}{ww}0301FF",
    "down": "81010601{vv}{ww}0302FF",
    "upleft": "81010601{vv}{ww}0101FF",
    "upright": "81010601{vv}{ww}0201FF",
    "downleft": "81010601{vv}{ww}0102FF",
    "downright": "81010601{vv}{ww}0202FF",
    "absolute_position": "81010602{vv}{ww}{Y}{Z}FF",
    "relative_position": "81010603{vv}{ww}{Y}{Z}FF",
    "home": "81010604FF",
    "reset": "81010605FF",
}


def send_visca_command(command, device):
    """
    通过串口向摄像机发送VISCA命令。

    参数:
    command (str): 要发送的VISCA命令，格式为十六进制字符串。

    返回:
    response (bytes): 从摄像机接收到的响应。
    """
    try:
        ser = serial.Serial(device, 9600, timeout=1)  # 初始化串口
        command_bytes = bytearray.fromhex(command)  # 将命令转换为字节
        ser.write(command_bytes)  # 发送命令
        response = ser.read_all()  # 读取响应
        ser.close()  # 关闭串口
        return response
    except:
        ports_list = list(serial.tools.list_ports.comports())
        if len(ports_list) <= 0:
            print("未发现端口")
        else:
            for comport in ports_list:
                if "USB" in str(comport):
                    print("发现USB端口：", comport.device, comport.description)


def calculate_pan_speed_bytes(pan_speed_value):
    """
    计算轴（旋转）的位置字节。

    参数:
    pan_speed_value (int): 速度值，0-16

    返回:
    pan_step_str (str): 计算得到的平移位置字节，格式为十六进制字符串。
    """

    pan_speed_value = max(0, min(pan_speed_value, 16))  # 限制取值范围

    # 转为2位16进制
    return f"{pan_speed_value:02X}"


def calculate_pan_position_bytes(pan_pos_value, axis_type):
    """
    计算轴（旋转）的位置字节。

    参数:
    pan_pos_value (int): 位置值，
    axis_type (str): 轴的类型 ('y' or 'Y' for Y-axis, others for Z-axis)

    返回:
    pan_step_str (str): 计算得到的平移位置字节，格式为十六进制字符串。
    """
    if axis_type.lower() == "y":
        pan_pos_value = max(-177.5, min(pan_pos_value, 177.5))  # 限制取值范围
    else:
        pan_pos_value = max(-21, min(pan_pos_value, 21))  # 限制取值范围

    pan_pos_value = int(pan_pos_value * 25)  # 将角度转换为步长
    pan_direction = "-" if pan_pos_value < 0 else "+"  # 设定旋转方向
    pan_pos_value = abs(pan_pos_value)  # 取绝对值

    HEX_VALUES = [4096, 256, 16, 1]  # 定义常量

    pan_pos_ints = []
    for i, value in enumerate(HEX_VALUES):
        if pan_direction == "+":
            pan_pos_ints.append(pan_pos_value // value)
        else: # 异或操作
            pan_pos_ints.append((pan_pos_value // value)^ 0xF)
            if i == 3 : # 最后一个数字，取反后加1
                pan_pos_ints[-1] = pan_pos_ints[-1]+1
        pan_pos_value %= value

    # 转换为2位16进制字符串
    pan_pos_strs = [f"{i:02X}" for i in pan_pos_ints]
    return "".join(pan_pos_strs)


def create_command(command_key, vv=10, ww=10, Y=None, Z=None):
    """
    创建VISCA命令。

    参数:
    command_key (str): 命令键名。
    vv (str): 水平方向速度，取值范围为0-16
    ww (str): 垂直方向速度，取值范围为0-16
    Y (str): 控制水平旋转的位置。
    Z (str): 控制垂直旋转的位置。

    返回:
    command (str): 格式化后的VISCA命令字符串。

    异常:
    ValueError: 当命令需要Y和Z参数时，若未提供，则抛出异常。
    """
    if command_key in ["home", "reset"]:
        return commands[command_key]
    if command_key in ["absolute_position", "relative_position"]:
        if Y is None or Z is None:
            raise ValueError("Y和Z为位置命令,必须提供")
        return commands[command_key].format(
            vv=calculate_pan_speed_bytes(vv),
            ww=calculate_pan_speed_bytes(ww),
            Y=calculate_pan_position_bytes(Y, "y"),
            Z=calculate_pan_position_bytes(Z, "z"),
        )

    return commands[command_key].format(
        vv=calculate_pan_speed_bytes(vv),
        ww=calculate_pan_speed_bytes(ww),
    )


# 控制函数示例
def turn_stop(vv=0, ww=0, device="/dev/ttyUSB0"):
    return send_visca_command(create_command("stop", vv, ww), device)


def turn_left(vv=10, ww=10, device="/dev/ttyUSB0"):
    return send_visca_command(create_command("left", vv, ww), device)


def turn_right(vv=10, ww=10, device="/dev/ttyUSB0"):
    return send_visca_command(create_command("right", vv, ww), device)


def turn_up(vv=10, ww=10, device="/dev/ttyUSB0"):
    return send_visca_command(create_command("up", vv, ww), device)


def turn_down(vv=10, ww=10, device="/dev/ttyUSB0"):
    return send_visca_command(create_command("down", vv, ww), device)


def move_home(device="/dev/ttyUSB0"):
    return send_visca_command(create_command("home"), device)


def move_to_absolute_position(vv=10, ww=10, Y=0, Z=0, device="/dev/ttyUSB0"):
    return send_visca_command(create_command("absolute_position", vv, ww, Y, Z), device)

from flask import Flask, Response, render_template_string, stream_with_context
import time
import cv2
from PIL import ImageGrab, Image
import threading
from io import BytesIO
from greenlet import getcurrent as get_ident
import pyaudio


class Audio:
    def __init__(self):
        self.FORMAT = pyaudio.paInt16
        self.CHANNELS = 2
        self.RATE = 44100
        self.CHUNK = 1024
        self.bitsPerSample = 16
        self.audio = pyaudio.PyAudio()
        self.first_run = True
        self.wav_header = self.genHeader(self.RATE, self.bitsPerSample, self.CHANNELS)
        self.stream = self.audio.open(
            format=self.FORMAT,
            channels=self.CHANNELS,
            rate=self.RATE,
            input=True,
            input_device_index=1,
            frames_per_buffer=self.CHUNK,
        )
        self.get_audio = self._get_audio_first_time

    def genHeader(self, sampleRate, bitsPerSample, channels):
        datasize = 2000 * 10**6
        o = bytes("RIFF", "ascii")
        o += (datasize + 36).to_bytes(4, "little")
        o += bytes("WAVE", "ascii")
        o += bytes("fmt ", "ascii")
        o += (16).to_bytes(4, "little")
        o += (1).to_bytes(2, "little")
        o += (channels).to_bytes(2, "little")
        o += (sampleRate).to_bytes(4, "little")
        o += (sampleRate * channels * bitsPerSample // 8).to_bytes(4, "little")
        o += (channels * bitsPerSample // 8).to_bytes(2, "little")
        o += (bitsPerSample).to_bytes(2, "little")
        o += bytes("data", "ascii")
        o += (datasize).to_bytes(4, "little")
        return o

    def _get_audio_first_time(self):
        data = self.stream.read(self.CHUNK)
        self.get_audio = self._get_audio_subsequent_times  # switch the method
        return self.wav_header + data

    def _get_audio_subsequent_times(self):
        return self.stream.read(self.CHUNK)

class CameraEvent(object):
    def __init__(self):
        self.events = {}

    def wait(self):
        ident = get_ident()
        if ident not in self.events:
            self.events[ident] = [threading.Event(), time.time()]
        return self.events[ident][0].wait()

    def set(self):
        now = time.time()
        remove = None
        for ident, event in self.events.items():
            if not event[0].is_set():
                event[0].set()
                event[1] = now
            else:
                if now - event[1] > 5:
                    remove = ident
        if remove:
            del self.events[remove]

    def clear(self):
        self.events[get_ident()][0].clear()


class BaseCamera(object):
    thread = None
    frame = None
    last_access = 0
    event = CameraEvent()

    def __init__(self):
        if BaseCamera.thread is None:
            BaseCamera.last_access = time.time()
            BaseCamera.thread = threading.Thread(target=self._thread)
            BaseCamera.thread.start()
            while self.get_frame() is None:
                time.sleep(0)

    def get_frame(self):
        BaseCamera.last_access = time.time()
        BaseCamera.event.wait()
        BaseCamera.event.clear()
        return BaseCamera.frame

    @staticmethod
    def frames():
        raise RuntimeError("Must be implemented by subclasses.")

    @classmethod
    def _thread(cls):
        print("Starting camera thread.")
        frames_iterator = cls.frames()
        for frame in frames_iterator:
            BaseCamera.frame = frame
            BaseCamera.event.set()
            time.sleep(0)
            if time.time() - BaseCamera.last_access > 10:
                frames_iterator.close()
                print("Stopping camera thread due to inactivity.")
                break
        BaseCamera.thread = None


class Camera(BaseCamera):
    video_source = 0

    @staticmethod
    def set_video_source(source):
        Camera.video_source = source

    @staticmethod
    def frames():
        camera = cv2.VideoCapture(Camera.video_source)
        if not camera.isOpened():
            raise RuntimeError("Error")
        while True:
            image = ImageGrab.grab()
            image = image.resize((1366, 750), Image.LANCZOS)
            output_buffer = BytesIO()
            image.save(output_buffer, format="JPEG", quality=100)
            frame = output_buffer.getvalue()
            yield frame
app = Flask(__name__)

def gen(camera):
    while True:
        frame = camera.get_frame()
        yield (b"--frame\r\n" b"Content-Type: image/jpeg\r\n\r\n" + frame + b"\r\n")

def gen_audio(audio):
    while True:
        data = audio.get_audio()
        yield data

@app.route("/video_feed")
def video_feed():
    return Response(gen(Camera()), mimetype="multipart/x-mixed-replace; boundary=frame")

@app.route("/audio_feed")
def audio_feed():
    return Response(stream_with_context(gen_audio(Audio())))

@app.route("/")
def index():
    global mode
    video_tag = """<img src="{{ url_for('video_feed') }}">"""
    audio_tag = """<audio autoplay style="display:none;"><source src="{{ url_for('audio_feed') }}" type="audio/x-wav; codec=pcm">Your browser does not support the audio element.</audio>"""

    tags = {0: video_tag + audio_tag, 1: audio_tag, 2: video_tag}

    content = tags[mode]

    return render_template_string(
        """<html>
    <head>
        <title>{title}</title>
        <link rel="icon" href="data:image/svg+xml;base64,CjxzdmcgeG1sbnM9Imh0dHA6Ly93d3cudzMub3JnLzIwMDAvc3ZnIiB2aWV3Qm94PSIwIDAgNTAgNTAiPgogICAgPGNpcmNsZSBjeD0iMjUiIGN5PSIyNSIgcj0iMjAiIGZpbGw9InJlZCIgLz4KPC9zdmc+Cg==">
    </head>
    <body>{content}</body>
    </html>""".format(
            title=["Intranet Broadcast", "Audio Sharing", "Screen Sharing"][mode],
            content=content,
        )
    )

if __name__ == "__main__":
    local_host = "127.0.0.1"
    ip_host = "0.0.0.0"
    port = 8001
    mode = int(input("Please select the mode: 0 for Intranet Broadcast, 1 for Audio Sharing, 2 for Screen Sharing: "))
    app.run(threaded=True, host=ip_host, port=port)