import sys import os sys.path.append("../arduino_api") import arduino_api as ctrl import time import os import threading import json import subprocess import psutil import threading class Flank(object): motor = None name = None curr_angle = None code = None type = "flank" def __init__(self, name, code): self.motor = ctrl.Motor(code) self.curr_angle = self.motor.get() self.name = name self.code = code print("初始化操纵面 {}, 硬件代号绑定为 {}".format(name, code)) def set_angle(self, angle): self.motor.set(angle) self.curr_angle = self.motor.get() #print("将 {} 的角度设置为 {} 度".format(self.name, angle)) def add_angle(self, add_angle): self.motor.set(self.curr_angle + add_angle) self.curr_angle = self.motor.get() def status(self): return {"type":"Flank", "name":self.name, "angle":self.curr_angle} def selfchk(self): print("开始自检操纵面 {}".format(self.name)) self.add_angle(30) self.add_angle(-60) self.set_angle(0) class Engine(object): engine = None curr_speed = None code = None name = None type = "engine" def __init__(self, name, code): self.code = code self.name = name self.engine = ctrl.Engine(code) self.curr_speed = self.engine.get() print("初始化引擎 {}, 硬件代号绑定为 {}".format(name, code)) def set_speed(self, new_speed): self.engine.tune(new_speed) self.curr_speed = self.engine.get() ##print("将 {} 的转速调谐为 {} RPM".format(self.name, new_speed)) def add_speed(self, add_speed): self.engine.tune(self.curr_speed + add_speed) self.curr_speed = self.engine.get() ##print("将 {} 的转速加成 {} RPM".format(self.name, add_speed)) def get_speed(self): ##print("{} 当前转速为 {} RPM".format(self.name, self.curr_speed)) self.curr_speed = self.engine.get() return self.curr_speed def status(self): ##print(self.curr_speed) return {"type":"Engine", "name":self.name, "speed":self.get_speed()} def selfchk(self): print("开始自检引擎 {}".format(self.name)) self.add_speed(3) self.add_speed(-6) self.set_speed(0) self.curr_speed = self.engine.get() """ @liteon-proximity: 光学接近传感器,用于检测物体的接近 !yas537-mag: 磁力计传感器,用于测量磁场强度和方向 @liteon-light: 光传感器,用于测量环境光强度 @MPU6050-gyro: 陀螺仪传感器,用于测量角速度 @MPU6050-accel: 加速度计传感器,用于测量线性加速度 !liteon-pocket: 可能是特定用途的传感器,具体功能需参考设备文档 *yas537-orientation: 方向传感器,结合磁力计和加速度计数据来确定设备的方向 @Game Rotation Vector Sensor: 用于测量设备的旋转向量,常用于游戏和增强现实应用 @GeoMag Rotation Vector Sensor: 结合地磁和加速度数据来测量设备的旋转向量 @Gravity Sensor: 测量重力加速度,用于确定设备的姿态 *Linear Acceleration Sensor: 测量去除重力影响后的线性加速度 @Rotation Vector Sensor: 综合陀螺仪和加速度计数据来测量设备的旋转向量 """ class Sensor: data = { "speed": 0, "battery": 0, "xangle": 0, "yangle": 0, "zangle": 0, "sign": 0, "cpuload": 0, "memload": 0, "memsum": 0, "torch": 0, "acc": 0 } is_torch_on = 0 @staticmethod def speed(): #TODO Sensor.data["speed"] = -1 return Sensor.data["speed"] @staticmethod def battery(): # 电池剩余 return Sensor.data["sysbattery"] def battery(): # TODO: 电机电池剩余 return Sensor.data["battery"] @staticmethod def xangle(): # X迎角 return Sensor.data["xangle"] @staticmethod def yangle(): return Sensor.data["yangle"] @staticmethod def zangle(): return Sensor.data["zangle"] @staticmethod def sign(): # 信号强度 Sensor.data["sign"] = -1 return Sensor.data["sign"] @staticmethod def cpuload(): # CPU占用(百分比) Sensor.data["cpuload"] = psutil.cpu_percent(interval=1) / 100 return Sensor.data["cpuload"] @staticmethod def memload(): # 内存占用(百分比) Sensor.data["memload"] = psutil.virtual_memory().percent / 100 return Sensor.data["memload"] @staticmethod def memsum(): # 内存总量(MB) return Sensor.data["memsum"] @staticmethod def torch(): return Sensor.data["torch"] @staticmethod def torchon(): # 打开电筒 Sensor.is_torch_on = not Sensor.is_torch_on to = {0:"off", 1:"on"} os.system(f"torch {to[Sensor.is_torch_on]} &") @staticmethod def refresh(): Sensor.speed() Sensor.battery() Sensor.xangle() Sensor.yangle() Sensor.zangle() Sensor.sign() Sensor.cpuload() Sensor.memload() Sensor.memsum() Sensor.torch() @staticmethod def stat(): #Sensor.refresh() return Sensor.data def update_battery(): while True: result = subprocess.run(['termux-battery-status'], capture_output=True, text=True) data = json.loads(result.stdout) Sensor.data["sysbattery"] = data["percentage"] def update_orientation(): while True: # TODO: 优化 result = subprocess.run(['termux-sensor', '-s', 'yas537-orientation', '-n', '1'], capture_output=True, text=True) data = json.loads(result.stdout) Sensor.data["xangle"] = data["yas537-orientation"]["values"][0] Sensor.data["yangle"] = data["yas537-orientation"]["values"][1] Sensor.data["zangle"] = data["yas537-orientation"]["values"][2] def update_acceleration(): while True: result = subprocess.run(['termux-sensor', '-s', 'Linear Acceleration Sensor', '-n', '1'], capture_output=True, text=True) data = json.loads(result.stdout) Sensor.data["acc"] = data["Linear Acceleration Sensor"]["values"] def update_func(cmd): while True: exec(cmd) time.sleep(0.1) def init(): Sensor.data = { "speed": 0, "battery": 0, "xangle": 0, "yangle": 0, "zangle": 0, "sign": 0, "cpuload": 0, "memload": 0, "memsum": 0, "torch": 0, "acc": 0 } os.system("termux-torch off &") Sensor.is_torch_on = 0 Sensor.data["memsum"] = psutil.virtual_memory().total / (1024 * 1024) def deamon(): # 启动线程 Sensor.battery_thread = threading.Thread(target=Sensor.update_battery) Sensor.orientation_thread = threading.Thread(target=Sensor.update_orientation) Sensor.acceleration_thread = threading.Thread(target=Sensor.update_acceleration) Sensor.other_thread = threading.Thread(target=Sensor.update_func, args=("""Sensor.sign()\nSensor.cpuload()\nSensor.memload()\nSensor.torch()""")) Sensor.battery_thread.start() Sensor.orientation_thread.start() Sensor.acceleration_thread.start() def stop(): Sensor.orientation_thread.join() Sensor.battery_thread.join() Sensor.acceleration_thread.join() Sensor.other_thread.join()