diff --git a/app_backup.py b/app_backup.py index bb1e636..df304db 100644 --- a/app_backup.py +++ b/app_backup.py @@ -1,4 +1,5 @@ -"""App > Get with Files: a full backup of the connected device's app as one zip. +"""App > Get with Files / Put with Files: a full backup of the device's app as one +zip, and putting such a zip back onto a device. The zip follows the layout Ontrol's tools read back on a restore: @@ -6,7 +7,7 @@ The zip follows the layout Ontrol's tools read back on a restore: device/kits.scode device/images/.bin device/fonts/ - device/system/logo.bin + device/system/logo.jpg Orion's boot logo; Rion's is system/logo.bin app.sax app.sab decoded, as sedonac would write it imageBank[XL]/.png the sources of the device's images and fonts, taken from fontBank[XL]/ Sedona home's banks (XL for Orion's 800x480 screen) @@ -14,6 +15,9 @@ The zip follows the layout Ontrol's tools read back on a restore: The device cannot list its files, so which images and fonts to fetch is read from the app itself: every image name and font its Ox widgets and pages name. + +A put stages app.sab and kits.scode for the VM to take up on its next start; see +put_with_files for the order it goes in. """ import datetime @@ -23,11 +27,11 @@ import zipfile import app_state import sab_app -from ox_widget_properties import IMAGE_EXTENSIONS, ORION_KIT, RION_KIT +import sox +from ox_widget_properties import IMAGE_EXTENSIONS, ORION_KIT, ORION_PLATFORM_TYPE, RION_KIT APP_SAB = "app.sab" KITS_SCODE = "kits.scode" -LOGO_URI = "system/logo.bin" _IMAGE_SLOTS = ("imageFileName", "trueImageFileName", "falseImageFileName") _SOURCE_IMAGE_EXTENSIONS = tuple(e for e in IMAGE_EXTENSIONS if e != ".bin") @@ -35,25 +39,33 @@ _UNSAFE_FILE_CHARS = re.compile(r'[<>:"/\\|?*\x00-\x1f]') class _Platform: - def __init__(self, name, image_bank, font_bank, font_extensions): + def __init__(self, name, kit, service_type, image_bank=None, font_bank=None, + font_extensions=(), logo_uri=None, screen=None): self.name = name + self.kit = kit # the kit only this platform's apps hold + self.service_type = service_type # its platform service + self.platform_id = "ontrol-" + name.lower() # the device's id starts so self.image_bank = image_bank self.font_bank = font_bank self.font_extensions = font_extensions + self.logo_uri = logo_uri + self.screen = screen # (width, height) in landscape; None: none -# Orion keeps a landscape (.oxfl) and a portrait (.oxfp) cut of each font. -_ORION = _Platform("ORION", "imageBankXL", "fontBankXL", (".oxfl", ".oxfp")) -_RION = _Platform("RION", "imageBank", "fontBank", (".bin",)) +# Orion keeps a landscape (.oxfl) and a portrait (.oxfp) cut of each font, and its boot +# logo as a jpg; Rion's is a native image (Ontrol's BGetWithFilesTool). A P-ION has no +# screen: nothing but the app goes to it. +_ORION = _Platform("ORION", ORION_KIT, ORION_PLATFORM_TYPE, "imageBankXL", "fontBankXL", + (".oxfl", ".oxfp"), "system/logo.jpg", (800, 480)) +_RION = _Platform("RION", RION_KIT, "ontrolRion::RionPlatformService", "imageBank", + "fontBank", (".bin",), "system/logo.bin", (320, 240)) +_PION = _Platform("PION", "ontrolPion", "ontrolPion::PionPlatformService") +_PLATFORMS = (_ORION, _RION, _PION) def platform_of(app): kits = app.kit_names() - if ORION_KIT in kits: - return _ORION - if RION_KIT in kits: - return _RION - return None + return next((p for p in _PLATFORMS if p.kit in kits), None) def backup_folder(): @@ -98,6 +110,8 @@ def _font_stem(app, comp): def media_files(app, platform): """The images and fonts the app names, each with its device path and bank source.""" files, seen = [], set() + if platform.screen is None: + return files def add(uri, bank_folder, bank_name, group, required=True): if uri not in seen: @@ -134,7 +148,7 @@ def media_files(app, platform): add_font(f"bold {stem}", group, True) add_font(stem, group, not bold) - add(LOGO_URI, platform.image_bank, "logo", "logo", required=False) + add(platform.logo_uri, platform.image_bank, "logo", "logo", required=False) return files @@ -154,7 +168,7 @@ def _bank_source(bank_folder, bank_name, is_image): def _source_entry(media, source_path): """Where the bank file goes in the zip: next to its bank's name, the logo in system/.""" file_name = os.path.basename(source_path) - if media.device_uri == LOGO_URI: + if media.device_uri.startswith("system/"): return "system/logo" + os.path.splitext(file_name)[1] return f"{media.bank_folder}/{file_name}" @@ -259,3 +273,454 @@ def get_with_file(client, path, status=None): result.entries = [name for name, _ in entries] report("Done", 1.0) return result + + +# --- Put with Files --------------------------------------------------------------------- +# +# Follows Ontrol's Workbench tool (ontrolSedonaOx BPutWithFilesTool, ontrolSedonanet +# PutApp): the backup's app.sax is encoded against the device's kits; app.sab goes as +# '.writing' renamed to '.stage'; the platform service's deleteFile clears fonts/* and +# images/*; then the images, the fonts and the boot logo go. Fonts come from Sedona +# home's bank in the cut the device draws with, images in the form its screen takes. + +_FONT_FOLDERS = ("device/fonts/", "fonts/", "fontBank/", "fontBankXL/") +_JPG = (".jpg", ".jpeg") + + +def _checksum(value): + """A kit checksum as an unsigned int, from a signed Sox int or a .sab hex string.""" + return int(value, 16) if isinstance(value, str) else value & 0xFFFFFFFF + + +def kits_fit(app_kits, device_kits): + """Whether the device's kits.scode can load an app.sab of `app_kits`. + + App.loadSchema's rule: the app's kits must appear among the device's, in the same + order, each with the same checksum; the device may hold more. + """ + device = iter((n, _checksum(c)) for n, c in device_kits) + return all(any(d == (n, _checksum(c)) for d in device) for n, c in app_kits) + + +def kit_differences(app_kits, device_kits): + """Lines telling why an app's kits do not fit the device's.""" + on_device = {n: _checksum(c) for n, c in device_kits} + lines = [] + for name, cks in app_kits: + cks = _checksum(cks) + if name not in on_device: + lines.append(f"{name}: not on the device") + elif on_device[name] != cks: + lines.append(f"{name}: {on_device[name]:08x} on the device, {cks:08x} in the backup") + return lines or ["The same kits, in a different order"] + + +class DeviceTarget: + """What a put needs to know of the connected device.""" + + def __init__(self, platform, kits, name=None, portrait=False, + platform_comp_id=None, delete_slot_id=None): + self.platform = platform # one of _PLATFORMS, or None + self.kits = kits # [(name, checksum)] in kit-id order + self.name = name # the app's deviceName + self.portrait = portrait # Orion's screenOrientation + self.platform_comp_id = platform_comp_id + self.delete_slot_id = delete_slot_id + + @property + def font_extension(self): + if self.platform is _ORION: + return ".oxfp" if self.portrait else ".oxfl" + return ".bin" + + @property + def screen(self): + """(width, height) as the device is set up, as Ontrol's ScreenInfo has it.""" + w, h = self.platform.screen + return (h, w) if self.portrait else (w, h) + + +def device_target(ds): + """Reads the platform, kits, deviceName and platform service off the device. + + The platform comes from the platform id, as Ontrol's tools read it, the kits + standing in for a device that does not give one. + """ + client = ds.sox_client + kits = [(p.name, p.checksum) for p in client.read_schema()] + kit_names = {n for n, _ in kits} + platform_id = (client.read_version().platform_id or "").lower() + platform = (next((p for p in _PLATFORMS if platform_id.startswith(p.platform_id)), None) + or next((p for p in _PLATFORMS if p.kit in kit_names), None)) + + target = DeviceTarget(platform, kits) + comp = client.cache_get(0) + if comp is not None: + client.update([comp], sox.CONFIG) + target.name = comp.values.get("deviceName") or None + if platform is not None and platform.screen is not None: + plat = ds.find_component_of_type(platform.service_type, "screenOrientation") + if plat is not None: + target.platform_comp_id = plat.id + target.portrait = (platform is _ORION and + str(plat.values.get("screenOrientation")).lower() == "true") + slots = client.slot_registry.get((plat.kit_id, plat.type_id)) or [] + target.delete_slot_id = next((s["id"] for s in slots if s["name"] == "deleteFile" + and s.get("is_action")), None) + return target + + +# --- images in the device's form -------------------------------------------------------- + +def native_image(data, name, target): + """A png (or any image PIL reads) as the device's native .bin. + + Ported from OntOxImageUtil.convertImageToNativeFormat: a 32-byte name, a 32-byte + header (flags, width, height), then the pixels. Orion draws BGR(A), turned for a + portrait screen, with alpha unless the image fills the screen; Rion RGB565, with + alpha only for a translucent image smaller than the screen. + """ + import io + from PIL import Image + + img = Image.open(io.BytesIO(data)) + img.load() + width, height = img.size + translucent = img.mode in ("RGBA", "LA", "PA") or ( + img.mode == "P" and "transparency" in img.info and + any(0 < a < 255 for a in img.convert("RGBA").getchannel("A").getdata())) + background = (width, height) == target.screen + orion = target.platform is _ORION + rgb_mode = use_alpha = orion and not background + if not orion: + use_alpha = translucent and not background + rotate = orion and target.portrait + + rgba = img.convert("RGBA") + if rotate: # the Java walks columns right to left: a quarter turn anticlockwise + rgba = rgba.transpose(Image.Transpose.ROTATE_90) + w, h = rgba.size + + out = bytearray(name.encode("ascii", errors="replace")) + out += bytes(max(0, 32 - len(out))) + out += bytes([(4 if rotate else 0) + (2 if rgb_mode else 0) + (1 if use_alpha else 0), + w & 0xFF, w >> 8, h & 0xFF, h >> 8]) + bytes(27) + pixels = rgba.tobytes() + if rgb_mode: + for i in range(0, len(pixels), 4): + r, g, b, a = pixels[i:i + 4] + out += bytes((b, g, r, a)) if use_alpha else bytes((b, g, r)) + else: + for i in range(0, len(pixels), 4): + r, g, b, a = pixels[i:i + 4] + rgb565 = (r >> 3) << 11 | (g >> 2) << 5 | (b >> 3) + if use_alpha: + out.append(a) + out += bytes((rgb565 & 0xFF, rgb565 >> 8)) + return bytes(out) + + +def device_jpg(data, target): + """A jpg as the device takes it: turned a quarter anticlockwise for a portrait Orion + (OntOxImageUtil.rotateJpg), otherwise as it is.""" + if not (target.platform is _ORION and target.portrait): + return data + import io + from PIL import Image + img = Image.open(io.BytesIO(data)) + buf = io.BytesIO() + img.transpose(Image.Transpose.ROTATE_90).save(buf, "JPEG", quality=75) + return buf.getvalue() + + +# --- the plan --------------------------------------------------------------------------- + +class PutFile: + def __init__(self, device_uri, data, source): + self.device_uri = device_uri + self.data = data + self.source = source # where the bytes come from, for the summary + + +class PutPlan: + """What a put of the zip would do, and whether the zip fits the device.""" + + def __init__(self, path): + self.path = path + self.problems = [] # any of these stops the put + self.warnings = [] # worth confirming, but the put can go on + self.app_source = "" # how app.sab was come by, for the summary + self.reencoded = [] # kits app.sax was encoded against other builds of + self.kit_changes = [] # set when the backup's kits.scode has to go too + self.app_files = [] # PutFile to '.writing', renamed to '.stage' + self.media = [] # PutFile: images, fonts, logo + self.converted = 0 # images made from the backup's image bank + self.images_missing = [] # images neither in device form nor as a source + self.fonts_from_zip = [] # fonts the bank lacks; the backup's copy goes + self.fonts_missing = [] # fonts the app names that neither has + self.staged = False # set once the device holds app.sab.stage + self.has_screen = True # False for a P-ION: the app is all that goes + + @property + def ok(self): + return not self.problems + + +def _label(platform): + return platform.name if platform else "no platform this editor knows" + + +def _app_portrait(app): + comp = next((c for c in app.comps.values() if c["type"] == ORION_PLATFORM_TYPE), None) + return bool(comp and app.prop(comp, "screenOrientation", False)) + + +def _sax_kits(sax): + import xml.etree.ElementTree as ET + schema_elem = ET.fromstring(sax).find("schema") + return [] if schema_elem is None else [k.get("name") for k in schema_elem.findall("kit")] + + +def plan_put(path, target): + """Reads the zip at `path` and checks it against the device `target`.""" + plan = PutPlan(path) + try: + z = zipfile.ZipFile(path) + except (OSError, zipfile.BadZipFile) as e: + plan.problems.append(f"Not a readable zip: {e}") + return plan + with z: + names = {n for n in z.namelist() if not n.endswith("/")} + app, backup_portrait = _plan_app(plan, z, names, target) + if app is None: + return plan + + platform = platform_of(app) + if platform is None or platform is not target.platform: + plan.problems.append(f"The backup is from {_label(platform)}; the device is " + f"{_label(target.platform)}.") + return plan + + zip_name = app.prop(app.app, "deviceName", "") or "" + if target.name and zip_name and zip_name != target.name: + plan.warnings.append(f"The backup was taken from '{zip_name}'; this device is " + f"'{target.name}'.") + if platform.screen is None: + plan.has_screen = False # no images or fonts to put + return plan + if target.delete_slot_id is None: + plan.warnings.append("The device's platform service has no deleteFile action, " + "so its old fonts and images stay.") + + same_screen = platform is _RION or backup_portrait == target.portrait + if not same_screen: + plan.warnings.append( + f"The backup is from a {'portrait' if backup_portrait else 'landscape'} " + f"Orion; this one is {'portrait' if target.portrait else 'landscape'}. Its " + "images are made again from the backup's image bank.") + _plan_images(plan, z, names, platform, target, same_screen) + _plan_fonts(plan, z, names, app, platform, target.font_extension) + _plan_logo(plan, z, names, platform, target, same_screen) + return plan + + +def _plan_app(plan, z, names, target): + """Settles app.sab, Ontrol's way first: the backup's app.sax encoded against the + device's kits. Failing that, the backup's own app.sab when the device's kits can + load it, or with the backup's kits.scode. Returns (decoded app, its portrait).""" + sax_name = "app.sax" if "app.sax" in names else next( + (n for n in sorted(names) if n.endswith(".sax") and "/" not in n), None) + sab_name, scode_name = "device/" + APP_SAB, "device/" + KITS_SCODE + device_cks = {n: f"{_checksum(c):08x}" for n, c in target.kits} + sax_problem = None + + if sax_name: + sax = z.read(sax_name) + try: + missing = [k for k in _sax_kits(sax) if k not in device_cks] + if missing: + raise sab_app.SabError("the device lacks the kits " + ", ".join(missing)) + sab = sab_app.encode_sax(sax, device_cks) + app = sab_app.SabApp(sab) + except Exception as e: + sax_problem = f"{sax_name} cannot be encoded for this device: {e}" + else: + original = sab_app.SabApp(z.read(sab_name)) if sab_name in names else None + if original is not None: + plan.reencoded = [line for line in kit_differences(original.kit_parts, + target.kits) + if "on the device," in line] + plan.app_source = f"{sax_name}, encoded for the device's kits" + plan.app_files.append(PutFile(APP_SAB, sab, sax_name)) + return app, _app_portrait(original or app) + + if sab_name not in names: + plan.problems.append(sax_problem or "The zip holds no app: neither app.sax nor " + f"{sab_name}.") + return None, False + sab = z.read(sab_name) + try: + app = sab_app.SabApp(sab) + except Exception as e: + plan.problems += [p for p in (sax_problem,) if p] + plan.problems.append(f"{sab_name} cannot be read: {e}") + return None, False + if kits_fit(app.kit_parts, target.kits): + plan.app_source = sab_name + elif scode_name in names: + plan.kit_changes = kit_differences(app.kit_parts, target.kits) + plan.app_source = f"{sab_name}, with the backup's {KITS_SCODE}" + plan.app_files.append(PutFile(KITS_SCODE, z.read(scode_name), scode_name)) + else: + plan.problems += [p for p in (sax_problem,) if p] + plan.problems.append(f"The device's kits cannot load {sab_name}, and the backup " + f"has no {scode_name}:") + plan.problems += [" " + line for line in kit_differences(app.kit_parts, target.kits)] + return None, False + if sax_problem: + plan.warnings.append(sax_problem) + plan.app_files.append(PutFile(APP_SAB, sab, sab_name)) + return app, _app_portrait(app) + + +def _plan_images(plan, z, names, platform, target, same_screen): + """Every image the backup holds, in the form the device's screen takes. + + The backup's device copy goes when it was taken off the same kind of screen; + otherwise the image bank's source is made over, as Ontrol's tool always does. + """ + native, source = {}, {} + for name in names: + folder, _, file_name = name.rpartition("/") + stem, ext = os.path.splitext(file_name) + if folder == "device/images": + native[stem] = name + elif folder == platform.image_bank and ext.lower() in _SOURCE_IMAGE_EXTENSIONS: + source[stem] = name + for stem in sorted(set(native) | set(source), key=str.lower): + if same_screen and stem in native: + name = native[stem] + plan.media.append(PutFile("images/" + name.rpartition("/")[2], z.read(name), + "backup")) + elif stem in source: + name = source[stem] + try: + if name.lower().endswith(_JPG): + plan.media.append(PutFile(f"images/{stem}.jpg", + device_jpg(z.read(name), target), name)) + else: + plan.media.append(PutFile(f"images/{stem}.bin", + native_image(z.read(name), stem, target), name)) + plan.converted += 1 + except Exception as e: + plan.images_missing.append(f"{stem} ({name}: {e})") + else: + plan.images_missing.append(stem) + + +def _plan_logo(plan, z, names, platform, target, same_screen): + """The boot logo: the device's copy, or the backup's source made over (Ontrol's + tool sends it into images/ -- system/ is where the device looks for it).""" + copy = "device/" + platform.logo_uri + if same_screen and copy in names: + plan.media.append(PutFile(platform.logo_uri, z.read(copy), "backup")) + return + source = next((n for n in sorted(names) if n.rpartition("/")[0] == "system" and + os.path.splitext(n)[0].endswith("logo")), None) + if source is None: + return + try: + data = z.read(source) + if platform is _ORION: + if not source.lower().endswith(_JPG): + return + plan.media.append(PutFile(platform.logo_uri, device_jpg(data, target), source)) + else: + plan.media.append(PutFile(platform.logo_uri, native_image(data, "logo", target), + source)) + except Exception as e: + plan.warnings.append(f"The boot logo could not be made from {source}: {e}") + + +def _plan_fonts(plan, z, names, app, platform, ext): + """Every font the backup holds or the app names, in the device's cut `ext`. + + As Ontrol's tool does, the bank's file goes rather than the backup's; the backup's + goes only for a font the bank lacks. + """ + in_zip = {} # stem -> {extension: entry} + for name in names: + folder = next((f for f in _FONT_FOLDERS if name.startswith(f)), None) + if folder and "/" not in name[len(folder):]: + stem, file_ext = os.path.splitext(name[len(folder):]) + in_zip.setdefault(stem, {}).setdefault(file_ext.lower(), name) + required = {} + for m in media_files(app, platform): + if m.device_uri.startswith("fonts/"): + stem = os.path.splitext(m.device_uri[len("fonts/"):])[0] + required[stem] = required.get(stem, False) or m.required + + for stem in sorted(set(in_zip) | set(required)): + bank = _bank_source(platform.font_bank, stem + ext, False) + if bank: + with open(bank, "rb") as f: + plan.media.append(PutFile("fonts/" + stem + ext, f.read(), + f"{platform.font_bank}/{stem}{ext}")) + elif ext in in_zip.get(stem, {}): + plan.media.append(PutFile("fonts/" + stem + ext, + z.read(in_zip[stem][ext]), "backup")) + plan.fonts_from_zip.append(stem + ext) + elif required.get(stem) or stem in in_zip: + plan.fonts_missing.append(stem + ext) + + +def put_with_files(client, plan, target, status=None): + """Puts the plan on the device; the device takes the app up when it restarts. + + Runs off the main thread; status(text, fraction) reports progress. A failure + raises, naming the step. Once app.sab.stage is in place the app is committed: a + later failure leaves the new app staged with some of its media missing. + """ + def report(text, fraction=None): + if status: + status(text, fraction) + + total = sum(len(f.data) for f in plan.app_files + plan.media) or 1 + done = 0 + + def put(uri, f): + nonlocal done + base = done + + def progress(sent, size): + report(f"Writing {uri} ...", (base + sent) / total) + report(f"Writing {uri} ...", base / total) + try: + client.put_file(uri, f.data, progress) + except Exception as e: + raise RuntimeError(f"Writing {uri}: {e}") from e + done += len(f.data) + + for f in plan.app_files: + put(f.device_uri + ".writing", f) + for f in plan.app_files: + writing, stage = f.device_uri + ".writing", f.device_uri + ".stage" + report(f"Renaming {writing} to {stage} ...", done / total) + try: + client.rename_file(writing, stage) + except Exception as e: + raise RuntimeError(f"Renaming {writing} to {stage}: {e}") from e + plan.staged = True + + if target.delete_slot_id is not None: + for pattern in ("fonts/*", "images/*"): + report(f"Deleting {pattern} on the device ...", done / total) + try: + client.invoke(target.platform_comp_id, target.delete_slot_id, + sox.STR_ID, pattern) # an asStr Buf goes as a Str + except Exception as e: + raise RuntimeError(f"Deleting {pattern}: {e}") from e + + for f in plan.media: + put(f.device_uri, f) + report("Done", 1.0) diff --git a/app_state.py b/app_state.py index 2b18d2b..7ee309f 100644 --- a/app_state.py +++ b/app_state.py @@ -8,7 +8,7 @@ so later rebinding would only ever be seen by the module that did it. import configparser import os -VERSION = "0.0.0.164" +VERSION = "0.0.0.165" PROPERTIES_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "editor.properties") DEFAULT_GRID_SIZE = 10 DEFAULT_ADD_TIMESTAMP = True diff --git a/main.py b/main.py index 41d1743..54ddb41 100644 --- a/main.py +++ b/main.py @@ -19,6 +19,7 @@ import palette import properties import sax_file import schema +import sox import undo import ox_page_view import user_service_view @@ -615,15 +616,22 @@ def _invoke_app_action(slot_name): if slot is None: app_state.set_status(f"Action slot '{slot_name}' not found on app", warn=True) return False + goes_down = slot_name in ("hibernate", "quit", "restart", "reboot") try: ds.sox_client.invoke(0, slot["id"]) app_state.set_status(f"Invoked '{slot_name}' on device") + except sox.SoxException as e: + # A P-ION restarts before it answers: no reply is what a restart looks like. + if not (goes_down and "timeout" in str(e).lower()): + app_state.set_status(f"Invoke '{slot_name}' failed: {e}", warn=True) + return False + app_state.set_status(f"Invoked '{slot_name}'; the device went down without replying") except Exception as e: app_state.set_status(f"Invoke '{slot_name}' failed: {e}", warn=True) return False if slot_name == "save": app_state.mark_clean() - elif slot_name in ("hibernate", "quit", "restart", "reboot"): + elif goes_down: do_device_disconnect() return True @@ -656,9 +664,28 @@ def do_get_with_file(): if not path: return + def finish(result, error): + if error is not None: + app_state.set_status(f"Get with Files failed: {error}", warn=True) + messagebox.showerror("Get with Files", f"The backup failed:{NL}{NL}{error}") + return + app_state.mark_clean() # the app was saved on the device first + _report_backup(result) + + _run_with_progress("Get with Files", + lambda status: app_backup.get_with_file(ds.sox_client, path, status), + finish) + + +def _run_with_progress(title, work, finish): + """Runs work(status) off the main thread behind a small progress window. + + status(text, fraction) updates the window; finish(result, error) then runs on the + main thread, error being the exception work raised, or None. + """ window = app_state.window dlg = tk.Toplevel(window) - dlg.title("Get with Files") + dlg.title(title) dlg.resizable(False, False) dlg.transient(window) dlg.protocol("WM_DELETE_WINDOW", lambda: None) # a transfer cannot be stopped midway @@ -681,42 +708,42 @@ def do_get_with_file(): bar.config(value=fraction) window.after(0, show) - def finish(result, error): + def done(result, error): dlg.grab_release() dlg.destroy() - if error is not None: - app_state.set_status(f"Get with Files failed: {error}", warn=True) - messagebox.showerror("Get with Files", f"The backup failed:{NL}{NL}{error}") - return - app_state.mark_clean() # the app was saved on the device first - _report_backup(result) + finish(result, error) - def work(): + def run(): try: - result = app_backup.get_with_file(ds.sox_client, path, status) + result = work(status) except Exception as e: - window.after(0, lambda e=e: finish(None, e)) + window.after(0, lambda e=e: done(None, e)) else: - window.after(0, lambda: finish(result, None)) + window.after(0, lambda: done(result, None)) - threading.Thread(target=work, daemon=True).start() + threading.Thread(target=run, daemon=True).start() -def _ask_backup_path(ds): - """The save dialog for the zip: /store/apps, the name filled in. +def _device_app_names(ds): + """(deviceName, appName) of the device's app, or None for each it does not give. - The name takes the app's deviceName and appName, read off the device now -- the - app component's config is not loaded until something asks for it. + Read off the device now -- the app component's config is not loaded until + something asks for it. """ - device_name, app_name = app_state.device_host, "app" try: comp = ds.sox_client.cache_get(0) if comp is not None: ds.sox_client.update([comp], device_session.CONFIG) - device_name = comp.values.get("deviceName") or device_name - app_name = comp.values.get("appName") or app_name + return comp.values.get("deviceName") or None, comp.values.get("appName") or None except Exception as e: app_state.set_status(f"Could not read the app's names: {e}", warn=True) + return None, None + + +def _ask_backup_path(ds): + """The save dialog for the zip: /store/apps, named after the app.""" + device_name, app_name = _device_app_names(ds) + device_name, app_name = device_name or app_state.device_host, app_name or "app" folder = app_backup.backup_folder() try: @@ -751,6 +778,109 @@ def _report_backup(result): (messagebox.showwarning if warn else messagebox.showinfo)("Get with Files", NL.join(lines)) +def do_put_with_files(): + """App > Put with Files: puts a Get with Files zip back onto the connected device. + + The zip is checked against the device first: the same platform, and an app the + device's kits can run -- the backup's app.sax encoded for them, as Ontrol's tool + does, or its app.sab. The app goes staged, so the device takes it up when + restarted, which is offered once the files are in. + """ + ds = app_state.device_session + if not app_state.live_mode or ds is None or not ds.is_connected: + messagebox.showinfo("Put with Files", "Connect to a device first.") + return + if not app_state.SEDONA_HOME: + messagebox.showerror("Put with Files", "The Sedona home folder is not set (Preferences).") + return + + folder = app_backup.backup_folder() + path = filedialog.askopenfilename( + title="Put with Files", + initialdir=folder if os.path.isdir(folder) else app_state.SEDONA_HOME, + filetypes=[("Sedona backup files", "*.zip"), ("All files", "*.*")], + ) + if not path: + return + + try: + target = app_backup.device_target(ds) + except Exception as e: + messagebox.showerror("Put with Files", f"Could not read the device:{NL}{NL}{e}") + return + plan = app_backup.plan_put(path, target) + if not plan.ok: + app_state.set_status(f"{os.path.basename(path)} does not match the device", warn=True) + messagebox.showerror( + "Put with Files", + f"{os.path.basename(path)} cannot be put on this device:{NL}{NL}" + + NL.join(plan.problems)) + return + if not messagebox.askokcancel("Put with Files", _describe_put(plan), + icon=messagebox.WARNING, default=messagebox.CANCEL): + return + + def finish(_result, error): + if error is not None: + app_state.set_status(f"Put with Files failed: {error}", warn=True) + messagebox.showerror( + "Put with Files", + f"The put failed:{NL}{NL}{error}{NL}{NL}" + + ("The new app is staged, but not all of its fonts and images are on the " + "device. Put the backup again before restarting it." + if plan.staged else + "app.sab.stage was not written; the device keeps its current app.")) + return + app_state.set_status(f"Put {os.path.basename(path)}; restart the device to take it up") + if messagebox.askyesno( + "Put with Files", + f"{len(plan.app_files) + len(plan.media)} files written to the device." + f"{NL}{NL}Restart the device now to complete provisioning?"): + if _invoke_app_action("restart"): + app_state.set_status(f"Put {os.path.basename(path)}; the device is restarting") + + _run_with_progress("Put with Files", + lambda status: app_backup.put_with_files(ds.sox_client, plan, target, + status), + finish) + + +def _describe_put(plan): + """The confirmation for a put: what goes to the device, and what that replaces.""" + images = sum(f.device_uri.startswith("images/") for f in plan.media) + fonts = [f for f in plan.media if f.device_uri.startswith("fonts/")] + lines = [f"Put {os.path.basename(plan.path)} on {app_state.device_host}?", "", + "The device's app is replaced; anything not saved on it is lost." + + (" Its fonts and images are deleted and replaced by the backup's." + if plan.has_screen else ""), "", + f" app.sab, from {plan.app_source}"] + lines += [f" {line}" for line in plan.reencoded] + if plan.kit_changes: + lines += [" kits.scode, since the device's kits cannot load the app:"] + lines += [f" {line}" for line in plan.kit_changes] + lines += [" The device's firmware must support these kits, or the app will not " + "start."] + if not plan.has_screen: + return NL.join(lines + ([""] + plan.warnings if plan.warnings else [])) + lines.append(f" {images} images" + + (f" ({plan.converted} made from the backup's image bank)" + if plan.converted else "")) + lines.append(f" {len(fonts)} fonts" + + (f" ({len(plan.fonts_from_zip)} from the backup, the rest from Sedona home)" + if plan.fonts_from_zip else " from Sedona home" if fonts else "")) + if any(f.device_uri.startswith("system/") for f in plan.media): + lines.append(" boot logo") + for title, items in (("The backup has no usable copy of these images:", + plan.images_missing), + ("Neither Sedona home nor the backup has these fonts:", + plan.fonts_missing)): + if items: + lines += ["", title] + [f" {item}" for item in items] + if plan.warnings: + lines += [""] + plan.warnings + return NL.join(lines) + + NL = chr(10) # newline, for the multi-line message boxes @@ -2653,6 +2783,7 @@ def _on_connected(ds, root_comp): menu_device.entryconfig("Disconnect", state="normal") menu_device.entryconfig("Download to SAX...", state="normal") menu_app_info.entryconfig("Get with Files...", state="normal") + menu_app_info.entryconfig("Put with Files...", state="normal") app_state.set_status(f"Connected to {app_state.device_host}:{app_state.device_port}") sax_root = ds.build_dom_tree(root_comp) @@ -2720,6 +2851,7 @@ def do_device_disconnect(): menu_device.entryconfig("Disconnect", state="disabled") menu_device.entryconfig("Download to SAX...", state="disabled") menu_app_info.entryconfig("Get with Files...", state="disabled") + menu_app_info.entryconfig("Put with Files...", state="disabled") app_state.set_status("Disconnected") sax_file.reset_to_startup_state() @@ -3008,6 +3140,7 @@ menu_app_info = tk.Menu(menu_bar, tearoff=0) menu_app_info.add_command(label="App Info", command=show_app_info) menu_app_info.add_separator() menu_app_info.add_command(label="Get with Files...", command=do_get_with_file, state="disabled") +menu_app_info.add_command(label="Put with Files...", command=do_put_with_files, state="disabled") menu_bar.add_cascade(label="App", menu=menu_app_info) menu_prefs = tk.Menu(menu_bar, tearoff=0) diff --git a/sab_app.py b/sab_app.py index feedaa1..0b7d5d4 100644 --- a/sab_app.py +++ b/sab_app.py @@ -1,4 +1,5 @@ -"""Reads a device's app.sab and writes it back out as app.sax. +"""Reads a device's app.sab and writes it back out as app.sax, and encodes an app.sax +into app.sab (encode_sax). Ported from Py_sab2sax (sax.py/sab.py), itself a port of sedona.offline.OfflineApp (decodeAppBinary + encodeAppXml); its output matches `sedonac app.sab` byte for byte, @@ -408,3 +409,215 @@ def _path_of(c): parts.append(c["name"]) c = c["parent"] return "/" + "/".join(reversed(parts)) if parts else "/" + + +# --- encoding app.sax to app.sab ------------------------------------------------------ +# +# Ported from Py_sab2sax's sab.py (sedona.offline.OfflineApp decodeAppXml + +# encodeAppBinary), with the value parsing brought in line with the Java Value classes: +# asStr text is unescaped as TextUtil.fromLiteral does, "nullbool"/"nullfloat" are +# read, and integers take Integer.decode's '#' hex and leading-zero octal. + +def _decode_int(raw): + """Integer.decode.""" + if raw in (None, "", "null"): + return 0 + s = raw.strip() + neg = s.startswith("-") + if s[:1] in "+-": + s = s[1:] + if s[:2].lower() == "0x": + v = int(s[2:], 16) + elif s.startswith("#"): + v = int(s[1:], 16) + elif len(s) > 1 and s.startswith("0"): + v = int(s[1:], 8) + else: + v = int(s) + return -v if neg else v + + +def _from_literal(s): + """TextUtil.fromLiteral.""" + if "\\" not in s: + return s + escapes = {"0": "\0", "n": "\n", "r": "\r", "t": "\t", "\\": "\\", '"': '"', "$": "$"} + out, i = [], 0 + while i < len(s): + c = s[i] + if c == "\\": + i += 1 + if i >= len(s) or s[i] not in escapes: + raise SabError(f"unknown escape sequence in {s!r}") + c = escapes[s[i]] + out.append(c) + i += 1 + return "".join(out) + + +class _Writer: + def __init__(self): + self.b = bytearray() + + def u1(self, v): self.b.append(v & 0xFF) + def u2(self, v): self.b += struct.pack(">H", v & 0xFFFF) + def i4(self, v): self.b += struct.pack(">i", ((v + 2**31) % 2**32) - 2**31) + def i8(self, v): self.b += struct.pack(">q", v) + def f4(self, v): self.b += struct.pack(">f", _to_f32(v)) + def f8(self, v): self.b += struct.pack(">d", v) + + def str_(self, s): + self.b += s.encode("ascii") + self.u1(0) + + +def _encode_value(out, slot, raw): + """One config prop, from its sax text (or the manifest default when absent).""" + vtype, flags = slot.type, slot.flags + if "s" in flags: # asStr Buf: u2 len (incl NUL) + data = _from_literal(raw or "").encode("ascii") + out.u2(len(data) + 1) + out.b += data + out.u1(0) + elif vtype == "bool": + if raw in ("null", "nullbool"): + out.u1(2) + elif raw in (None, "", "false", "true"): + out.u1(1 if raw == "true" else 0) + else: + raise SabError(f"invalid bool {raw!r}") + elif vtype == "byte": + out.u1(_decode_int(raw)) + elif vtype == "short": + out.u2(_decode_int(raw)) + elif vtype == "int": + out.i4(_decode_int(raw)) + elif vtype == "long": + out.i8(_decode_int(raw)) + elif vtype in ("float", "double"): + v = float("nan") if raw in ("null", "nullfloat", "nulldouble") else _flt(raw) + (out.f4 if vtype == "float" else out.f8)(v) + elif vtype in ("sys::Buf", "buf"): + data = base64.b64decode(raw) if raw else b"" + out.u2(len(data)) + out.b += data + else: + raise SabError(f"unhandled slot type {vtype}") + + +def encode_sax(sax, checksums=None): + """app.sax (text or bytes) -> app.sab bytes, as `sedonac app.sax` writes it. + + checksums, {kit name: checksum hex}, stands in for the sax's own -- how an app goes + to a device whose kits are other builds than the ones it was saved with. Every + kit's manifest must be in Sedona home under the checksum used. + """ + root = ET.fromstring(sax) + schema_elem, app_elem = root.find("schema"), root.find("app") + if schema_elem is None or app_elem is None: + raise SabError("not a Sedona app: no or ") + parts = [] + for k in schema_elem.findall("kit"): + name = k.get("name") + cks = (checksums or {}).get(name) or k.get("checksum") + if not cks: + raise SabError(f"no checksum for kit '{name}'") + parts.append((name, f"{int(cks, 16):08x}")) + parts.sort(key=lambda p: ("" if p[0] == "sys" else "\x01" + p[0])) # sortKits + for name, cks in parts: + path = schema.find_manifest_path(name, cks) + if path is None or not path.endswith(f"-{cks}.xml"): + raise SabError(f"no manifest for kit '{name}' {cks} in Sedona home") + kits, types_by_q = _load_kits(parts) + type_ids = {} + for ki, ((kname, _), tlist) in enumerate(zip(parts, kits)): + for t in tlist: + type_ids[f"{kname}::{t.get('name')}"] = (ki, int(t.get("id"))) + cache = {} + + comps = [] + + def walk(elem, qtype, parent): + if qtype not in type_ids: + raise SabError(f"unknown type '{qtype}' ({elem.get('name')})") + c = {"elem": elem, "type": qtype, "parent": parent, "kids": [], + "id": int(elem.get("id")) if elem.get("id") else None} + comps.append(c) + if parent is not None: + parent["kids"].append(c) + for kid in elem.findall("comp"): + walk(kid, kid.get("type"), c) + return c + + app = walk(app_elem, "sys::App", None) + app["id"] = 0 + used = {c["id"] for c in comps if c["id"] is not None} + next_id = 1 + for c in comps: # ids for comps the sax gives none + if c["id"] is None: + while next_id in used: + next_id += 1 + c["id"] = next_id + used.add(next_id) + by_id = {} + for c in comps: + if c["id"] in by_id: + raise SabError(f"duplicate id: {c['id']}") + by_id[c["id"]] = c + + out = _Writer() + out.i4(0x73617070) # "sapp" + out.i4(0x0003) # version 0.3 + out.u1(len(parts)) + for name, cks in parts: + out.str_(name) + out.i4(int(cks, 16)) + out.u2(max(by_id)) + + for cid in sorted(by_id): + c = by_id[cid] + ki, ti = type_ids[c["type"]] + out.u2(cid) + out.u1(ki) + out.u1(ti) + out.str_(c["elem"].get("name", "app") if cid else "app") + out.u2(c["parent"]["id"] if c["parent"] else 0xFFFF) + out.u2(c["kids"][0]["id"] if c["kids"] else 0xFFFF) + sibling = 0xFFFF + if c["parent"]: + kids = c["parent"]["kids"] + i = kids.index(c) + if i + 1 < len(kids): + sibling = kids[i + 1]["id"] + out.u2(sibling) + props = {p.get("name"): p.get("val") for p in c["elem"].findall("prop")} + for slot in _resolve_slots(c["type"], types_by_q, cache): + if "a" in slot.flags or "c" not in slot.flags: + continue # config props only + _encode_value(out, slot, props.get(slot.name, slot.default)) + out.u1(ord(";")) + out.u2(0xFFFF) + + def ref(text): + path, slot_name = text.rsplit(".", 1) + cur = app + for part in (p for p in path.split("/") if p): + cur = next((k for k in cur["kids"] if k["elem"].get("name") == part), None) + if cur is None: + raise SabError(f"link to a missing component: {text}") + names = [s.name for s in _resolve_slots(cur["type"], types_by_q, cache)] + if slot_name not in names: + raise SabError(f"link to a missing slot: {text}") + return cur["id"], names.index(slot_name) + + links = root.find("links") + for link in (links.findall("link") if links is not None else ()): + fc, fs = ref(link.get("from")) + tc, ts = ref(link.get("to")) + out.u2(fc) + out.u1(fs) + out.u2(tc) + out.u1(ts) + out.u2(0xFFFF) + out.u1(ord(".")) + return bytes(out.b) diff --git a/sox.py b/sox.py index a8e6f72..0faf3cc 100644 --- a/sox.py +++ b/sox.py @@ -384,7 +384,7 @@ class SoxClient: # file transfer: one at a time; its chunks arrive unasked, so the receiver # hands them to whichever transfer is open self._xfer_lock = threading.Lock() - self._xfer = None # _FileGet while a get is open + self._xfer = None # _FileGet or _FilePut while a transfer is open # ── Connect / Disconnect ──────────────────────────────────── @@ -989,6 +989,77 @@ class SoxClient: progress(file_size, file_size) return bytes(data) + def put_file(self, uri, data, progress=None): + """Writes `data` to the device as `uri`. + + Ported from FileTransfer.putFile, as Ontrol's BSoxFilePut does it: an 'f' + request opens the transfer, the chunks go out as 'k' messages -- the DASP send + window paces them -- and the device answers the last with an unsolicited 'z' + once the file is closed. progress(done_bytes, total_bytes) is called from this + thread. + """ + if self.is_closed: + raise SoxException("SoxClient closed") + with self._xfer_lock: + # 21, not FileTransfer's 18: room for Orion's message-length header too. + chunk_size = max(16, self._session.ideal_max - 21) + req = _prepare_request(ord('f')) + req.write_str("p") + req.write_str(uri) + req.write_i4(len(data)) + req.write_u2(chunk_size) + req.write_u1(0) # end of headers + + xfer = _FilePut() + self._xfer = xfer + opened = False + try: + res = self.request(req) + res.check_response('F') + opened = True + file_size = res.i4() + chunk_size = res.u2() + if file_size != len(data) or chunk_size <= 0: + raise SoxException( + f"The device opened {uri} for {file_size} bytes in chunks of " + f"{chunk_size}; {len(data)} bytes were offered") + num_chunks = max(1, -(-file_size // chunk_size)) + for num in range(num_chunks): + if self.is_closed: + raise SoxException(f"Session closed while writing {uri}") + chunk = data[num * chunk_size:(num + 1) * chunk_size] + msg = _prepare_request(ord('k'), 0) + msg.write_u2(num) + msg.write_u2(len(chunk)) + msg.write_bytes(chunk) + self._session.send(msg.to_bytes()) + if progress: + progress(min(file_size, (num + 1) * chunk_size), file_size) + + timeout = getattr(self._session, "receive_timeout", 30.0) or 30.0 + deadline = time.monotonic() + timeout + while not xfer.closed.wait(0.5): + if self.is_closed: + raise SoxException(f"Session closed while writing {uri}") + if time.monotonic() > deadline: + raise SoxException(f"Timed out waiting for the device to close {uri}") + opened = False + finally: + self._xfer = None + # Left open, the transfer keeps the device answering "busy" to the next. + if opened and not self.is_closed: + try: + self.request(_prepare_request(ord('z'))) + except Exception: + pass + + def rename_file(self, from_uri, to_uri): + """Renames a file on the device ('b'), as SoxClient.renameFile.""" + req = _prepare_request(ord('b')) + req.write_str(from_uri) + req.write_str(to_uri) + self.request(req).check_response('B') + # ── Networking ────────────────────────────────────────────── def request(self, req): @@ -1111,6 +1182,13 @@ class _FileGet: # SoxReceiver — daemon thread # ──────────────────────────────────────────────────────────────── +class _FilePut: + """The put in progress; `closed` is set when the device's 'z' arrives.""" + + def __init__(self): + self.closed = threading.Event() + + class _SoxReceiver(threading.Thread): def __init__(self, client): super().__init__(name="SoxReceiver", daemon=True) @@ -1136,10 +1214,12 @@ class _SoxReceiver(threading.Thread): client._receive_event(msg) elif cmd == ord('k'): xfer = client._xfer - if xfer is not None: + if isinstance(xfer, _FileGet): xfer.receive_chunk(msg) elif cmd == ord('z'): - pass # close of a put -- the editor only gets files + xfer = client._xfer # the device closing a put + if isinstance(xfer, _FilePut): + xfer.closed.set() else: client._receive_response(msg) except DaspException: