1#!/usr/bin/env python3
2# -*- coding: utf-8 -*-
3
4#
5# Copyright (c) 2023 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 shutil
20import sys
21import os
22import time
23import threading
24import subprocess
25from enum import Enum
26
27from containers.status import throw_exception
28from exceptions.ohos_exception import OHOSException
29from services.interface.build_file_generator_interface import BuildFileGeneratorInterface
30from containers.arg import Arg, ModuleType
31from util.system_util import SystemUtil
32from util.io_util import IoUtil
33from util.log_util import LogUtil
34from util.component_util import ComponentUtil
35
36
37class CMDTYPE(Enum):
38    PUSH = 1
39    LIST_TARGETS = 2
40
41
42class Hdc(BuildFileGeneratorInterface):
43
44    def __init__(self):
45        super().__init__()
46        self._regist_hdc_path()
47        self._regist_hpm_cache()
48
49    def run(self):
50        self.execute_hdc_cmd(CMDTYPE.PUSH)
51
52    @throw_exception
53    def execute_hdc_cmd(self, cmd_type: int, **kwargs):
54        if cmd_type == CMDTYPE.PUSH:
55            return self._execute_hdc_push_cmd()
56        elif cmd_type == CMDTYPE.LIST_TARGETS:
57            return self._execute_hdc_list_targets_cmd()
58        else:
59            raise OHOSException(
60                'You are tring to use an unsupported hdc cmd type "{}"'.format(cmd_type), '3001')
61
62    @throw_exception
63    def _regist_hdc_path(self):
64        hdc_path = shutil.which("hdc")
65        if os.path.exists(hdc_path):
66            self.exec = hdc_path
67        else:
68            raise OHOSException(
69                'There is no hdc executable file at {}'.format(hdc_path), '0001')
70
71    @throw_exception
72    def _regist_hpm_cache(self):
73        self.hpm_config_path = "~/.hpm"
74        hpm_path = shutil.which("hpm")
75        if os.path.exists(hpm_path):
76            command = [hpm_path, "config", "get", "modelRepository"]
77            output = subprocess.check_output(command)
78            if output:
79                self.hpm_config_path = output.decode('utf-8').rstrip("\n")
80
81    # 通过部件名获取部件二进制路径
82    def get_send_file(self, part_name):
83        srcs = []
84        target = ""
85        bundle_file = os.path.join(self.hpm_config_path, '.hpmcache/binarys/subsystem', part_name, "bundle.json")
86        if not os.path.exists(bundle_file):
87            return srcs, target
88        bundle_info = IoUtil.read_json_file(bundle_file)
89        deployment = bundle_info.get("deployment")
90        if deployment:
91            src = deployment.get("src", "")
92            target = deployment.get("target", "")
93            if not src.startswith("/"):
94                src = os.path.join(os.path.dirname(bundle_file), src)
95            srcs = get_files_by_path(src)
96        return srcs, target
97
98    @throw_exception
99    def _execute_hdc_push_cmd(self, **kwargs):
100        connect_key = self.flags_dict.get("target")
101        # mount
102        hdc_mount_cmd = [self.exec, '-t', connect_key, 'shell', 'mount', '-o', 'rw,remount', '/']
103        SystemUtil.exec_command(hdc_mount_cmd)
104        # parse part_name
105        part_name = self.flags_dict.get("part_name")
106        send_files, target = self.get_send_file(part_name)
107        send_src = self.flags_dict.get("src")
108        if send_src:
109            send_files = get_files_by_path(send_src)
110        # send file
111        for send_file in send_files:
112            hdc_push_cmd = [self.exec, "-t", connect_key, "file", "send", send_file,
113                            os.path.join(target, os.path.basename(send_file))]
114            SystemUtil.exec_command(hdc_push_cmd)
115            hdc_push_chown_cmd = [self.exec, "-t", connect_key, "shell", "chown", "root:root",
116                                  os.path.join(target, os.path.basename(send_file))]
117            SystemUtil.exec_command(hdc_push_chown_cmd)
118
119        # reboot
120        if self.flags_dict.get("reboot"):
121            hdc_reboot_cmd = [self.exec, '-t', connect_key, 'shell', 'reboot']
122            SystemUtil.exec_command(hdc_reboot_cmd)
123
124    @throw_exception
125    def _execute_hdc_list_targets_cmd(self, **kwargs):
126        hdc_list_targets_cmd = [self.exec, "list", "targets"]
127        SystemUtil.exec_command(hdc_list_targets_cmd)
128
129
130def get_files_by_path(path):
131    output = []
132    if os.path.isdir(path):
133        for root, dirnames, filenames in os.walk(path):
134            for filename in filenames:
135                output.append(os.path.join(root, filename))
136    else:
137        output.append(path)
138    return output
139