Unable use ai_module on tt Drone

[FaterYU]

Problem

AI module apply the example model from DJI Education Hub, which detect EP.

TT Drone run the program 01_ai_module_tt.py in this repository.

# -*-coding:utf-8-*-
# Copyright (c) 2020 DJI.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License in the file LICENSE.txt or at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.


import time
from robomaster import robot


def sub_ia_info_handler(ai_info):
    id, x, y, w, h, C = ai_info
    print("ai target id:{0}, x:{1:.3f}, y:{2:.3f}, w:{3:.3f}, h:{4:.3f}, C:{5}".format(id, x, y, w, h, C))


if __name__ == '__main__':
    tl_drone = robot.Drone()
    tl_drone.initialize()

    # 订阅AI模块信息
    tl_drone.ai_module.sub_ai_info(freq=5, callback=sub_ia_info_handler)
    time.sleep(60)
    # 取消订阅
    tl_drone.ai_module.unsub_ai_info()

    tl_drone.close()

I hope it can return detecting data, but it print nothing actually.

Explore

1. program doesn't call the callback function.

2. when run tl_drone.ai_module.get_ai(), it return None.

       def get_ai(self):
           """ 获取AI模块信息
   
           :return: int: AI模块明文字符串
           """
           cmd = "ai?"
           proto = protocol.TextProtoDrone()
           proto.text_cmd = cmd
           msg = protocol.TextMsg(proto)
           try:
               resp_msg = self._client.send_sync_msg(msg)
               if resp_msg:
                   proto = resp_msg.get_proto()
                   if proto:
                       return int(proto.resp)
                   else:
                       return None
               else:
                   logger.warning("Drone: get_ai failed.")
           except Exception as e:
               logger.warning("Drone: get_ai, send_sync_msg exception {0}".format(str(e)))
               return None
   

   Obviously, it raise error when failed to send message.

3. In dds.py, the exec function has not called successfully.

       def exec(self):
           self._callback(self.data_info(), *self._cb_args, **self._cb_kw)
   

   In class Subscriber, I can't locate the problem why it can't run the exec function.

   def _dispatch_task(self):
           self._dispatcher_running = True
           logger.info("Subscriber: dispatcher_task is running...")
           while self._dispatcher_running:
               msg = self._msg_queue.get(1)
               if msg is None:
                   if not self._dispatcher_running:
                       break
                   continue
               self._dds_mutex.acquire()
               for name in self._publisher:
                   handler = self._publisher[name]
                   logger.debug("Subscriber: msg: {0}".format(msg))
                   proto = msg.get_proto()
                   if proto is None:
                       logger.warning("Subscriber: _publish, msg.get_proto None, msg:{0}".format(msg))
                       continue
                   if handler.subject.type == DDS_SUB_TYPE_PERIOD and\
                           msg.cmdset == 0x48 and msg.cmdid == 0x08:
                       logger.debug("Subscriber: _publish: msg_id:{0}, subject_id:{1}".format(proto._msg_id,
                                                                                              handler.subject._subject_id))
                       if proto._msg_id == handler.subject._subject_id:
                           handler.subject.decode(proto._data_buf)
                           if handler.subject._task is None:
                               handler.subject._task = self.excutor.submit(handler.subject.exec)
                           if handler.subject._task.done() is True:
                               handler.subject._task = self.excutor.submit(handler.subject.exec)
                   elif handler.subject.type == DDS_SUB_TYPE_EVENT:
                       if handler.subject.cmdset == msg.cmdset and handler.subject.cmdid == msg.cmdid:
                           handler.subject.decode(proto._data_buf)
                           if handler.subject._task is None:
                               handler.subject._task = self.excutor.submit(handler.subject.exec)
                           if handler.subject._task.done() is True:
                               handler.subject._task = self.excutor.submit(handler.subject.exec)
               self._dds_mutex.release()
               logger.info("Subscriber: _publish, msg is {0}".format(msg))
   
   

Other

I am not sure whether the ai model can not detect the EP so that the program haven't return message? Hope can get some helps.

[kenton2000]

Please use "Dji Edu Hub-Robomaster Assistant" to update your firmware on Robomaster EP or TT.