2 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)

import machine
adc32 = machine.ADC(machine.Pin(32))
while True:
    print(adc32.read_u16())

import machine
pwm0 = machine.PWM(machine.Pin(0))
pwm0.duty_u16(0)
pwm0.duty_u16(255)

from machine import Pin, PWM
import time

# 创建一个PWM对象
pwm = PWM(Pin(2))

# 设置PWM信号的频率为50Hz
# 每秒50个周期，所以每个周期的时间是1秒/50，即20ms。
pwm.freq(50)

# 一般来说，当PWM信号的高电平时间为1ms时，舵机转到0度；
# 当高电平时间为2ms时，舵机转到最大角度。
# 这个范围内的其他高电平时间对应的是0到180度之间的其他角度。

# 转到0度（）
pwm.duty(52) # 1ms / 20ms * 1024 = 51.2 取不低于最小值的整:52
time.sleep(1) # 等待一段时间让舵机转到指定位置

# 转到180度
pwm.duty(102) # 2ms / 20ms * 1024 = 102.4 取不高于最大值的整：102
time.sleep(1) # 等待一段时间让舵机转到指定位置

# 关闭PWM
pwm.deinit()

import machine

pin0 = machine.Pin(0, machine.Pin.IN)
while True:
    print(pin0.value())

import machine
import time

pin13 = machine.Pin(13, machine.Pin.OUT)
while True:
    pin13.value(0)
    time.sleep_ms(50)
    pin13.value(1)
    time.sleep_ms(50)

import mixiot
import machine
import seniverse_api


mixiot.wlan_connect('wifiname','wifipassword')
print(seniverse_api.weather_now('SGJl0ExVN-4j27msR','北京'))

import json
import asyncio
import websockets
from uuid import uuid4

# 音频文件测试路径。
audioFile = "test.mp3"
# 使用自己产品Id和apikey替换下列参数。
productId = "x"
apikey = "x"

#发送文本请求
async def textRequest(ws):
    content = {
        "aiType":"dm",
        "topic": 'nlu.input.text',
        "recordId": uuid4().hex,
        "refText": "测试"  #修改文本请求的输入
    }
    try:
        await ws.send(json.dumps(content))
        resp = await ws.recv()
        print(resp)
    except websockets.exceptions.ConnectionClosed as exp:
        print(exp)

#发送音频请求
async def audioRequest(ws):
    content = {
        "aiType": "dm",  #可选dm/asr, dm获取对话结果，asr只获取asr结果
        "topic": "recorder.stream.start",
        "recordId": uuid4().hex,
        "audio": {
            "audioType": "mp3",  #修改为测试文件的类型
            "sampleRate": 16000, #修改为测试文件的sampleRate
            "channel": 1,    #修改为测试文件的channel
            "sampleBytes": 2   #修改为测试文件的sampleBytes
        },
        "asrParams": {
            "realBack": True,   #实时返回asr结果
            "enableVAD": True,   #启动VAD
            "enablePunctuation": True,   #返回结果是否带拼音
            "enableTone": True,   #返回结果是否带声调
            "enableConfidence": True,   #返回结果是否带置信度
            "enableNumberConvert": True,  #返回结果是否进行数字转换
        },
    }
    try:
        #发送文本消息
        await ws.send(json.dumps(content))
        # 发送音频消息
        with open(audioFile, 'rb') as f:
            while True:
                chunk = f.read(400) #wav buffsize=3200 其他的400
                if not chunk:
                    await ws.send(bytes("", encoding="utf-8"))
                    break
                print(len(chunk))
                await ws.send(chunk)
        async for message in ws:
            print(message)
            resp = json.loads(message)
            if 'dm' in resp:
                break
    except websockets.exceptions.ConnectionClosed as exp:
        print(exp)
        ws.close()

async def dds_demo():
    url = f"ws://botai-dsg.and-home.cn:4443/dsg/v1/prod?productId={productId}&apikey={apikey}"
    print(url)
    async with websockets.connect(url) as websocket:
        #await textRequest(websocket)  #发送文本请求
        await audioRequest(websocket)  #发送音频请求
asyncio.get_event_loop().run_until_complete(dds_demo())