• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1#!/usr/bin/env python
2# -*- coding: utf-8 -*-
3
4#
5# Copyright (c) 2020 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 collections
21import subprocess
22import shutil
23import sys
24import json
25
26
27def encode(data, encoding: str='utf-8'):
28    if sys.version_info.major == 2:
29        return data.encode(encoding)
30    return data
31
32
33def decode(data, encoding: str='utf-8'):
34    if sys.version_info.major == 2:
35        return data.decode(encoding)
36    return data
37
38
39def remove_path(path):
40    if os.path.exists(path):
41        shutil.rmtree(path)
42
43
44# Read json file data
45def read_json_file(input_file: str):
46    if not os.path.exists(input_file):
47        print('file [{}] no exist.'.format(input_file))
48        return None
49    data = None
50    with open(input_file, 'rb') as input_f:
51        data = json.load(input_f)
52    return data
53
54
55def exec_command(cmd: str, log_path: str='out/build.log', **kwargs):
56    with open(log_path, 'at', encoding='utf-8') as log_file:
57        process = subprocess.Popen(cmd,
58                                   stdout=subprocess.PIPE,
59                                   stderr=subprocess.PIPE,
60                                   encoding='utf-8',
61                                   **kwargs)
62        for line in iter(process.stdout.readline, ''):
63            sys.stdout.write(line)
64            log_file.write(line)
65
66    process.wait()
67    ret_code = process.returncode
68
69    if ret_code != 0:
70        with open(log_path, 'at', encoding='utf-8') as log_file:
71            for line in iter(process.stderr.readline, ''):
72                sys.stdout.write(line)
73                log_file.write(line)
74        print('you can check build log in {}'.format(log_path))
75        raise Exception("{} failed, return code is {}".format(cmd, ret_code))
76
77
78def check_output(cmd: str, **kwargs):
79    try:
80        ret = subprocess.check_output(cmd,
81                                      stderr=subprocess.STDOUT,
82                                      universal_newlines=True,
83                                      **kwargs)
84    except subprocess.CalledProcessError as e:
85        ret = e.output
86        raise Exception("{} failed, failed log is {}".format(cmd, ret))
87
88    return ret
89
90
91def makedirs(path: str, exist_ok: bool=True):
92    try:
93        os.makedirs(path, exist_ok=True)
94    except OSError:
95        if not os.path.isdir(path):
96            raise Exception("{} makedirs failed".format(path))
97        if not exist_ok:
98            raise Exception("{} exists, makedirs failed".format(path))
99
100
101class CallbackDict(object):
102    handlers = None
103
104    def __init__(self):
105        self.handlers = collections.defaultdict(list)
106
107    def register(self, event, callback):
108        self.handlers[event].append(callback)
109
110    def excute(self, event, **kwargs):
111        if event not in self.handlers:
112            raise Exception('{} not found in callback dict'.format(event))
113        for handler in self.handlers.get(event, []):
114            handler(**kwargs)
115