# 背景

内部开发者Java版本官方SDK[：java版鸿蒙远控底层OS对应SDK](https://gitcode.com/OpenHarmonyToolkitsPlaza/HOScrcpy/tree/main)

可以通过此，获取稳定的h264流及远控up down move实现


协议底层：设备跟pc的协议是通过grpc协议





```Java
待实现
```

hypium版本
参考官方源码实现（可以投屏，但没法实现丝滑的up down move[：华为官方hypium文档](https://developer.huawei.com/consumer/cn/doc/harmonyos-guides-V5/hypium-python-guidelines-V5)

```python3
import logging
import os
import socket
import json
import asyncio

import websockets
from threading import Thread
from datetime import datetime

from hypium import UiDriver, KeyCode, BY
from hmdriver2.driver import Driver

# 配置 logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
)

# 设置 root logger 的级别
logging.getLogger().setLevel(logging.INFO)


class HmScrcpyServer:
    """录制屏幕并通过 WebSocket 提供实时投屏流"""

    def __init__(self, driver):
        self.driver = driver
        # self.client = driver._client
        self.clients = set()
        self.loop = asyncio.new_event_loop()
        asyncio.set_event_loop(self.loop)
        # self.ws_port = None

        self.gesture = driver.gesture

    def start(self, free_port, timeout=3600):

        self.is_running = True
        self.timeout = timeout
        self.stop_capture_if_running()
        msg_json = {'api': "startCaptureScreen", 'args': []}
        full_msg = {"module": "com.ohos.devicetest.hypiumApiHelper",
                    "method": "Captures", "params": msg_json,
                    "request_id": datetime.now().strftime("%Y%m%d%H%M%S%f")}
        self.driver._client._send_msg(full_msg)
        reply = self.driver._client._recv_msg(1024)
        logging.info(f'reply: {reply}')
        if b"true" in reply:
            thread_ws = Thread(target=self.run_loop, args=(self._start_websocket_server(free_port),))
            thread_ws.start()
            thread_record = Thread(target=self._record_worker)
            thread_record.start()
        else:
            raise RuntimeError("Fail to start screen capture")

    def run_loop(self, coro):
        asyncio.set_event_loop(self.loop)
        self.loop.run_until_complete(coro)
        self.loop.run_forever()

    def stop_capture_if_running(self):
        msg_json = {
            'api': "stopCaptureScreen", 'args': []
        }
        full_msg = {"module": "com.ohos.devicetest.hypiumApiHelper",
                    "method": "Captures", "params": msg_json,
                    "request_id": datetime.now().strftime("%Y%m%d%H%M%S%f")}
        self.driver._client._send_msg(full_msg)
        reply = self.driver._client._recv_msg(1024)
        logging.info(f'stop reply: {reply}')

    async def _start_websocket_server(self, free_port=8765):
        # 动态选择一个可用端口
        self.ws_port = free_port
        while True:
            try:
                server = await websockets.serve(self._handler, "0.0.0.0", self.ws_port)
                logging.info(f"WebSocket server started at ws://0.0.0.0:{self.ws_port}")
                await server.wait_closed()
                break
            except OSError as e:
                if e.errno == 10048:
                    self.ws_port += 1
                else:
                    raise e

    async def _handler(self, websocket, path):
        self.clients.add(websocket)
        logging.info(f"New WebSocket connection from {websocket.remote_address}")
        try:
            async for message in websocket:
                logging.info(f"Received message from {websocket.remote_address}: {message}")
                try:
                    data = json.loads(message)
                    if data.get('type') == 'touch':
                        action = data.get('action')
                        x = data.get('x')
                        y = data.get('y')

                        if action == 'down':
                            logging.info(f'down {x} {y}')
                            self.gesture.start(x, y, 0.1).action()
                        elif action == 'up':
                            logging.info(f'up {x} {y}')
                            self.gesture.pause(0.1).action()
                        elif action == 'move':
                            logging.info(f'move {x} {y}')
                            self.gesture.move(x, y, 0.1).action()
                        elif action == 'normal':
                            # 调用触摸操作函数，传递 x 和 y 坐标
                            self.driver.touch((x, y))
                        elif action == 'long':
                            # 处理长按操作
                            self.driver.touch(target=(x, y), mode='long')
                        elif action == 'double':
                            # 处理双击操作
                            self.driver.touch(target=(x, y), mode='double')
                        elif action == 'move':
                            x1 = data.get('x1')
                            y1 = data.get('y1')
                            x2 = data.get('x2')
                            y2 = data.get('y2')
                            self.driver.slide(start=(x1, y1), end=(x2, y2), slide_time=0.1)
                        else:
                            logging.info(f'not support action {action}')
                    elif data.get('type') == 'keyEvent':
                        # {"type":"keyEvent","eventNumber":3}
                        # 使用 adbutils 提供的 shell 方法执行按键事件命令
                        event_number = data['eventNumber']
                        # tudo 这里应该要把这些数字抽象出来为一个类，这里先跑通再优化
                        if event_number == 187:
                            # self. driver.press_key(KeyCode.MENU)
                            self.driver.swipe_to_recent_task()
                        elif event_number == 3:
                            # home
                            self.driver.go_home()
                        elif event_number == 4:
                            # home
                            self.driver.go_back()  # press_back 也可以用
                        elif event_number == 224:
                            # wake up display
                            self.driver.wake_up_display()
                    elif data.get('type') == 'text':
                        # input: {"type":"text","detail":"hello"}
                        # 1-借助hdc/shell命令执行输入文本操作 但是执行太慢了
                        # exam: hdc shell uitest uiInput inputText point_x point_y text
                        # self.driver.hdc(f"shell uitest uiInput inputText {data.get('x')} {data.get('y')} {data.get('detail')}")
                        if data.get('detail') == 'CODE_AC_BACK':
                            self.driver.press_key(KeyCode.DEL)
                        elif data.get('detail') == 'CODE_AC_ENTER':
                            self.driver.press_key(KeyCode.ENTER)
                        else:
                            # 常规字符输入
                            self.driver.shell(
                                f"uitest uiInput inputText {data.get('x')} {data.get('y')} {data.get('detail')}")
                        # 2-使用input_text方法
                        # self.driver.input_text(BY.type("TextInput"), data.get('detail'))
                except json.JSONDecodeError as e:
                    logging.info(f"Error decoding JSON: {e}")
                except Exception as e:
                    logging.info(f"Error processing message: {e}")
        except websockets.exceptions.ConnectionClosedError:
            pass
        finally:
            self.clients.remove(websocket)

    def _record_worker(self):
        tmp_data = b''
        start_flag = b'\xff\xd8'
        end_flag = b'\xff\xd9'
        while self.is_running:
            try:
                result = self.driver._client._recv_msg(4096 * 1024, print=False)
                # logging.info(f'len: {len(result)}')
                tmp_data += result

                while start_flag in tmp_data and end_flag in tmp_data:
                    start_index = tmp_data.index(start_flag)
                    end_index = tmp_data.index(end_flag) + 2
                    frame = tmp_data[start_index:end_index]
                    tmp_data = tmp_data[end_index:]

                    # 发送完整的JPEG图像帧给所有客户端
                    asyncio.run_coroutine_threadsafe(self._broadcast(frame), self.loop)

            except socket.timeout:
                logging.info("timeout")
                continue
            except Exception as e:
                logging.info("connection closed", e)
                self.is_running = False
                self.driver._client._connect_sock = None
                break

    async def _broadcast(self, data):
        for client in self.clients:
            try:
                await client.send(data)
            except websockets.exceptions.ConnectionClosedError:
                logging.info("WebSocket connection closed")
                self.clients.remove(client)
            except Exception as e:
                logging.info(f"Error sending data to client: {e}")

    def stop(self):
        self.is_running = False
        if self.driver._client._connect_sock is None:
            return
        msg_json = {'api': "stopCaptureScreen", 'args': []}
        full_msg = {"module": "com.ohos.devicetest.hypiumApiHelper",
                    "method": "Captures", "params": msg_json,
                    "request_id": datetime.now().strftime("%Y%m%d%H%M%S%f")}
        self.driver._client.sendall(full_msg)
        reply = self.driver._client._recv_msg(1024)
        if b"true" not in reply:
            logging.info("Fail to stop capture")
        self.driver._client.release()
        self.driver._client._connect_sock = None
        self.loop.stop()
        for client in self.clients:
            asyncio.run_coroutine_threadsafe(client.close(), self.loop)


def main():
    # 从环境变量中读取空闲端口
    free_port = int(os.environ.get('FREE_PORT'))
    logging.info(f'free_port: {free_port}')
    # 从环境变量中读取UDID
    udid = os.environ.get('UDID')
    logging.info(f'udid: {udid}')
    driver = Driver(udid)
    recorder = HmScrcpyServer(driver)
    recorder.start(free_port)


if __name__ == "__main__":
    main()
    # result = input("press to stop:")
    # recorder.stop()

```

基于官方python版本SDK实现

```python3.11
待补充
```

纯血鸿蒙远控实现

# 背景

1、需要通过mjpeg高速静态截图传流的方式获取视频流；

2、需要实现远控安卓设备，抽象出来需要实现up、down、move




# 代码

1、需要tidevice阿里库（目前最多支持iOS16及之前版本，17及之后的不支持，需要考虑兼容）
2、需要wda



```python
import contextlib
import logging
import os
import socket
import threading
import asyncio
import websockets
import tidevice
import wda
import time
import json

# 配置 logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
)

# 设置 root logger 的级别
logging.getLogger().setLevel(logging.INFO)


class SocketBuffer:

    def __init__(self, sock: socket.socket):
        self._sock = sock
        self._buf = bytearray()

    async def _drain(self):
        loop = asyncio.get_event_loop()
        _data = await loop.run_in_executor(None, self._sock.recv, 4096)
        # logging.info(f'_drain: Received {len(_data)} bytes')
        if _data is None or len(_data) == 0:
            raise IOError('socket closed')
        self._buf.extend(_data)
        return len(_data)

    async def read_until(self, delimiter: bytes) -> bytes:
        """ return without delimiter """
        while True:
            index = self._buf.find(delimiter)
            if index != -1:
                _return = self._buf[:index]
                self._buf = self._buf[index + len(delimiter):]
                return _return
            await self._drain()

    async def read_bytes(self, length: int) -> bytes:
        while length > len(self._buf):
            await self._drain()

        _return, self._buf = self._buf[:length], self._buf[length:]
        return _return

    async def write(self, data: bytes):
        loop = asyncio.get_event_loop()
        start_time = time.time()
        await loop.run_in_executor(None, self._sock.sendall, data)
        end_time = time.time()
        logging.info(f'write: Sent {len(data)} bytes in {end_time - start_time:.4f} seconds')


@contextlib.asynccontextmanager
async def make_screenrecord(c: wda.Client, t: tidevice.Device, websocket):
    logging.info(f'New WebSocket connection from {websocket.remote_address}')
    _old_fps = c.appium_settings()['mjpegServerFramerate']
    _fps = 60  # 提高帧率
    c.appium_settings({'mjpegServerFramerate': _fps})

    # Read image from WDA mjpeg server
    pconn = t.create_inner_connection(9100)  # default WDA mjpeg server port
    sock = pconn.get_socket()
    buf = SocketBuffer(sock)
    await buf.write(b'GET / HTTP/1.0\r\nHost: localhost\r\n\r\n')
    await buf.read_until(b'\r\n\r\n')
    logging.info('screenrecord is ready to begin')
    s = c.session()
    logging.info('session is ready to begin')

    async def _drain(stop_event, done_event):
        while not stop_event.is_set():
            try:
                # read http header
                length = None
                while True:
                    line = await buf.read_until(b'\r\n')
                    if line.startswith(b'Content-Length'):
                        length = int(line.decode('utf-8').split(': ')[1])
                        break
                while True:
                    if await buf.read_until(b'\r\n') == b'':
                        break

                imdata = await buf.read_bytes(length)
                # logging.info(f'Received frame of length {length} bytes')  # 输出图像数据的大小
                await websocket.send(imdata)
            except Exception as e:
                logging.info(f'Error in _drain: {e}')
                break
        done_event.set()

    async def _handle_messages():
        try:
            async for message in websocket:
                logging.info(f'Received message: {message}')
                try:
                    data = json.loads(message)
                    if data.get('type') == 'touch':
                        action = data.get('action')
                        x = data.get('x')
                        y = data.get('y')

                        if action == 'normal':
                            # 调用触摸操作函数，传递 x 和 y 坐标
                            logging.info(f'Touch action at ({x}, {y})')
                            await asyncio.to_thread(s.click, x, y)
                        elif action == 'long':
                            # 处理长按操作
                            logging.info(f'Long touch action at ({x}, {y})')
                        elif action == 'double':
                            # 处理双击操作
                            logging.info(f'Double touch action at ({x}, {y})')
                        elif action == 'move':
                            x1 = data.get('x1')
                            y1 = data.get('y1')
                            x2 = data.get('x2')
                            y2 = data.get('y2')
                            logging.info(f'Move action from ({x1}, {y1}) to ({x2}, {y2})')
                            await asyncio.to_thread(s.swipe, x1, y1, x2, y2)
                        else:
                            logging.info(f'Not supported action {action}')
                    elif data.get('type') == 'keyEvent':
                        event_number = data['eventNumber']
                        logging.info(f'Key event: {event_number}')
                        # tudo 这里应该要把这些数字抽象出来为一个类，这里先跑通再优化
                        if event_number == 187:
                            # self. driver.press_key(KeyCode.MENU)
                            pass
                        elif event_number == 3:
                            # home
                            await asyncio.to_thread(s.home)
                        elif event_number == 4:
                            # back
                            pass
                        elif event_number == 224:
                            # wake up display
                            await asyncio.to_thread(s.unlock)
                    elif data.get('type') == 'text':
                        detail = data.get('detail')
                        logging.info(f'Text input: {detail}')
                except json.JSONDecodeError as e:
                    logging.info(f"Error decoding JSON: {e}")
                except Exception as e:
                    logging.info(f"Error processing message: {e}")
        except websockets.ConnectionClosed:
            logging.info('WebSocket connection closed')
        except Exception as e:
            logging.error(f'Unexpected error: {e}')

    stop_event = threading.Event()
    done_event = threading.Event()
    drain_task = asyncio.create_task(_drain(stop_event, done_event))
    handle_messages_task = asyncio.create_task(_handle_messages())
    try:
        yield
    finally:
        stop_event.set()
        await done_event.wait()
        drain_task.cancel()
        handle_messages_task.cancel()
        c.appium_settings({'mjpegServerFramerate': _old_fps})
        logging.info('Screen recording stopped')


async def mjpeg_stream(websocket):
    # 从环境变量中读取UDID
    udid = os.environ.get('UDID')
    logging.info(f'udid: {udid}')

    c = wda.USBClient(udid)
    c.unlock()
    t = tidevice.Device(udid)

    try:
        async with make_screenrecord(c, t, websocket):
            await asyncio.Future()  # run forever
    except websockets.ConnectionClosed:
        logging.info('WebSocket connection closed')
    except Exception as e:
        logging.error(f'Unexpected error: {e}')


async def main():
    # 从环境变量中读取空闲端口
    free_port = int(os.environ.get('FREE_PORT'))
    logging.info(f'free_port: {free_port}')
    # # 从环境变量中读取UDID
    # udid = os.environ.get('UDID')
    # logging.info(f'udid: {udid}')
    server = await websockets.serve(mjpeg_stream, '0.0.0.0', free_port)
    logging.info(
        f'WebSocket server started at ws://{server.sockets[0].getsockname()[0]}:{server.sockets[0].getsockname()[1]}')
    await asyncio.Future()  # run forever


if __name__ == '__main__':
    asyncio.run(main())

```

iOS远控类scrcpy中touch事件实现

# 背景
对安卓、iOS、纯血鸿蒙做远控操控和投屏时，可以发现其底层获取视频的方式是有所不同的，有的h264模式，有的是高速静态截图的方式



# 区别
- 安卓：

参考scrcpy的h264格式，核心协议底层： [scrcpy开发者文档](https://github.com/Genymobile/scrcpy/blob/master/doc/develop.mdl)

- iOS：

参考wda/appium获取的视频流，也是mjpeg

建议参考实现：在 agent 侧再将mjpeg 转 h264，再推流（200ms 以下）


- 纯血鸿蒙：

1、基于agent.so提供的mjpeg截图流合成的视频（mjpeg截图传输数据量太大，比较耗带宽）；

2、针对h264的录屏方式，是采用libscreen_recorder.z.so这个动态库，它是rpc调用；

移动端远控底层原理分析

安卓远控scrcpy中touch事件实现

# 背景
初步开通了openai或deepseek开发者后台，拿到了密钥，需要通过后台api调起
deepseek开发者后台：https://api-docs.deepseek.com/zh-cn/



# api调用代码
```python3
# Please install OpenAI SDK first: `pip3 install openai`

from openai import OpenAI

client = OpenAI(api_key="<key>", base_url="https://api.deepseek.com")

response = client.chat.completions.create(
    model="deepseek-chat",
    messages=[
        {"role": "system",
         "content": "假如你是一名物理老师，需要出一份卷子给考生，要求10到选择题，5道多选，5道填空，2道大题。初二物理上题目"},
        {"role": "user", "content": "Hello"},
    ],
    stream=False
)

print(response.choices[0].message.content)

```
输出：

```bash
/home/lixuefu/env_deepseekProject/bin/python /home/lixuefu/deepseekProject/deepseek_chat.py 
你好！很高兴为你提供帮助。以下是一份初二物理上册的试卷，包含10道选择题（其中5道为多选题）、5道填空题和2道大题。希望对你有帮助！

---

### **初二物理上册试卷**

#### **一、选择题（每题2分，共20分）**
**（1-5题为单选题，6-10题为多选题）**

1. 下列哪个单位是国际单位制中长度的基本单位？  
   A. 米（m）  
   B. 厘米（cm）  
   C. 千米（km）  
   D. 毫米（mm）  

2. 一个物体做匀速直线运动，下列说法正确的是：  
   A. 速度大小和方向都不变  
   B. 速度大小不变，方向可能改变  
   C. 速度大小和方向都可能改变  
   D. 速度大小可能改变，方向不变  

3. 下列现象中，属于光的折射现象的是：  
   A. 水中倒影  
   B. 筷子在水中看起来弯曲  
   C. 平面镜成像  
   D. 小孔成像  

4. 关于声音的传播，下列说法正确的是：  
   A. 声音可以在真空中传播  
   B. 声音的传播速度与介质无关  
   C. 声音在固体中传播速度最快  
   D. 声音在空气中传播速度最快  

5. 下列哪个现象是由于惯性引起的？  
   A. 汽车刹车时乘客向前倾  
   B. 苹果从树上落下  
   C. 水往低处流  
   D. 气球升空  

6. （多选题）关于力的作用效果，下列说法正确的是：  
   A. 力可以改变物体的形状  
   B. 力可以改变物体的运动状态  
   C. 力可以改变物体的质量  
   D. 力可以改变物体的颜色  

7. （多选题）下列哪些现象属于光的反射现象？  
   A. 镜子中的像  
   B. 水中的倒影  
   C. 彩虹的形成  
   D. 日食的形成  

8. （多选题）关于摩擦力，下列说法正确的是：  
   A. 摩擦力总是阻碍物体的运动  
   B. 摩擦力的大小与接触面的粗糙程度有关  
   C. 摩擦力的大小与物体的质量无关  
   D. 摩擦力可以是有益的  

9. （多选题）下列哪些是能量的形式？  
   A. 动能  
   B. 势能  
   C. 热能  
   D. 电能  

10. （多选题）关于声音的特性，下列说法正确的是：  
    A. 音调与频率有关  
    B. 响度与振幅有关  
    C. 音色与波形有关  
    D. 声音的传播速度与温度无关  

---

#### **二、填空题（每空2分，共10分）**

1. 光在真空中的传播速度是________ m/s。  
2. 力的单位是________。  
3. 声音的传播需要________，不能在真空中传播。  
4. 物体由于地球吸引而受到的力叫做________。  
5. 光的反射定律中，入射角________反射角。  

---

#### **三、大题（每题10分，共20分）**

1. **计算题**  
   一辆汽车以20 m/s的速度匀速行驶了5分钟，求汽车行驶的路程是多少千米？

2. **实验题**  
   设计一个实验，验证“光的反射定律”。  
   （1）写出实验所需的器材。  
   （2）简述实验步骤。  
   （3）写出实验结论。

---

希望这份试卷符合你的需求！如果有任何修改或补充的地方，请随时告诉我！

Process finished with exit code 0

```