import logging
import apt
import stat
+import argparse
if sys.version_info[0] < 3:
f.write(line)
-def generate_delta_image(BASE_OLD, BASE_NEW, OUT_DIR, DELTA_BIN, UPDATE_CFG_PATH, COMPRESSION_METHOD):
- oldsize_d = os.path.getsize(BASE_OLD)
- newsize_d = os.path.getsize(BASE_NEW)
- SHA_BIN_DEST = hash_file(BASE_NEW)
- SHA_BIN_BASE = hash_file(BASE_OLD)
+def generate_delta_image(args, COMPRESSION_METHOD):
+ #for sizes
+ oldsize_d = os.path.getsize(args.BASE_OLD)
+ newsize_d = os.path.getsize(args.BASE_NEW)
+ SHA_BIN_DEST = hash_file(args.BASE_NEW)
+ SHA_BIN_BASE = hash_file(args.BASE_OLD)
- DELTA = DELTA_BIN
- update_size(BASE_OLD, BASE_NEW)
- if UPDATE_CFG_PATH:
- with open(UPDATE_CFG_PATH, 'r') as f:
+ DELTA = args.PART_NAME
+ update_size(args.BASE_OLD, args.BASE_NEW)
+ if args.UPDATE_CFG_PATH:
+ with open(args.UPDATE_CFG_PATH, 'r') as f:
lines = f.readlines()
- with open(UPDATE_CFG_PATH, 'w') as f:
+ with open(args.UPDATE_CFG_PATH, 'w') as f:
for line in lines:
ConfigItems = line.split()
- if ConfigItems[0] == DELTA_BIN:
- logging.info('Updating %s config' % DELTA_BIN)
+ if ConfigItems[0] == args.PART_NAME:
+ logging.info('Updating %s config' % args.PART_NAME)
DELTA = ConfigItems[1]
line = line.rstrip('\n')
line = line.replace(line, line + '\t' + str(oldsize_d) + '\t\t' + str(newsize_d) + '\t\t' + str(SHA_BIN_BASE) + '\t\t' + str(SHA_BIN_DEST) + '\n')
f.write(line)
- patchLoc = '%s/%s' % (OUT_DIR, DELTA)
- logging.info('Make Delta Image %s <--> %s ==> %s %s' % (BASE_OLD, BASE_NEW, DELTA_BIN, patchLoc))
- subprocess.call([DIFF_UTIL, "-c", COMPRESSION_METHOD, BASE_OLD, BASE_NEW, patchLoc])
+ patchLoc = '%s/%s' % (args.OUT_DIR, DELTA)
+ logging.info('Make DELTA_IMAGE type delta %s <--> %s ==> %s %s' % (args.BASE_OLD, args.BASE_NEW, args.PART_NAME, patchLoc))
+ subprocess.call([DIFF_UTIL, "-c", COMPRESSION_METHOD, args.BASE_OLD, args.BASE_NEW, patchLoc])
-def generate_full_image(BASE_OLD, BASE_NEW, OUT_DIR, DELTA_BIN, UPDATE_CFG_PATH):
- logging.info('Make Full Image %s <--> %s ==> %s' % (BASE_OLD, BASE_NEW, DELTA_BIN))
- oldsize_d = os.path.getsize(BASE_OLD)
- newsize_d = os.path.getsize(BASE_NEW)
- SHA_BIN_DEST = hash_file(BASE_NEW)
- SHA_BIN_BASE = hash_file(BASE_OLD)
- update_size(BASE_OLD, BASE_NEW)
+def generate_full_image(args):
+ logging.info('Make FULL_IMAGE type delta %s <--> %s ==> %s' % (args.BASE_OLD, args.BASE_NEW, args.PART_NAME))
+ oldsize_d = os.path.getsize(args.BASE_OLD)
+ newsize_d = os.path.getsize(args.BASE_NEW)
+ SHA_BIN_DEST = hash_file(args.BASE_NEW)
+ SHA_BIN_BASE = hash_file(args.BASE_OLD)
+ update_size(args.BASE_OLD, args.BASE_NEW)
- if UPDATE_CFG_PATH:
- with open(UPDATE_CFG_PATH, 'r') as f:
+ if args.UPDATE_CFG_PATH:
+ with open(args.UPDATE_CFG_PATH, 'r') as f:
lines = f.readlines()
- with open(UPDATE_CFG_PATH, 'w') as f:
+ with open(args.UPDATE_CFG_PATH, 'w') as f:
for line in lines:
ConfigItems = line.split()
- if ConfigItems[0] == DELTA_BIN:
- logging.info('Updating %s config' % DELTA_BIN)
+ if ConfigItems[0] == args.PART_NAME:
+ logging.info('Updating %s config' % args.PART_NAME)
DELTA = ConfigItems[1]
line = line.rstrip('\n')
line = line.replace(line, line + '\t' + str(oldsize_d) + '\t\t' + str(newsize_d) + '\t\t' + str(SHA_BIN_BASE) + '\t\t' + str(SHA_BIN_DEST) + '\n')
f.write(line)
-def generate_delta_fs(PART_NAME, BASE_OLD, BASE_NEW, OUT_DIR, ATTR_FILE, ATTR_NEW):
- delta_fs_generator = DeltaFsGenerator(PART_NAME, BASE_OLD, BASE_NEW, OUT_DIR, ATTR_FILE, ATTR_NEW)
+def generate_delta_fs(args):
+ ATTR_FILE = os.path.join(args.OUT_DIR, (args.PART_NAME + ATTR_DOC_EXT))
+ diff_attr_files(args.ATTR_OLD, args.ATTR_NEW, ATTR_FILE)
+ delta_fs_generator = DeltaFsGenerator(args.PART_NAME, args.BASE_OLD, args.BASE_NEW, args.OUT_DIR, ATTR_FILE, args.ATTR_NEW)
delta_fs_generator.generate_delta_fs()
+def create_parser():
+ parser = argparse.ArgumentParser(description="Generate a delta of one of three types: DELTA_IMG, FULL_IMG, DELTA_FS")
+
+ parser.add_argument('UPDATE_TYPE_TIME', type=str, help='TYPE_OF_UPDATE:TIME_TO_UPGRADE (no time required for DELTA_FS)')
+ parser.add_argument('PART_NAME', metavar='PARTITION_NAME', type=str, help='name of partition for which we generate the delta')
+ parser.add_argument('BASE_OLD', type=str, help='older image path / mount point of the older version of the partition')
+ parser.add_argument('BASE_NEW', type=str, help='newer image path / mount point of the newer version of the partition')
+ parser.add_argument('OUT_DIR', type=str, help='directory where generated delta nad logs will be placed')
+ parser.add_argument('UPDATE_CFG_PATH', type=str, help='path to config file which will be updated')
+ parser.add_argument('--attr_old', type=str, dest='ATTR_OLD', help='path to old attribute file')
+ parser.add_argument('--attr_new', type=str, dest='ATTR_NEW', help='path to new attribute file')
+
+ return parser
+
+
def main():
+ global DIFF_UTIL
+ global DIFFPATCH_UTIL
+
logging.basicConfig(filename=LOGFILE, level=logging.DEBUG)
try:
- if len(sys.argv) < 5:
- sys.exit('Usage: CreatePatch.py UPDATE_TYPE PARTITION_NAME OLD_BASE_DIR NEW_BASE_DIR OUTFOLDER')
- UPDATE_TYPE = sys.argv[1]
- UPDATE_TYPE_S = UPDATE_TYPE.split(":")[0]
- # TODO make PART_NAME optional
- PART_NAME = sys.argv[2]
-
- BASE_OLD = sys.argv[3]
- BASE_NEW = sys.argv[4]
- OUT_DIR = sys.argv[5]
- ATTR_OLD = EMPTY
- ATTR_NEW = EMPTY
- UPDATE_CFG_PATH = EMPTY
-
- global DIFF_UTIL
- global DIFFPATCH_UTIL
+ parser = create_parser()
+ args = parser.parse_args()
+ UPDATE_TYPE_S = args.UPDATE_TYPE_TIME.split(":")[0]
+
if UPDATE_TYPE_S == DELTA_FS:
- if len(sys.argv) == 9:
- ATTR_OLD = sys.argv[6]
- ATTR_NEW = sys.argv[7]
- UPDATE_CFG_PATH = os.path.join(PARENT_DIR, sys.argv[8])
+ if not args.ATTR_OLD or not args.ATTR_NEW:
+ parser.error('In case of DELTA_FS type delta ATTR_NEW and ATTR_OLD are required!')
- elif UPDATE_TYPE_S in [DELTA_IMAGE, FULL_IMAGE]:
- if len(sys.argv) == 7:
- UPDATE_CFG_PATH = os.path.join(PARENT_DIR, sys.argv[6])
+ elif UPDATE_TYPE_S not in [DELTA_IMAGE, FULL_IMAGE]:
+ parser.error('Unknown update type!')
+
+ if not (os.path.exists(args.BASE_OLD) and os.path.exists(args.BASE_NEW)):
+ parser.error('Paths to images/mountpoints do not exist!')
+
+ args.UPDATE_CFG_PATH = os.path.join(PARENT_DIR, args.UPDATE_CFG_PATH)
if not (os.path.isfile(DIFF_UTIL) and os.access(DIFF_UTIL, os.X_OK)):
DIFF_UTIL = os.path.join(COMMON_BIN_PATH, DIFF_UTIL)
text = f'Started CreatePatch.py at {start}'
logging.info(f'{text:^70}')
print(f'{text:^70}')
- logging.info('Arguments Passed: [UpdateType - %s][Part Name - %s] [BaseOld - %s] [BaseNew - %s] \n [OUTPUTDir - %s] \
- [BASE ATTR - %s] [TARGET ATTR - %s]' % (UPDATE_TYPE, PART_NAME, BASE_OLD, BASE_NEW, OUT_DIR, ATTR_OLD, ATTR_NEW))
+ logging.info('Arguments passed: [UpdateType - %s][Part Name - %s] [BaseOld - %s] [BaseNew - %s] \n [OUTPUTDir - %s]' \
+ % (args.UPDATE_TYPE_TIME, args.PART_NAME, args.BASE_OLD, args.BASE_NEW, args.OUT_DIR))
+ if UPDATE_TYPE_S == DELTA_FS:
+ logging.info('Attribute files for DELTA_FS: [BASE ATTR - %s] [TARGET ATTR - %s]' % (args.ATTR_OLD, args.ATTR_NEW))
try:
- ensure_dir_exists(OUT_DIR)
+ ensure_dir_exists(args.OUT_DIR)
except FileExistsError as exc:
- logging.error('Argument passed as OUT_DIR - %s is already an existing file' % OUT_DIR)
+ logging.error('Argument passed as OUT_DIR - %s is already an existing file' % args.OUT_DIR)
raise exc
+
if UPDATE_TYPE_S == DELTA_FS:
- if not (os.path.isfile(ATTR_OLD) and os.path.isfile(ATTR_NEW)):
- print("Attributes missing -- ABORT", file=sys.stderr)
+ if not (os.path.isfile(args.ATTR_OLD) and os.path.isfile(args.ATTR_NEW)):
+ print("Attribute files do not exist -- ABORT", file=sys.stderr)
sys.exit(1)
# TODO verify if other linux distributions support APT library
sys.exit(1)
if UPDATE_TYPE_S == FULL_IMAGE:
- generate_full_image(BASE_OLD, BASE_NEW, OUT_DIR, PART_NAME, UPDATE_CFG_PATH)
+ generate_full_image(args)
elif UPDATE_TYPE_S == DELTA_IMAGE:
# generating LZMA deltas is supported by underlying software but it would require at least some kind of format autodetection mechanism,
# which is not available currently. Disable for now.
- generate_delta_image(BASE_OLD, BASE_NEW, OUT_DIR, PART_NAME, UPDATE_CFG_PATH, COMPRESSION_BROTLI)
- elif UPDATE_TYPE_S == DELTA_FS:
- ATTR_FILE = os.path.join(OUT_DIR, (PART_NAME + ATTR_DOC_EXT))
- diff_attr_files(ATTR_OLD, ATTR_NEW, ATTR_FILE)
- generate_delta_fs(PART_NAME, BASE_OLD, BASE_NEW, OUT_DIR, ATTR_FILE, ATTR_NEW)
-
- if UPDATE_CFG_PATH:
- update_cfg_file(PART_NAME, UPDATE_CFG_PATH)
+ generate_delta_image(args, COMPRESSION_BROTLI)
else:
- print('UPDATE_TYPE ---- UNKNOWN FORMAT')
- raise TypeError
+ generate_delta_fs(args)
- if UPDATE_TYPE_S == DELTA_FS:
- if os.path.exists(ATTR_OLD) and os.path.exists(ATTR_NEW):
- os.remove(ATTR_OLD)
- os.remove(ATTR_NEW)
+ if args.UPDATE_CFG_PATH:
+ update_cfg_file(args.PART_NAME, args.UPDATE_CFG_PATH)
+
+ if os.path.exists(args.ATTR_OLD) and os.path.exists(args.ATTR_NEW):
+ os.remove(args.ATTR_OLD)
+ os.remove(args.ATTR_NEW)
end = datetime.datetime.now().time()
- logging.info('Max memory requried to upgrade [%s] is [%d] for file [%s]' % (PART_NAME, MEM_REQ, MEM_FILE))
+ logging.info('Max memory requried to upgrade [%s] is [%d] for File[%s]' % (args.PART_NAME, MEM_REQ, MEM_FILE))
text = f'Done with CreatePath.py at {end}'
logging.info(f'{text:^70}')
print(f'{text:^70}')
logging.info('Time start [%s] - Time end [%s]' % (start, end))
- print('Done with [%s][%d]---- Time start [%s] - Time end [%s]' % (PART_NAME, MEM_REQ, start, end))
+ print('Done with [%s][%d]---- Time start [%s] - Time end [%s]' % (args.PART_NAME, MEM_REQ, start, end))
except Exception as exc:
logging.error('Usage: {} <Update_Type> <Part_Name> <OLD_Base> <NEW_Base> <OUT_DIR>'.format(os.path.basename(sys.argv[0])))
projects / platform / core / system / upgrade-tools.git / commitdiff