• 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
19
20import os
21import sys
22import shutil
23import argparse
24sys.path.append(os.path.dirname(os.path.dirname(
25    os.path.dirname(os.path.abspath(__file__)))))
26from lite.hb_internal.common import utils
27from lite.hb_internal.common.config import Config
28
29class Packer():
30    def __init__(self, packer_args) -> None:
31        self.config = Config()
32        self.replace_items = {
33            r'${product_name}': self.config.product,
34            r'${root_path}': self.config.root_path,
35            r'${out_path}': self.config.out_path
36        }
37        self.packing_process = [
38            self.mv_usr_libs, self.create_fs_dirs, self.fs_link,
39            self.fs_filemode, self.fs_make_cmd, self.fs_tear_down
40        ]
41        self.fs_cfg = None
42        self.chmod_dirs = []
43
44    def mv_usr_libs(self):
45        src_path = self.config.out_path
46        libs = [lib for lib in os.listdir(src_path) if self.is_lib(lib)]
47        target_path = os.path.join(src_path, 'usr', 'lib')
48        utils.makedirs(target_path, exist_ok=True)
49
50        for lib in libs:
51            source_file = os.path.join(src_path, lib)
52            target_file = os.path.join(target_path, lib)
53            shutil.move(source_file, target_file)
54
55    @classmethod
56    def is_lib(cls, lib):
57        return lib.startswith('lib') and lib.endswith('.so')
58
59    @classmethod
60    def is_incr(cls, fs_incr):
61        exist_ok = False if fs_incr is None else True
62        with_rm = True if fs_incr is None else False
63        return exist_ok, with_rm
64
65    def create_fs_dirs(self):
66        fs_path = os.path.join(self.config.out_path,
67                               self.fs_cfg.get('fs_dir_name', 'rootfs'))
68        exist_ok, with_rm = self.is_incr(self.fs_cfg.get('fs_incr', None))
69
70        utils.makedirs(fs_path, exist_ok=exist_ok, with_rm=with_rm)
71        self.replace_items[r'${fs_dir}'] = fs_path
72
73        for fs_dir in self.fs_cfg.get('fs_dirs', []):
74            source_dir = fs_dir.get('source_dir', '')
75            target_dir = fs_dir.get('target_dir', '')
76            if target_dir == '':
77                continue
78
79            source_path = self.fs_dirs_replace(source_dir,
80                                               self.config.out_path)
81            target_path = self.fs_dirs_replace(target_dir, fs_path)
82
83            if source_dir == '' or not os.path.exists(source_path):
84                utils.makedirs(target_path)
85                target_mode_tuple = (target_path, fs_dir.get('dir_mode', 755))
86                self.chmod_dirs.append(target_mode_tuple)
87                continue
88
89            self.copy_files(source_path, target_path, fs_dir)
90
91    def fs_dirs_replace(self, path, default_path):
92        source_path, is_changed = self.replace(path)
93        if not is_changed:
94            source_path = os.path.join(default_path, path)
95        return source_path
96
97    def copy_files(self, spath, tpath, fs_dir):
98        ignore_files = fs_dir.get('ignore_files', [])
99        dir_mode = fs_dir.get('dir_mode', 755)
100        file_mode = fs_dir.get('file_mode', 555)
101
102        def copy_file_process(source_path, target_path):
103            if not os.path.isdir(target_path):
104                utils.makedirs(target_path)
105                self.chmod_dirs.append((target_path, dir_mode))
106            tfile = os.path.join(target_path, os.path.basename(source_path))
107            try:
108                shutil.copy(sfile, tfile, follow_symlinks=False)
109                self.chmod_dirs.append((tfile, file_mode))
110            except FileExistsError:
111                utils.hb_warning(f'Target file: {tfile} already exists!')
112
113        if os.path.isfile(spath):
114            sfile = spath
115            copy_file_process(spath, tpath)
116            return
117
118        for srelpath, sfile in self.list_all_files(spath, ignore_files):
119            tdirname = srelpath.replace(spath, tpath)
120            copy_file_process(sfile, tdirname)
121
122    @classmethod
123    def chmod(cls, file, mode):
124        mode = int(str(mode), base=8)
125        if os.path.exists(file):
126            os.chmod(file, mode)
127
128    @classmethod
129    def filter(cls, files, ignore_list):
130        if ignore_list is None or not len(ignore_list):
131            return files
132        filter_files = []
133        for file in files:
134            flag = True
135            for ignore in ignore_list:
136                if file.startswith(ignore) or file.endswith(ignore):
137                    flag = False
138                    break
139            if flag:
140                filter_files.append(file)
141        return filter_files
142
143    def list_all_files(self, path, ignore_list=None):
144        for relpath, _, files in os.walk(path):
145            files = self.filter(files, ignore_list)
146            for file in files:
147                full_path = os.path.join(path, relpath, file)
148                if os.path.isfile(full_path):
149                    yield relpath, full_path
150
151    def replace(self, raw_str):
152        old_str = raw_str
153        for old, new in self.replace_items.items():
154            raw_str = raw_str.replace(old, new)
155        return raw_str, old_str != raw_str
156
157    def fs_link(self):
158        fs_symlink = self.fs_cfg.get('fs_symlink', [])
159        for symlink in fs_symlink:
160            source, _ = self.replace(symlink.get('source', ''))
161            link_name, _ = self.replace(symlink.get('link_name', ''))
162            if os.path.exists(link_name):
163                os.remove(link_name)
164            os.symlink(source, link_name)
165
166    def fs_filemode(self):
167        fs_filemode = self.fs_cfg.get('fs_filemode', [])
168        for filestat in fs_filemode:
169            file_dir = os.path.join(self.replace_items[r'${fs_dir}'],
170                                    filestat.get('file_dir', ''))
171            file_mode = filestat.get('file_mode', 0)
172            if os.path.exists(file_dir) and file_mode > 0:
173                self.chmod_dirs.append((file_dir, file_mode))
174
175        for file_dir, file_mode in self.chmod_dirs:
176            self.chmod(file_dir, file_mode)
177
178    def fs_make_cmd(self):
179        fs_make_cmd = self.fs_cfg.get('fs_make_cmd', [])
180        log_path = self.config.log_path
181
182        for cmd in fs_make_cmd:
183            cmd, _ = self.replace(cmd)
184            cmd = cmd.split(' ')
185            utils.exec_command(cmd, log_path=log_path)
186
187    def fs_tear_down(self):
188        while len(self.chmod_dirs):
189            tfile = self.chmod_dirs.pop()[0]
190
191            if os.path.isfile(tfile):
192                self.chmod(tfile, 555)
193            elif os.path.isdir(tfile):
194                self.chmod(tfile, 755)
195
196    def fs_attr_process(self, fs_cfg):
197        fs_attr = fs_cfg.get('fs_attr', {})
198        for attr_key, attr_value in fs_attr.items():
199            if attr_key in self.config.fs_attr:
200                for target_key, target_value in attr_value.items():
201                    if target_key in fs_cfg:
202                        fs_cfg[target_key] += target_value
203                    else:
204                        fs_cfg[target_key] = target_value
205
206        return fs_cfg
207
208    def fs_make(self, cmd_args):
209        fs_cfg_path = os.path.join(self.config.product_path, 'fs.yml')
210        if not os.path.isfile(fs_cfg_path):
211            utils.hb_info(f'{fs_cfg_path} not found, stop packing fs. '
212                          'If the product does not need to be packaged, ignore it.')
213            return
214        if self.config.fs_attr is None:
215            utils.hb_info('component compiling, no need to pack fs')
216            return
217
218        fs_cfg_list = utils.read_yaml_file(fs_cfg_path)
219        for fs_cfg in fs_cfg_list:
220            self.fs_cfg = self.fs_attr_process(fs_cfg)
221            if self.fs_cfg.get('fs_dir_name', None) is None:
222                continue
223
224            for fs_process_func in self.packing_process:
225                fs_process_func()
226
227
228if __name__ == "__main__":
229    parser = argparse.ArgumentParser()
230    parser.add_argument('--product', required=True)
231    parser.add_argument('--root-path', required=True)
232    parser.add_argument('--out-path', required=True)
233    parser.add_argument('--log-path', required=True)
234    parser.add_argument('--product-path', required=True)
235    args = parser.parse_args()
236
237    packer = Packer(args)
238
239    packer.replace_items = {
240        r'${product_name}': args.product,
241        r'${root_path}': args.root_path,
242        r'${out_path}': args.out_path,
243        r'${log_path}': args.log_path,
244        r'${product_path}': args.product_path
245    }
246    packer.fs_make('')