• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1#!/usr/bin/env python3
2# coding=utf-8
3
4#
5# Copyright (c) 2022 Huawei Device Co., Ltd.
6# Licensed under the Apache License, Version 2.0 (the "License");
7# you may not use this file except in compliance with the License.
8# You may obtain a copy of the License at
9#
10#     http://www.apache.org/licenses/LICENSE-2.0
11#
12# Unless required by applicable law or agreed to in writing, software
13# distributed under the License is distributed on an "AS IS" BASIS,
14# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15# See the License for the specific language governing permissions and
16# limitations under the License.
17#
18
19import os
20import platform
21import socket
22import struct
23import threading
24import time
25import shutil
26import stat
27from dataclasses import dataclass
28
29from xdevice import DeviceOsType
30from xdevice import ReportException
31from xdevice import ExecuteTerminate
32from xdevice import platform_logger
33from xdevice import Plugin
34from xdevice import get_plugin
35from xdevice import IShellReceiver
36from xdevice import exec_cmd
37from xdevice import get_file_absolute_path
38from xdevice import FilePermission
39from xdevice import DeviceError
40from xdevice import HdcError
41from xdevice import HdcCommandRejectedException
42from xdevice import ShellCommandUnresponsiveException
43from xdevice import DeviceState
44from xdevice import convert_serial
45from xdevice import is_proc_running
46from xdevice import convert_ip
47from xdevice import create_dir
48
49ID_OKAY = b'OKAY'
50ID_FAIL = b'FAIL'
51ID_STAT = b'STAT'
52ID_RECV = b'RECV'
53ID_DATA = b'DATA'
54ID_DONE = b'DONE'
55ID_SEND = b'SEND'
56ID_LIST = b'LIST'
57ID_DENT = b'DENT'
58
59DEFAULT_ENCODING = "ISO-8859-1"
60SYNC_DATA_MAX = 64 * 1024
61REMOTE_PATH_MAX_LENGTH = 1024
62SOCK_DATA_MAX = 256
63
64INSTALL_TIMEOUT = 2 * 60 * 1000
65DEFAULT_TIMEOUT = 40 * 1000
66
67MAX_CONNECT_ATTEMPT_COUNT = 10
68DATA_UNIT_LENGTH = 4
69HEXADECIMAL_NUMBER = 16
70SPECIAL_FILE_MODE = 41471
71FORMAT_BYTES_LENGTH = 4
72DEFAULT_OFFSET_OF_INT = 4
73
74INVALID_MODE_CODE = -1
75DEFAULT_STD_PORT = 8710
76HDC_NAME = "hdc"
77HDC_STD_NAME = "hdc_std"
78LOG = platform_logger("Hdc")
79
80
81class HdcMonitor:
82    """
83    A Device monitor.
84    This monitor connects to the Device Connector, gets device and
85    debuggable process information from it.
86    """
87    MONITOR_MAP = {}
88
89    def __init__(self, host="127.0.0.1", port=None, device_connector=None):
90        self.channel = dict()
91        self.channel.setdefault("host", host)
92        self.channel.setdefault("port", port)
93        self.main_hdc_connection = None
94        self.connection_attempt = 0
95        self.is_stop = False
96        self.monitoring = False
97        self.server = device_connector
98        self.devices = []
99        self.last_msg_len = 0
100        self.changed = True
101
102    @staticmethod
103    def get_instance(host, port=None, device_connector=None):
104        if host not in HdcMonitor.MONITOR_MAP:
105            monitor = HdcMonitor(host, port, device_connector)
106            HdcMonitor.MONITOR_MAP[host] = monitor
107            LOG.debug("HdcMonitor map add host %s, map is %s" %
108                      (host, HdcMonitor.MONITOR_MAP))
109
110        return HdcMonitor.MONITOR_MAP[host]
111
112    def start(self):
113        """
114        Starts the monitoring.
115        """
116        try:
117            server_thread = threading.Thread(target=self.loop_monitor,
118                                             name="HdcMonitor", args=())
119            server_thread.setDaemon(True)
120            server_thread.start()
121        except FileNotFoundError as _:
122            LOG.error("HdcMonitor can't find connector, init device "
123                      "environment failed!")
124
125    def init_hdc(self, connector_name=HDC_NAME):
126        env_hdc = shutil.which(connector_name)
127        # if not, add xdevice's own hdc path to environ path.
128        # tell if hdc has already been in the environ path.
129        if env_hdc is None:
130            LOG.error("Can not find {} or {} environment variable, "
131                      "please set it first!".format(HDC_NAME, HDC_STD_NAME))
132        if not is_proc_running(connector_name):
133            port = DEFAULT_STD_PORT
134            self.start_hdc(
135                connector=connector_name,
136                local_port=self.channel.setdefault(
137                    "port", port))
138            time.sleep(1)
139
140    def _init_hdc_connection(self):
141        if self.main_hdc_connection is not None:
142            return
143        # set all devices disconnect
144        devices = [item for item in self.devices]
145        devices.reverse()
146        for local_device1 in devices:
147            local_device1.device_state = DeviceState.OFFLINE
148            self.server.device_changed(local_device1)
149
150        connector_name = HDC_STD_NAME if HdcHelper.is_hdc_std() else HDC_NAME
151        self.init_hdc(connector_name)
152        self.connection_attempt = 0
153        self.monitoring = False
154        while self.main_hdc_connection is None:
155            self.main_hdc_connection = self.open_hdc_connection()
156            if self.main_hdc_connection is None:
157                self.connection_attempt += 1
158                if self.connection_attempt > MAX_CONNECT_ATTEMPT_COUNT:
159                    self.is_stop = True
160                    LOG.error(
161                        "HdcMonitor attempt %s, can't connect to hdc "
162                        "for Device List Monitoring" %
163                        str(self.connection_attempt))
164                    raise HdcError(
165                        "HdcMonitor cannot connect hdc server(%s %s),"
166                        " please check!" %
167                        (self.channel.get("host"),
168                         str(self.channel.get("post"))))
169
170                LOG.debug(
171                    "HdcMonitor Connection attempts: %s" %
172                    str(self.connection_attempt))
173                time.sleep(2)
174
175    def stop(self):
176        """
177        Stops the monitoring.
178        """
179        for host in HdcMonitor.MONITOR_MAP:
180            LOG.debug("HdcMonitor stop host %s" % host)
181            monitor = HdcMonitor.MONITOR_MAP[host]
182            try:
183                monitor.is_stop = True
184                if monitor.main_hdc_connection is not None:
185                    monitor.main_hdc_connection.shutdown(2)
186                    monitor.main_hdc_connection.close()
187                    monitor.main_hdc_connection = None
188            except (socket.error, socket.gaierror, socket.timeout) as _:
189                LOG.error("HdcMonitor close socket exception")
190        HdcMonitor.MONITOR_MAP.clear()
191        LOG.debug("HdcMonitor {} monitor stop!".format(HdcHelper.CONNECTOR_NAME))
192        LOG.debug("HdcMonitor map is %s" % HdcMonitor.MONITOR_MAP)
193
194    def loop_monitor(self):
195        """
196        Monitors the devices. This connects to the Debug Bridge
197        """
198        LOG.debug("current connector name is %s" % HdcHelper.CONNECTOR_NAME)
199        while not self.is_stop:
200            try:
201                if self.main_hdc_connection is None:
202                    self._init_hdc_connection()
203                    if self.main_hdc_connection is not None:
204                        LOG.debug(
205                            "HdcMonitor Connected to hdc for device "
206                            "monitoring, main_hdc_connection is %s" %
207                            self.main_hdc_connection)
208
209                self.list_targets()
210                time.sleep(1)
211            except (HdcError, Exception) as _:
212                self.handle_exception_monitor_loop()
213                time.sleep(2)
214
215    def handle_exception_monitor_loop(self):
216        LOG.debug("Handle exception monitor loop: %s" %
217                  self.main_hdc_connection)
218        if self.main_hdc_connection is None:
219            return
220        LOG.debug("Handle exception monitor loop, main hdc connection closed, "
221                  "main hdc connection: %s" % self.main_hdc_connection)
222        self.main_hdc_connection.close()
223        self.main_hdc_connection = None
224
225    def _get_device_instance(self, items, os_type):
226        device = get_plugin(plugin_type=Plugin.DEVICE, plugin_id=os_type)[0]
227        device_instance = device.__class__()
228        device_instance.__set_serial__(items[0])
229        device_instance.host = self.channel.get("host")
230        device_instance.port = self.channel.get("port")
231        if self.changed:
232            LOG.debug("Dmlib get device instance %s %s %s" %
233                      (device_instance.device_sn,
234                       device_instance.host, device_instance.port))
235        device_instance.device_state = DeviceState.get_state(items[3])
236        return device_instance
237
238    def update_devices(self, param_array_list):
239        devices = [item for item in self.devices]
240        devices.reverse()
241        for local_device1 in devices:
242            k = 0
243            for local_device2 in param_array_list:
244                if local_device1.device_sn == local_device2.device_sn and \
245                        local_device1.device_os_type == \
246                        local_device2.device_os_type:
247                    k = 1
248                    if local_device1.device_state != \
249                            local_device2.device_state:
250                        local_device1.device_state = local_device2.device_state
251                        self.server.device_changed(local_device1)
252                    param_array_list.remove(local_device2)
253                    break
254
255            if k == 0:
256                self.devices.remove(local_device1)
257                self.server.device_disconnected(local_device1)
258        for local_device in param_array_list:
259            self.devices.append(local_device)
260            self.server.device_connected(local_device)
261
262    def open_hdc_connection(self):
263        """
264        Attempts to connect to the debug bridge server. Return a connect socket
265        if success, null otherwise.
266        """
267        sock = None
268        try:
269            LOG.debug(
270                "HdcMonitor connecting to hdc for Device List Monitoring")
271            LOG.debug("HdcMonitor socket connection host: %s, port: %s" %
272                      (str(convert_ip(self.channel.get("host"))),
273                       str(int(self.channel.get("port")))))
274
275            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
276            sock.connect((self.channel.get("host"),
277                          int(self.channel.get("port"))))
278            return sock
279
280        except (socket.error, socket.gaierror, socket.timeout) as exception:
281            LOG.error("HdcMonitor hdc socket connection Error: %s, "
282                      "host is %s, port is %s" % (str(exception),
283                                                  self.channel.get("host"),
284                                                  self.channel.get("port")))
285            return None
286
287    def start_hdc(self, connector=HDC_NAME, kill=False, local_port=None):
288        """Starts the hdc host side server.
289
290        Args:
291            connector: connector type, like "hdc"
292            kill: if True, kill exist host side server
293            local_port: local port to start host side server
294
295        Returns:
296            None
297        """
298        if kill:
299            LOG.debug("HdcMonitor {} kill".format(connector))
300            exec_cmd([connector, "kill"])
301        LOG.debug("HdcMonitor {} start".format(connector))
302        exec_cmd(
303            [connector, "-l5", "start"],
304            error_print=False, redirect=True)
305
306    def list_targets(self):
307        if self.main_hdc_connection:
308            self.server.monitor_lock.acquire(timeout=1)
309            try:
310                self.monitoring_list_targets()
311                len_buf = HdcHelper.read(self.main_hdc_connection,
312                                         DATA_UNIT_LENGTH)
313                length = struct.unpack("!I", len_buf)[0]
314                if length >= 0:
315                    if self.last_msg_len != length:
316                        LOG.debug("had received length is: %s" % length)
317                        self.last_msg_len = length
318                        self.changed = True
319                    else:
320                        self.changed = False
321                    self.connection_attempt = 0
322                    self.process_incoming_target_data(length)
323            except Exception as e:
324                LOG.error(e)
325                raise e
326            finally:
327                self.server.monitor_lock.release()
328
329    def monitoring_list_targets(self):
330        if not self.monitoring:
331            HdcHelper.handle_shake(self.main_hdc_connection)
332            request = HdcHelper.form_hdc_request("alive")
333            HdcHelper.write(self.main_hdc_connection, request)
334            self.monitoring = True
335        request = HdcHelper.form_hdc_request('list targets -v')
336        HdcHelper.write(self.main_hdc_connection, request)
337
338    def process_incoming_target_data(self, length):
339        local_array_list = []
340        data_buf = HdcHelper.read(self.main_hdc_connection, length)
341        data_str = HdcHelper.reply_to_string(data_buf)
342        if 'Empty' not in data_str:
343            lines = data_str.split('\n')
344            for line in lines:
345                items = line.strip().split('\t')
346                # Example: sn    USB     Offline localhost       hdc
347                if not items[0] or len(items) < 5:
348                    continue
349                device_instance = self._get_device_instance(
350                    items, DeviceOsType.default)
351                local_array_list.append(device_instance)
352        else:
353            if self.changed:
354                LOG.debug("please check device actually.[%s]" % data_str.strip())
355        self.update_devices(local_array_list)
356
357    @staticmethod
358    def peek_hdc():
359        LOG.debug("Peek running process to check expect connector.")
360        # if not find hdc_std, try find hdc
361        connector_name = HDC_STD_NAME
362        env_hdc = shutil.which(connector_name)
363        # if not, add xdevice's own hdc path to environ path.
364        # tell if hdc has already been in the environ path.
365        if env_hdc is None:
366            connector_name = HDC_NAME
367        LOG.debug("Peak end")
368        return connector_name
369
370
371@dataclass
372class HdcResponse:
373    """Response from HDC."""
374    okay = ID_OKAY  # first 4 bytes in response were "OKAY"?
375    message = ""  # diagnostic string if okay is false
376
377
378class SyncService:
379    """
380    Sync service class to push/pull to/from devices/emulators,
381    through the debug bridge.
382    """
383
384    def __init__(self, device, host=None, port=None):
385        self.device = device
386        self.host = host
387        self.port = port
388        self.sock = None
389
390    def open_sync(self, timeout=DEFAULT_TIMEOUT):
391        """
392        Opens the sync connection. This must be called before any calls to
393        push[File] / pull[File].
394        Return true if the connection opened, false if hdc refuse the
395        connection. This can happen device is invalid.
396        """
397        LOG.debug("Open sync, timeout=%s" % int(timeout / 1000))
398        self.sock = HdcHelper.socket(host=self.host, port=self.port,
399                                     timeout=timeout)
400        HdcHelper.set_device(self.device, self.sock)
401
402        request = HdcHelper.form_hdc_request("sync:")
403        HdcHelper.write(self.sock, request)
404
405        resp = HdcHelper.read_hdc_response(self.sock)
406        if not resp.okay:
407            self.device.log.error(
408                "Got unhappy response from HDC sync req: %s" % resp.message)
409            raise HdcError(
410                "Got unhappy response from HDC sync req: %s" % resp.message)
411
412    def close(self):
413        """
414        Closes the connection.
415        """
416        if self.sock is not None:
417            try:
418                self.sock.close()
419            except socket.error as error:
420                LOG.error("Socket close error: %s" % error, error_no="00420")
421            finally:
422                self.sock = None
423
424    def pull_file(self, remote, local, is_create=False):
425        """
426        Pulls a file.
427        The top directory won't be created if is_create is False (by default)
428        and vice versa
429        """
430        mode = self.read_mode(remote)
431        self.device.log.debug("Remote file %s mode is %d" % (remote, mode))
432        if mode == 0:
433            raise HdcError("Remote object doesn't exist!")
434
435        if str(mode).startswith("168"):
436            if is_create:
437                remote_file_split = os.path.split(remote)[-1] \
438                    if os.path.split(remote)[-1] else os.path.split(remote)[-2]
439                remote_file_basename = os.path.basename(remote_file_split)
440                new_local = os.path.join(local, remote_file_basename)
441                create_dir(new_local)
442            else:
443                new_local = local
444
445            collect_receiver = CollectingOutputReceiver()
446            HdcHelper.execute_shell_command(self.device, "ls %s" % remote,
447                                            receiver=collect_receiver)
448            files = collect_receiver.output.split()
449            for file_name in files:
450                self.pull_file("%s/%s" % (remote, file_name),
451                               new_local, is_create=True)
452        elif mode == SPECIAL_FILE_MODE:
453            self.device.log.info("skipping special file '%s'" % remote)
454        else:
455            if os.path.isdir(local):
456                local = os.path.join(local, os.path.basename(remote))
457
458            self.do_pull_file(remote, local)
459
460    def do_pull_file(self, remote, local):
461        """
462        Pulls a remote file
463        """
464        self.device.log.info(
465            "%s pull %s to %s" % (convert_serial(self.device.device_sn),
466                                  remote, local))
467        remote_path_content = remote.encode(DEFAULT_ENCODING)
468        if len(remote_path_content) > REMOTE_PATH_MAX_LENGTH:
469            raise HdcError("Remote path is too long.")
470
471        msg = self.create_file_req(ID_RECV, remote_path_content)
472        HdcHelper.write(self.sock, msg)
473        pull_result = HdcHelper.read(self.sock, DATA_UNIT_LENGTH * 2)
474        if not self.check_result(pull_result, ID_DATA) and \
475                not self.check_result(pull_result, ID_DONE):
476            raise HdcError(self.read_error_message(pull_result))
477        if platform.system() == "Windows":
478            flags = os.O_WRONLY | os.O_CREAT | os.O_APPEND | os.O_BINARY
479        else:
480            flags = os.O_WRONLY | os.O_CREAT | os.O_APPEND
481        pulled_file_open = os.open(local, flags, FilePermission.mode_755)
482        with os.fdopen(pulled_file_open, "wb") as pulled_file:
483            while True:
484                if self.check_result(pull_result, ID_DONE):
485                    break
486
487                if not self.check_result(pull_result, ID_DATA):
488                    raise HdcError(self.read_error_message(pull_result))
489
490                try:
491                    length = self.swap32bit_from_array(
492                        pull_result, DEFAULT_OFFSET_OF_INT)
493                except IndexError as index_error:
494                    self.device.log.debug("do_pull_file: %s" %
495                                          str(pull_result))
496                    if pull_result == ID_DATA:
497                        pull_result = self.sock.recv(DATA_UNIT_LENGTH)
498                        self.device.log.debug(
499                            "do_pull_file: %s" % str(pull_result))
500                        length = self.swap32bit_from_array(pull_result, 0)
501                        self.device.log.debug("do_pull_file: %s" % str(length))
502                    else:
503                        raise IndexError(str(index_error)) from index_error
504
505                if length > SYNC_DATA_MAX:
506                    raise HdcError("Receiving too much data.")
507
508                pulled_file.write(HdcHelper.read(self.sock, length))
509                pulled_file.flush()
510                pull_result = self.sock.recv(DATA_UNIT_LENGTH * 2)
511
512    def push_file(self, local, remote, is_create=False):
513        """
514        Push a single file.
515        The top directory won't be created if is_create is False (by default)
516        and vice versa
517        """
518        if not os.path.exists(local):
519            raise HdcError("Local path doesn't exist.")
520
521        if os.path.isdir(local):
522            if is_create:
523                local_file_split = os.path.split(local)[-1] \
524                    if os.path.split(local)[-1] else os.path.split(local)[-2]
525                local_file_basename = os.path.basename(local_file_split)
526                remote = "{}/{}".format(
527                    remote, local_file_basename)
528                HdcHelper.execute_shell_command(
529                    self.device, "mkdir -p %s" % remote)
530
531            for child in os.listdir(local):
532                file_path = os.path.join(local, child)
533                if os.path.isdir(file_path):
534                    self.push_file(
535                        file_path,  "%s/%s" % (remote, child),
536                        is_create=False)
537                else:
538                    self.do_push_file(file_path, "%s/%s" % (remote, child))
539        else:
540            self.do_push_file(local, remote)
541
542    def do_push_file(self, local, remote):
543        """
544        Push a single file
545
546        Args:
547        ------------
548        local : string
549            the local file to push
550        remote : string
551            the remote file (length max is 1024)
552        """
553        mode = self.read_mode(remote)
554        self.device.log.debug("Remote file %s mode is %d" % (remote, mode))
555        self.device.log.debug("%s execute command: hdc push %s %s" % (
556            convert_serial(self.device.device_sn), local, remote))
557        if str(mode).startswith("168"):
558            remote = "%s/%s" % (remote, os.path.basename(local))
559
560        try:
561            try:
562                remote_path_content = remote.encode(DEFAULT_ENCODING)
563            except UnicodeEncodeError as _:
564                remote_path_content = remote.encode("UTF-8")
565            if len(remote_path_content) > REMOTE_PATH_MAX_LENGTH:
566                raise HdcError("Remote path is too long.")
567
568            # create the header for the action
569            # and send it. We use a custom try/catch block to make the
570            # difference between file and network IO exceptions.
571            msg = self.create_send_file_req(ID_SEND, remote_path_content,
572                                            FilePermission.mode_644)
573
574            HdcHelper.write(self.sock, msg)
575            flags = os.O_RDONLY
576            modes = stat.S_IWUSR | stat.S_IRUSR
577            with os.fdopen(os.open(local, flags, modes), "rb") as test_file:
578                while True:
579                    if platform.system() == "Linux":
580                        data = test_file.read(1024 * 4)
581                    else:
582                        data = test_file.read(SYNC_DATA_MAX)
583
584                    if not data:
585                        break
586
587                    buf = struct.pack(
588                        "%ds%ds%ds" % (len(ID_DATA), FORMAT_BYTES_LENGTH,
589                                       len(data)), ID_DATA,
590                        self.swap32bits_to_bytes(len(data)), data)
591                    self.sock.send(buf)
592        except Exception as exception:
593            self.device.log.error("exception %s" % exception)
594            raise exception
595
596        msg = self.create_req(ID_DONE, int(time.time()))
597        HdcHelper.write(self.sock, msg)
598        result = HdcHelper.read(self.sock, DATA_UNIT_LENGTH * 2)
599        if not self.check_result(result, ID_OKAY):
600            self.device.log.error("exception %s" % result)
601            raise HdcError(self.read_error_message(result))
602
603    def read_mode(self, path):
604        """
605        Returns the mode of the remote file.
606        Return an Integer containing the mode if all went well or null
607        """
608        msg = self.create_file_req(ID_STAT, path)
609        HdcHelper.write(self.sock, msg)
610
611        # read the result, in a byte array containing 4 ints
612        stat_result = HdcHelper.read(self.sock, DATA_UNIT_LENGTH * 4)
613        if not self.check_result(stat_result, ID_STAT):
614            return INVALID_MODE_CODE
615
616        return self.swap32bit_from_array(stat_result, DEFAULT_OFFSET_OF_INT)
617
618    def create_file_req(self, command, path):
619        """
620        Creates the data array for a file request. This creates an array with a
621        4 byte command + the remote file name.
622
623        Args:
624        ------------
625        command :
626            the 4 byte command (ID_STAT, ID_RECV, ...)
627        path : string
628            The path, as a byte array, of the remote file on which to execute
629            the command.
630
631        return:
632        ------------
633            return the byte[] to send to the device through hdc
634        """
635        if isinstance(path, str):
636            try:
637                path = path.encode(DEFAULT_ENCODING)
638            except UnicodeEncodeError as _:
639                path = path.encode("UTF-8")
640
641        return struct.pack(
642            "%ds%ds%ds" % (len(command), FORMAT_BYTES_LENGTH, len(path)),
643            command, self.swap32bits_to_bytes(len(path)), path)
644
645    def create_send_file_req(self, command, path, mode=0o644):
646        # make the mode into a string
647        mode_str = ",%s" % str(mode & FilePermission.mode_777)
648        mode_content = mode_str.encode(DEFAULT_ENCODING)
649        return struct.pack(
650            "%ds%ds%ds%ds" % (len(command), FORMAT_BYTES_LENGTH, len(path),
651                              len(mode_content)),
652            command, self.swap32bits_to_bytes(len(path) + len(mode_content)),
653            path, mode_content)
654
655    def create_req(self, command, value):
656        """
657        Create a command with a code and an int values
658        """
659        return struct.pack("%ds%ds" % (len(command), FORMAT_BYTES_LENGTH),
660                           command, self.swap32bits_to_bytes(value))
661
662    @staticmethod
663    def check_result(result, code):
664        """
665        Checks the result array starts with the provided code
666
667        Args:
668        ------------
669        result :
670            the result array to check
671        path : string
672            the 4 byte code
673
674        return:
675        ------------
676        bool
677            return true if the code matches
678        """
679        return result[0:4] == code[0:4]
680
681    def read_error_message(self, result):
682        """
683        Reads an error message from the opened Socket.
684
685        Args:
686        ------------
687        result :
688            the current hdc result. Must contain both FAIL and the length of
689            the message.
690        """
691        if self.check_result(result, ID_FAIL):
692            length = self.swap32bit_from_array(result, 4)
693            if length > 0:
694                return str(HdcHelper.read(self.sock, length))
695
696        return None
697
698    @staticmethod
699    def swap32bits_to_bytes(value):
700        """
701        Swaps an unsigned value around, and puts the result in an bytes that
702        can be sent to a device.
703
704        Args:
705        ------------
706        value :
707            the value to swap.
708        """
709        return bytes([value & 0x000000FF,
710                      (value & 0x0000FF00) >> 8,
711                      (value & 0x00FF0000) >> 16,
712                      (value & 0xFF000000) >> 24])
713
714    @staticmethod
715    def swap32bit_from_array(value, offset):
716        """
717        Reads a signed 32 bit integer from an array coming from a device.
718
719        Args:
720        ------------
721        value :
722            the array containing the int
723        offset:
724            the offset in the array at which the int starts
725
726        Return:
727        ------------
728        int
729            the integer read from the array
730        """
731        result = 0
732        result |= (int(value[offset])) & 0x000000FF
733        result |= (int(value[offset + 1]) & 0x000000FF) << 8
734        result |= (int(value[offset + 2]) & 0x000000FF) << 16
735        result |= (int(value[offset + 3]) & 0x000000FF) << 24
736
737        return result
738
739
740class HdcHelper:
741    CONNECTOR_NAME = ""
742
743    @staticmethod
744    def check_if_hdc_running(timeout=30):
745        LOG.debug("Check if {} is running, timeout is {}s".format(
746            HdcHelper.CONNECTOR_NAME, timeout))
747        index = 1
748        while index < timeout:
749            if is_proc_running(HdcHelper.CONNECTOR_NAME):
750                return True
751            index = index + 1
752            time.sleep(1)
753        return False
754
755    @staticmethod
756    def push_file(device, local, remote, is_create=False,
757                  timeout=DEFAULT_TIMEOUT):
758        device.log.info("{} execute command: {} file send {} to {}".format(
759            convert_serial(device.device_sn), HdcHelper.CONNECTOR_NAME, local, remote))
760        HdcHelper._operator_file("file send", device, local, remote, timeout)
761
762    @staticmethod
763    def pull_file(device, remote, local, is_create=False,
764                  timeout=DEFAULT_TIMEOUT):
765        device.log.info("{} execute command: {} file recv {} to {}".format(
766            convert_serial(device.device_sn), HdcHelper.CONNECTOR_NAME, remote, local))
767        HdcHelper._operator_file("file recv", device, remote, local, timeout)
768
769    @staticmethod
770    def _install_remote_package(device, remote_file_path, command):
771        receiver = CollectingOutputReceiver()
772        cmd = "bm install %s %s" % (command.strip(), remote_file_path)
773        HdcHelper.execute_shell_command(device, cmd, INSTALL_TIMEOUT, receiver)
774        return receiver.output
775
776    @staticmethod
777    def install_package(device, package_file_path, command):
778        device.log.info("%s install %s" % (convert_serial(device.device_sn),
779                                           package_file_path))
780        remote_file_path = "/data/local/tmp/%s" % os.path.basename(
781            package_file_path)
782
783        service = None
784        try:
785            service = SyncService(device, host=device.host, port=device.port)
786            service.open_sync()
787            service.push_file(package_file_path, remote_file_path)
788        finally:
789            if service is not None:
790                service.close()
791
792        result = HdcHelper._install_remote_package(device, remote_file_path,
793                                                   command)
794        HdcHelper.execute_shell_command(device, "rm %s " % remote_file_path)
795        return result
796
797    @staticmethod
798    def uninstall_package(device, package_name):
799        receiver = CollectingOutputReceiver()
800        command = "bm uninstall -n %s " % package_name
801        device.log.info("%s %s" % (convert_serial(device.device_sn), command))
802        HdcHelper.execute_shell_command(device, command, INSTALL_TIMEOUT,
803                                        receiver)
804        return receiver.output
805
806    @staticmethod
807    def reboot(device, into=None):
808        device.log.info("{} execute command: {} target boot".format(convert_serial(device.device_sn),
809                                                                    HdcHelper.CONNECTOR_NAME))
810        with HdcHelper.socket(host=device.host, port=device.port) as sock:
811            HdcHelper.handle_shake(sock, device.device_sn)
812            request = HdcHelper.form_hdc_request("target boot")
813            HdcHelper.write(sock, request)
814
815    @staticmethod
816    def execute_shell_command(device, command, timeout=DEFAULT_TIMEOUT,
817                              receiver=None, **kwargs):
818        """
819        Executes a shell command on the device and retrieve the output.
820
821        Args:
822        ------------
823        device : IDevice
824            on which to execute the command.
825        command : string
826            the shell command to execute
827        timeout : int
828            max time between command output. If more time passes between
829            command output, the method will throw
830            ShellCommandUnresponsiveException (ms).
831        """
832        try:
833            if not timeout:
834                timeout = DEFAULT_TIMEOUT
835
836            with HdcHelper.socket(host=device.host, port=device.port,
837                                  timeout=timeout) as sock:
838                output_flag = kwargs.get("output_flag", True)
839                timeout_msg = " with timeout %ss" % str(timeout / 1000)
840                message = "{} execute command: {} shell {}{}".format(convert_serial(device.device_sn),
841                                                                     HdcHelper.CONNECTOR_NAME,
842                                                                     command, timeout_msg)
843                if output_flag:
844                    LOG.info(message)
845                else:
846                    LOG.debug(message)
847                from xdevice import Scheduler
848                HdcHelper.handle_shake(sock, device.device_sn)
849                request = HdcHelper.form_hdc_request("shell {}".format(command))
850                HdcHelper.write(sock, request)
851                resp = HdcResponse()
852                resp.okay = True
853                while True:
854                    len_buf = sock.recv(DATA_UNIT_LENGTH)
855                    if len_buf:
856                        length = struct.unpack("!I", len_buf)[0]
857                    else:
858                        break
859                    data = sock.recv(length)
860                    ret = HdcHelper.reply_to_string(data)
861                    if ret:
862                        if receiver:
863                            receiver.__read__(ret)
864                        else:
865                            LOG.debug(ret)
866                    if not Scheduler.is_execute:
867                        raise ExecuteTerminate()
868                return resp
869        except socket.timeout as error:
870            device.log.error("ShellCommandUnresponsiveException: {} shell {} timeout[{}S]".format(
871                convert_serial(device.device_sn), command, str(timeout / 1000)))
872            raise ShellCommandUnresponsiveException() from error
873        finally:
874            if receiver:
875                receiver.__done__()
876
877    @staticmethod
878    def set_device(device, sock):
879        """
880        Tells hdc to talk to a specific device
881        if the device is not -1, then we first tell hdc we're looking to talk
882        to a specific device
883        """
884        msg = "host:transport:%s" % device.device_sn
885        device_query = HdcHelper.form_hdc_request(msg)
886        HdcHelper.write(sock, device_query)
887        resp = HdcHelper.read_hdc_response(sock)
888        if not resp.okay:
889            raise HdcCommandRejectedException(resp.message)
890
891    @staticmethod
892    def form_hdc_request(req):
893        """
894        Create an ASCII string preceded by four hex digits.
895        """
896        try:
897            if not req.endswith('\0'):
898                req = "%s\0" % req
899            req = req.encode("utf-8")
900            fmt = "!I%ss" % len(req)
901            result = struct.pack(fmt, len(req), req)
902        except UnicodeEncodeError as ex:
903            LOG.error(ex)
904            raise ex
905        return result
906
907    @staticmethod
908    def read_hdc_response(sock, read_diag_string=False):
909        """
910        Reads the response from HDC after a command.
911
912        Args:
913        ------------
914        read_diag_string :
915            If true, we're expecting an OKAY response to be followed by a
916            diagnostic string. Otherwise, we only expect the diagnostic string
917            to follow a FAIL.
918        """
919        resp = HdcResponse()
920        reply = HdcHelper.read(sock, DATA_UNIT_LENGTH)
921        if HdcHelper.is_okay(reply):
922            resp.okay = True
923        else:
924            read_diag_string = True
925            resp.okay = False
926
927        while read_diag_string:
928            len_buf = HdcHelper.read(sock, DATA_UNIT_LENGTH)
929            len_str = HdcHelper.reply_to_string(len_buf)
930            msg = HdcHelper.read(sock, int(len_str, HEXADECIMAL_NUMBER))
931            resp.message = HdcHelper.reply_to_string(msg)
932            break
933
934        return resp
935
936    @staticmethod
937    def write(sock, req, timeout=10):
938        if isinstance(req, str):
939            req = req.encode(DEFAULT_ENCODING)
940        elif isinstance(req, list):
941            req = bytes(req)
942
943        start_time = time.time()
944        while req:
945            if time.time() - start_time > timeout:
946                LOG.debug("Socket write timeout, timeout:%ss" % timeout)
947                break
948
949            size = sock.send(req)
950            if size < 0:
951                raise DeviceError("channel EOF")
952
953            req = req[size:]
954            time.sleep(5 / 1000)
955
956    @staticmethod
957    def read(sock, length, timeout=10):
958        data = b''
959        recv_len = 0
960        start_time = time.time()
961        exc_num = 3
962        while length - recv_len > 0:
963            if time.time() - start_time > timeout:
964                LOG.debug("Socket read timeout, timout:%ss" % timeout)
965                break
966            try:
967                recv = sock.recv(length - recv_len)
968                if len(recv) > 0:
969                    time.sleep(5 / 1000)
970                else:
971                    break
972            except ConnectionResetError as error:
973                if exc_num <= 0:
974                    raise error
975                exc_num = exc_num - 1
976                recv = b''
977                time.sleep(1)
978                LOG.debug("ConnectionResetError occurs")
979
980            data += recv
981            recv_len += len(recv)
982
983        return data
984
985    @staticmethod
986    def is_okay(reply):
987        """
988        Checks to see if the first four bytes in "reply" are OKAY.
989        """
990        return reply[0:4] == ID_OKAY
991
992    @staticmethod
993    def reply_to_string(reply):
994        """
995        Converts an HDC reply to a string.
996        """
997        try:
998            return str(reply, encoding=DEFAULT_ENCODING)
999        except (ValueError, TypeError) as _:
1000            return ""
1001
1002    @staticmethod
1003    def socket(host=None, port=None, timeout=None):
1004        end = time.time() + 10 * 60
1005        sock = None
1006        hdc_connection = HdcMonitor.MONITOR_MAP.get(host, "127.0.0.1")
1007        while host not in HdcMonitor.MONITOR_MAP or \
1008                hdc_connection.main_hdc_connection is None:
1009            LOG.debug("Host: %s, port: %s, HdcMonitor map is %s" % (
1010                host, port, HdcMonitor.MONITOR_MAP))
1011            if host in HdcMonitor.MONITOR_MAP:
1012                LOG.debug("Monitor main hdc connection is %s" %
1013                          hdc_connection.main_hdc_connection)
1014            if time.time() > end:
1015                raise HdcError("Cannot detect HDC monitor!")
1016            time.sleep(2)
1017
1018        try:
1019            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
1020            sock.connect((host, int(port)))
1021        except socket.error as exception:
1022            LOG.exception("Connect hdc server error: %s" % str(exception),
1023                          exc_info=False)
1024            raise exception
1025
1026        if sock is None:
1027            raise HdcError("Cannot connect hdc server!")
1028
1029        if timeout is not None:
1030            sock.setblocking(False)
1031            sock.settimeout(timeout / 1000)
1032
1033        return sock
1034
1035    @staticmethod
1036    def handle_shake(connection, connect_key=""):
1037        reply = HdcHelper.read(connection, 48)
1038        struct.unpack(">I12s32s", reply)
1039        banner_str = b'OHOS HDC'
1040        connect_key = connect_key.encode("utf-8")
1041        size = struct.calcsize('12s256s')
1042        fmt = "!I12s256s"
1043        pack_cmd = struct.pack(fmt, size, banner_str, connect_key)
1044        HdcHelper.write(connection, pack_cmd)
1045        return True
1046
1047    @staticmethod
1048    def _operator_file(command, device, local, remote, timeout):
1049        sock = HdcHelper.socket(host=device.host, port=device.port,
1050                                timeout=timeout)
1051        HdcHelper.handle_shake(sock, device.device_sn)
1052        request = HdcHelper.form_hdc_request(
1053            "%s %s %s" % (command, local, remote))
1054        HdcHelper.write(sock, request)
1055        reply = HdcHelper.read(sock, DATA_UNIT_LENGTH)
1056        length = struct.unpack("!I", reply)[0]
1057        data_buf = HdcHelper.read(sock, length)
1058        HdcHelper.reply_to_string(data_buf)
1059        LOG.debug("result %s" % data_buf)
1060
1061    @staticmethod
1062    def is_hdc_std():
1063        return HDC_STD_NAME in HdcHelper.CONNECTOR_NAME
1064
1065
1066class DeviceConnector(object):
1067    __instance = None
1068    __init_flag = False
1069
1070    def __init__(self, host=None, port=None, usb_type=None):
1071        if DeviceConnector.__init_flag:
1072            return
1073        self.device_listeners = []
1074        self.device_monitor = None
1075        self.monitor_lock = threading.Condition()
1076        self.host = host if host else "127.0.0.1"
1077        self.usb_type = usb_type
1078        connector_name = HdcMonitor.peek_hdc()
1079        HdcHelper.CONNECTOR_NAME = connector_name
1080        if port:
1081            self.port = int(port)
1082        else:
1083            self.port = int(os.getenv("OHOS_HDC_SERVER_PORT", DEFAULT_STD_PORT))
1084
1085    def start(self):
1086        self.device_monitor = HdcMonitor.get_instance(
1087            self.host, self.port, device_connector=self)
1088        self.device_monitor.start()
1089
1090    def terminate(self):
1091        if self.device_monitor:
1092            self.device_monitor.stop()
1093        self.device_monitor = None
1094
1095    def add_device_change_listener(self, device_change_listener):
1096        self.device_listeners.append(device_change_listener)
1097
1098    def remove_device_change_listener(self, device_change_listener):
1099        if device_change_listener in self.device_listeners:
1100            self.device_listeners.remove(device_change_listener)
1101
1102    def device_connected(self, device):
1103        LOG.debug("DeviceConnector device connected:host %s, port %s, "
1104                  "device sn %s " % (self.host, self.port, device.device_sn))
1105        if device.host != self.host or device.port != self.port:
1106            LOG.debug("DeviceConnector device error")
1107        for listener in self.device_listeners:
1108            listener.device_connected(device)
1109
1110    def device_disconnected(self, device):
1111        LOG.debug("DeviceConnector device disconnected:host %s, port %s, "
1112                  "device sn %s" % (self.host, self.port, device.device_sn))
1113        if device.host != self.host or device.port != self.port:
1114            LOG.debug("DeviceConnector device error")
1115        for listener in self.device_listeners:
1116            listener.device_disconnected(device)
1117
1118    def device_changed(self, device):
1119        LOG.debug("DeviceConnector device changed:host %s, port %s, "
1120                  "device sn %s" % (self.host, self.port, device.device_sn))
1121        if device.host != self.host or device.port != self.port:
1122            LOG.debug("DeviceConnector device error")
1123        for listener in self.device_listeners:
1124            listener.device_changed(device)
1125
1126
1127class CollectingOutputReceiver(IShellReceiver):
1128    def __init__(self):
1129        self.output = ""
1130
1131    def __read__(self, output):
1132        self.output = "%s%s" % (self.output, output)
1133
1134    def __error__(self, message):
1135        pass
1136
1137    def __done__(self, result_code="", message=""):
1138        pass
1139
1140
1141class DisplayOutputReceiver(IShellReceiver):
1142    def __init__(self):
1143        self.output = ""
1144        self.unfinished_line = ""
1145
1146    def _process_output(self, output, end_mark="\n"):
1147        content = output
1148        if self.unfinished_line:
1149            content = "".join((self.unfinished_line, content))
1150            self.unfinished_line = ""
1151        lines = content.split(end_mark)
1152        if content.endswith(end_mark):
1153            # get rid of the tail element of this list contains empty str
1154            return lines[:-1]
1155        else:
1156            self.unfinished_line = lines[-1]
1157            # not return the tail element of this list contains unfinished str,
1158            # so we set position -1
1159            return lines[:-1]
1160
1161    def __read__(self, output):
1162        self.output = "%s%s" % (self.output, output)
1163        lines = self._process_output(output)
1164        for line in lines:
1165            line = line.strip()
1166            if line:
1167                LOG.info(line)
1168
1169    def __error__(self, message):
1170        pass
1171
1172    def __done__(self, result_code="", message=""):
1173        pass
1174
1175
1176def process_command_ret(ret, receiver):
1177    try:
1178        if ret != "" and receiver:
1179            receiver.__read__(ret)
1180            receiver.__done__()
1181    except Exception as error:
1182        LOG.exception("Error generating log report.", exc_info=False)
1183        raise ReportException() from error
1184
1185    if ret != "" and not receiver:
1186        lines = ret.split("\n")
1187        for line in lines:
1188            line = line.strip()
1189            if line:
1190                LOG.debug(line)
1191