Add App > Get with File: full app backup zip from the device
Reads app.sab and kits.scode over Sox file transfer, decodes app.sab to app.sax (byte-identical to sedonac), and on Orion/Rion also fetches the images and fonts the app names, with their sources from Sedona home's banks. Offers to save the app on the device first when it is dirty, and asks where to save with <Sedona home>/store/apps and app-<device>-<app>-<timestamp>.zip filled in. DASP trace lines are now printed by their own thread: printing from the network threads to an unread stdout pipe blocked the receiver and made file transfers time out. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 5.5
+244
@@ -0,0 +1,244 @@
|
||||
"""App > Get with File: a full backup of the connected device's app as one zip.
|
||||
|
||||
The zip follows the layout Ontrol's tools read back on a restore:
|
||||
|
||||
device/app.sab the files as the device holds them
|
||||
device/kits.scode
|
||||
device/images/<name>.bin
|
||||
device/fonts/<font file>
|
||||
device/system/logo.bin
|
||||
app.sax app.sab decoded, as sedonac would write it
|
||||
imageBank[XL]/<name>.png the sources of the device's images and fonts, taken from
|
||||
fontBank[XL]/<font file> Sedona home's banks (XL for Orion's 800x480 screen)
|
||||
system/logo.png
|
||||
|
||||
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.
|
||||
"""
|
||||
|
||||
import datetime
|
||||
import os
|
||||
import re
|
||||
import zipfile
|
||||
|
||||
import app_state
|
||||
import sab_app
|
||||
from ox_widget_properties import IMAGE_EXTENSIONS, ORION_KIT, 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")
|
||||
_UNSAFE_FILE_CHARS = re.compile(r'[<>:"/\\|?*\x00-\x1f]')
|
||||
|
||||
|
||||
class _Platform:
|
||||
def __init__(self, name, image_bank, font_bank, font_extensions):
|
||||
self.name = name
|
||||
self.image_bank = image_bank
|
||||
self.font_bank = font_bank
|
||||
self.font_extensions = font_extensions
|
||||
|
||||
|
||||
# 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",))
|
||||
|
||||
|
||||
def platform_of(app):
|
||||
kits = app.kit_names()
|
||||
if ORION_KIT in kits:
|
||||
return _ORION
|
||||
if RION_KIT in kits:
|
||||
return _RION
|
||||
return None
|
||||
|
||||
|
||||
def backup_folder():
|
||||
return os.path.join(app_state.SEDONA_HOME, "store", "apps")
|
||||
|
||||
|
||||
def backup_file_name(device_name, app_name, when=None):
|
||||
"""'app-<device name>-<app name>-<yyyyMMdd-HHmmss>.zip', made safe for Windows."""
|
||||
stamp = (when or datetime.datetime.now()).strftime("%Y%m%d-%H%M%S")
|
||||
parts = [_UNSAFE_FILE_CHARS.sub("_", p).strip(" .") or "unnamed"
|
||||
for p in (device_name, app_name)]
|
||||
return f"app-{parts[0]}-{parts[1]}-{stamp}.zip"
|
||||
|
||||
|
||||
class MediaFile:
|
||||
"""One image or font the app names: where it lives on the device and in the bank.
|
||||
|
||||
`required` is False for guesses -- a bold font's plain cut, Orion's other font
|
||||
orientation, an indexed page image -- whose absence is not worth reporting.
|
||||
`group` ties the alternatives of one need together: the need is met when any of
|
||||
them is found.
|
||||
"""
|
||||
|
||||
def __init__(self, device_uri, bank_folder, bank_name, group, required=True):
|
||||
self.device_uri = device_uri
|
||||
self.bank_folder = bank_folder
|
||||
self.bank_name = bank_name
|
||||
self.group = group
|
||||
self.required = required
|
||||
|
||||
|
||||
def _font_stem(app, comp):
|
||||
slot = app.slot(comp, "fontName")
|
||||
names = [n.strip() for n in (slot.facets.get("range") or "Tahoma").split(",")] if slot else ["Tahoma"]
|
||||
index = app.prop(comp, "fontName", 0) or 0
|
||||
family = names[index] if 0 <= index < len(names) else names[0]
|
||||
size = app.prop(comp, "fontSize", 16)
|
||||
bold = app.prop(comp, "fontAttributes", 0) == 1
|
||||
return f"{size}pt {family}", bold
|
||||
|
||||
|
||||
def media_files(app, platform):
|
||||
"""The images and fonts the app names, each with its device path and bank source."""
|
||||
files, seen = [], set()
|
||||
|
||||
def add(uri, bank_folder, bank_name, group, required=True):
|
||||
if uri not in seen:
|
||||
seen.add(uri)
|
||||
files.append(MediaFile(uri, bank_folder, bank_name, group, required))
|
||||
|
||||
def add_image(name, required=True):
|
||||
if name:
|
||||
add(f"images/{name}.bin", platform.image_bank, name, f"image {name}", required)
|
||||
|
||||
def add_font(stem, group, required):
|
||||
for i, ext in enumerate(platform.font_extensions):
|
||||
add(f"fonts/{stem}{ext}", platform.font_bank, stem + ext, group,
|
||||
required and i == 0)
|
||||
|
||||
for comp in app.comps.values():
|
||||
for slot_name in _IMAGE_SLOTS:
|
||||
add_image(app.prop(comp, slot_name, ""))
|
||||
|
||||
prefix = app.prop(comp, "imageFileNamePrefix", "")
|
||||
if prefix:
|
||||
for i in range((app.prop(comp, "maxIndex", 1) or 0) + 1):
|
||||
add_image(f"{prefix}{i}")
|
||||
|
||||
page_image = app.prop(comp, "imageFileName", "")
|
||||
if page_image:
|
||||
for i in range(1, (app.prop(comp, "maxImageId", 0) or 0) + 1):
|
||||
add_image(f"{page_image}{i}", required=False)
|
||||
|
||||
if app.slot(comp, "fontSize") is not None:
|
||||
stem, bold = _font_stem(app, comp)
|
||||
group = f"font {'bold ' if bold else ''}{stem}"
|
||||
if bold:
|
||||
add_font(f"bold {stem}", group, True)
|
||||
add_font(stem, group, not bold)
|
||||
|
||||
add(LOGO_URI, platform.image_bank, "logo", "logo", required=False)
|
||||
return files
|
||||
|
||||
|
||||
def _bank_source(bank_folder, bank_name, is_image):
|
||||
"""The local bank file for a device file, or None."""
|
||||
folder = os.path.join(app_state.SEDONA_HOME, "ontrol", bank_folder)
|
||||
if not is_image:
|
||||
path = os.path.join(folder, bank_name)
|
||||
return path if os.path.isfile(path) else None
|
||||
for ext in _SOURCE_IMAGE_EXTENSIONS + (".bin",):
|
||||
path = os.path.join(folder, bank_name + ext)
|
||||
if os.path.isfile(path):
|
||||
return path
|
||||
return None
|
||||
|
||||
|
||||
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:
|
||||
return "system/logo" + os.path.splitext(file_name)[1]
|
||||
return f"{media.bank_folder}/{file_name}"
|
||||
|
||||
|
||||
class BackupResult:
|
||||
def __init__(self):
|
||||
self.path = None
|
||||
self.entries = []
|
||||
self.missing_on_device = [] # media the app names that the device lacks
|
||||
self.missing_in_bank = [] # device files with no source in Sedona home
|
||||
self.sax_error = None # why app.sax could not be written
|
||||
|
||||
|
||||
def get_with_file(client, path, status=None):
|
||||
"""Reads the app and its media off the device and writes the backup zip to `path`.
|
||||
|
||||
Runs off the main thread; status(text, fraction) reports progress. Returns a
|
||||
BackupResult; a failure to read app.sab or kits.scode raises.
|
||||
"""
|
||||
def report(text, fraction=None):
|
||||
if status:
|
||||
status(text, fraction)
|
||||
|
||||
def fetch(uri, base, span):
|
||||
def progress(done, total):
|
||||
report(f"Reading {uri} ...", base + span * (done / total if total else 1.0))
|
||||
report(f"Reading {uri} ...", base)
|
||||
try:
|
||||
return client.get_file(uri, progress)
|
||||
except Exception as e:
|
||||
raise RuntimeError(f"Reading {uri}: {e}") from e
|
||||
|
||||
result = BackupResult()
|
||||
entries = [] # (zip name, bytes or local path)
|
||||
|
||||
sab = fetch(APP_SAB, 0.0, 0.15)
|
||||
entries.append(("device/" + APP_SAB, sab))
|
||||
entries.append(("device/" + KITS_SCODE, fetch(KITS_SCODE, 0.15, 0.35)))
|
||||
|
||||
app = None
|
||||
try:
|
||||
app = sab_app.SabApp(sab)
|
||||
entries.append(("app.sax", app.to_sax().encode("ascii")))
|
||||
except Exception as e:
|
||||
result.sax_error = str(e)
|
||||
|
||||
platform = platform_of(app) if app else None
|
||||
media = media_files(app, platform) if platform else []
|
||||
found_groups, wanted = set(), {}
|
||||
for i, m in enumerate(media):
|
||||
report(f"Reading {m.device_uri} ...", 0.5 + 0.45 * i / max(1, len(media)))
|
||||
if m.required:
|
||||
wanted.setdefault(m.group, m.device_uri)
|
||||
try:
|
||||
data = client.get_file(m.device_uri)
|
||||
except Exception:
|
||||
if client.is_closed:
|
||||
raise
|
||||
continue
|
||||
found_groups.add(m.group)
|
||||
entries.append(("device/" + m.device_uri, data))
|
||||
source = _bank_source(m.bank_folder, m.bank_name, m.device_uri.startswith(("images/", "system/")))
|
||||
if source:
|
||||
entries.append((_source_entry(m, source), source))
|
||||
else:
|
||||
result.missing_in_bank.append(f"{m.bank_folder}/{m.bank_name}")
|
||||
result.missing_on_device = [uri for group, uri in wanted.items() if group not in found_groups]
|
||||
|
||||
report("Writing the zip ...", 0.97)
|
||||
os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
|
||||
partial = path + ".part"
|
||||
try:
|
||||
with zipfile.ZipFile(partial, "w", zipfile.ZIP_DEFLATED) as z:
|
||||
for name, content in entries:
|
||||
if isinstance(content, bytes):
|
||||
z.writestr(name, content)
|
||||
else:
|
||||
z.write(content, name)
|
||||
os.replace(partial, path)
|
||||
finally:
|
||||
if os.path.exists(partial):
|
||||
os.remove(partial)
|
||||
|
||||
result.path = path
|
||||
result.entries = [name for name, _ in entries]
|
||||
report("Done", 1.0)
|
||||
return result
|
||||
+1
-1
@@ -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.141"
|
||||
VERSION = "0.0.0.142"
|
||||
PROPERTIES_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "editor.properties")
|
||||
DEFAULT_GRID_SIZE = 10
|
||||
DEFAULT_ADD_TIMESTAMP = True
|
||||
|
||||
@@ -98,7 +98,56 @@ def _trace(direction, msg, datagram=None):
|
||||
hexdump = ""
|
||||
if datagram:
|
||||
hexdump = " [" + datagram.hex(" ") + "]"
|
||||
print(f"[DASP] {ts} {direction} {name}{seq}{ack} payload={plen}{hexdump}")
|
||||
_trace_out(f"[DASP] {ts} {direction} {name}{seq}{ack} payload={plen}{hexdump}")
|
||||
|
||||
|
||||
# Trace lines are printed by a thread of their own. The send and receive threads must
|
||||
# never wait on the console: when stdout is a pipe nobody reads, print blocks once the
|
||||
# pipe fills, and a file transfer's burst of chunks filled it in seconds -- the
|
||||
# receiver stopped and every request timed out. A full queue drops lines instead.
|
||||
_trace_queue = queue.Queue(maxsize=5000)
|
||||
_trace_thread = None
|
||||
_trace_dropped = 0
|
||||
|
||||
|
||||
def _trace_out(line):
|
||||
global _trace_thread, _trace_dropped
|
||||
if _trace_thread is None:
|
||||
_trace_thread = threading.Thread(target=_trace_writer, name="DaspTrace", daemon=True)
|
||||
_trace_thread.start()
|
||||
try:
|
||||
_trace_queue.put_nowait(line)
|
||||
except queue.Full:
|
||||
_trace_dropped += 1
|
||||
|
||||
|
||||
def _trace_writer():
|
||||
global _trace_dropped
|
||||
while True:
|
||||
line = _trace_queue.get()
|
||||
if _trace_dropped:
|
||||
dropped, _trace_dropped = _trace_dropped, 0
|
||||
line = f"[DASP] ... {dropped} trace lines dropped (console not keeping up)\n{line}"
|
||||
_write_stdout(line + "\n")
|
||||
|
||||
|
||||
def _write_stdout(text):
|
||||
"""Writes to the process's stdout handle directly, past Python's buffered writer.
|
||||
|
||||
The buffered writer has a lock, and a daemon thread holding it while the
|
||||
interpreter shuts down aborts the process ("could not acquire lock for stdout").
|
||||
"""
|
||||
import sys
|
||||
stream = sys.__stdout__
|
||||
if stream is None: # pythonw: no console to write to
|
||||
return
|
||||
try:
|
||||
fd = stream.fileno()
|
||||
data = text.encode(stream.encoding or "utf-8", errors="replace")
|
||||
while data:
|
||||
data = data[os.write(fd, data):]
|
||||
except (OSError, ValueError):
|
||||
pass
|
||||
|
||||
|
||||
_ERROR_HINTS = {
|
||||
|
||||
@@ -7,6 +7,7 @@ from tkinter import filedialog, font as tkfont, messagebox, ttk
|
||||
|
||||
import threading
|
||||
|
||||
import app_backup
|
||||
import app_state
|
||||
import connect_dialog
|
||||
import device_session
|
||||
@@ -579,38 +580,171 @@ def do_download_to_sax():
|
||||
|
||||
|
||||
def _invoke_app_action(slot_name):
|
||||
"""Invokes an action slot on the app component (comp_id 0) of the connected device."""
|
||||
"""Invokes an action slot on the app component (comp_id 0) of the connected device.
|
||||
|
||||
Returns whether the device took it; why not is on the status bar.
|
||||
"""
|
||||
ds = app_state.device_session
|
||||
if ds is None or not ds.is_connected:
|
||||
app_state.set_status("Not connected to a device", warn=True)
|
||||
return
|
||||
return False
|
||||
comp = ds.sox_client.cache_get(0)
|
||||
if comp is None:
|
||||
app_state.set_status("App component not loaded", warn=True)
|
||||
return
|
||||
return False
|
||||
slots = ds.sox_client.slot_registry.get((comp.kit_id, comp.type_id))
|
||||
if slots is None:
|
||||
app_state.set_status("No slot registry for the app component", warn=True)
|
||||
return
|
||||
return False
|
||||
slot = next((s for s in slots if s["name"] == slot_name and s.get("is_action")), None)
|
||||
if slot is None:
|
||||
app_state.set_status(f"Action slot '{slot_name}' not found on app", warn=True)
|
||||
return
|
||||
return False
|
||||
try:
|
||||
ds.sox_client.invoke(0, slot["id"])
|
||||
app_state.set_status(f"Invoked '{slot_name}' on device")
|
||||
except Exception as e:
|
||||
app_state.set_status(f"Invoke '{slot_name}' failed: {e}", warn=True)
|
||||
return
|
||||
return False
|
||||
if slot_name == "save":
|
||||
app_state.mark_clean()
|
||||
elif slot_name in ("hibernate", "quit", "restart", "reboot"):
|
||||
do_device_disconnect()
|
||||
return True
|
||||
|
||||
|
||||
_APP_ACTIONS = ["save", "hibernate", "quit", "restart", "reboot"]
|
||||
|
||||
|
||||
def do_get_with_file():
|
||||
"""App > Get with File: backs the device's app up to a zip in <Sedona home>/store/apps.
|
||||
|
||||
The zip carries the device's app.sab, so changes not yet saved on the device are
|
||||
offered a save first. The transfer runs off the main thread behind a small progress
|
||||
window.
|
||||
"""
|
||||
ds = app_state.device_session
|
||||
if not app_state.live_mode or ds is None or not ds.is_connected:
|
||||
messagebox.showinfo("Get with File", "Connect to a device first.")
|
||||
return
|
||||
if not app_state.SEDONA_HOME:
|
||||
messagebox.showerror("Get with File", "The Sedona home folder is not set (Preferences).")
|
||||
return
|
||||
|
||||
if app_state.is_dirty:
|
||||
answer = messagebox.askyesnocancel(
|
||||
"Unsaved Changes",
|
||||
"The app has changes that are not saved on the device yet." + NL + NL +
|
||||
"Save the app on the device before getting it?" + NL +
|
||||
"(No gets the app as it was last saved.)",
|
||||
)
|
||||
if answer is None:
|
||||
return
|
||||
if answer and not _invoke_app_action("save"):
|
||||
messagebox.showerror(
|
||||
"Get with File",
|
||||
"The device did not save the app, so nothing was fetched. "
|
||||
"The status bar says why.")
|
||||
return
|
||||
|
||||
path = _ask_backup_path(ds)
|
||||
if not path:
|
||||
return
|
||||
|
||||
window = app_state.window
|
||||
dlg = tk.Toplevel(window)
|
||||
dlg.title("Get with File")
|
||||
dlg.resizable(False, False)
|
||||
dlg.transient(window)
|
||||
dlg.protocol("WM_DELETE_WINDOW", lambda: None) # a transfer cannot be stopped midway
|
||||
frame = ttk.Frame(dlg, padding=15)
|
||||
frame.pack(fill=tk.BOTH, expand=True)
|
||||
label = ttk.Label(frame, text="Starting ...", width=50)
|
||||
label.pack(anchor="w")
|
||||
bar = ttk.Progressbar(frame, length=360, maximum=1.0)
|
||||
bar.pack(fill=tk.X, pady=(8, 0))
|
||||
dlg.update_idletasks()
|
||||
dlg.geometry(f"+{window.winfo_x() + (window.winfo_width() - dlg.winfo_width()) // 2}"
|
||||
f"+{window.winfo_y() + (window.winfo_height() - dlg.winfo_height()) // 2}")
|
||||
dlg.grab_set()
|
||||
|
||||
def status(text, fraction):
|
||||
def show():
|
||||
if dlg.winfo_exists():
|
||||
label.config(text=text)
|
||||
if fraction is not None:
|
||||
bar.config(value=fraction)
|
||||
window.after(0, show)
|
||||
|
||||
def finish(result, error):
|
||||
dlg.grab_release()
|
||||
dlg.destroy()
|
||||
if error is not None:
|
||||
app_state.set_status(f"Get with File failed: {error}", warn=True)
|
||||
messagebox.showerror("Get with File", f"The backup failed:{NL}{NL}{error}")
|
||||
return
|
||||
_report_backup(result)
|
||||
|
||||
def work():
|
||||
try:
|
||||
result = app_backup.get_with_file(ds.sox_client, path, status)
|
||||
except Exception as e:
|
||||
window.after(0, lambda e=e: finish(None, e))
|
||||
else:
|
||||
window.after(0, lambda: finish(result, None))
|
||||
|
||||
threading.Thread(target=work, daemon=True).start()
|
||||
|
||||
|
||||
def _ask_backup_path(ds):
|
||||
"""The save dialog for the zip: <Sedona home>/store/apps, the name filled in.
|
||||
|
||||
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.
|
||||
"""
|
||||
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
|
||||
except Exception as e:
|
||||
app_state.set_status(f"Could not read the app's names: {e}", warn=True)
|
||||
|
||||
folder = app_backup.backup_folder()
|
||||
try:
|
||||
os.makedirs(folder, exist_ok=True)
|
||||
except OSError:
|
||||
folder = app_state.SEDONA_HOME
|
||||
return filedialog.asksaveasfilename(
|
||||
title="Get with File",
|
||||
initialdir=folder,
|
||||
initialfile=app_backup.backup_file_name(device_name, app_name),
|
||||
defaultextension=".zip",
|
||||
filetypes=[("Zip archive", "*.zip"), ("All files", "*.*")],
|
||||
)
|
||||
|
||||
|
||||
def _report_backup(result):
|
||||
"""Says where the zip went, and warns of anything the app names but could not get."""
|
||||
lines = [f"Saved {os.path.basename(result.path)}", f"in {os.path.dirname(result.path)}",
|
||||
"", f"{len(result.entries)} files in the zip."]
|
||||
warn = False
|
||||
if result.sax_error:
|
||||
warn = True
|
||||
lines += ["", f"app.sax could not be written: {result.sax_error}"]
|
||||
if result.missing_on_device:
|
||||
warn = True
|
||||
lines += ["", "The app names these files, but the device does not have them:"]
|
||||
lines += [f" {uri}" for uri in result.missing_on_device]
|
||||
if result.missing_in_bank:
|
||||
lines += ["", "No source for these in Sedona home's banks; only the device copy is in the zip:"]
|
||||
lines += [f" {name}" for name in result.missing_in_bank]
|
||||
app_state.set_status(f"Got {os.path.basename(result.path)}", warn=warn)
|
||||
(messagebox.showwarning if warn else messagebox.showinfo)("Get with File", NL.join(lines))
|
||||
|
||||
|
||||
NL = chr(10) # newline, for the multi-line message boxes
|
||||
|
||||
|
||||
@@ -2372,6 +2506,7 @@ def _on_connected(ds, root_comp):
|
||||
app_state.window.title(app_state.base_window_title_live())
|
||||
menu_device.entryconfig("Disconnect", state="normal")
|
||||
menu_device.entryconfig("Download to SAX...", state="normal")
|
||||
menu_app_info.entryconfig("Get with File...", 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)
|
||||
@@ -2438,6 +2573,7 @@ def do_device_disconnect():
|
||||
menu_device.entryconfig("Discover...", state="normal")
|
||||
menu_device.entryconfig("Disconnect", state="disabled")
|
||||
menu_device.entryconfig("Download to SAX...", state="disabled")
|
||||
menu_app_info.entryconfig("Get with File...", state="disabled")
|
||||
app_state.set_status("Disconnected")
|
||||
|
||||
sax_file.reset_to_startup_state()
|
||||
@@ -2693,6 +2829,8 @@ menu_bar.add_cascade(label="Kits", menu=menu_kits)
|
||||
|
||||
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 File...", command=do_get_with_file, state="disabled")
|
||||
menu_bar.add_cascade(label="App", menu=menu_app_info)
|
||||
|
||||
menu_prefs = tk.Menu(menu_bar, tearoff=0)
|
||||
|
||||
+410
@@ -0,0 +1,410 @@
|
||||
"""Reads a device's app.sab and writes it back out as app.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,
|
||||
down to the `<!-- /path -->` comments and Java's float formatting. Kit manifests come
|
||||
from Sedona home, found the way the editor finds them everywhere else.
|
||||
|
||||
A .sab holds config props only, so runtime values cannot come back: like sedonac, the
|
||||
manifest default stands in for them, and a value equal to its default is not written.
|
||||
"""
|
||||
|
||||
import base64
|
||||
import math
|
||||
import struct
|
||||
import xml.etree.ElementTree as ET
|
||||
|
||||
import schema
|
||||
|
||||
|
||||
class SabError(Exception):
|
||||
pass
|
||||
|
||||
|
||||
class _Reader:
|
||||
def __init__(self, data):
|
||||
self.b, self.p = data, 0
|
||||
|
||||
def take(self, n):
|
||||
if self.p + n > len(self.b):
|
||||
raise SabError(f"truncated sab at offset {self.p}")
|
||||
v = self.b[self.p:self.p + n]
|
||||
self.p += n
|
||||
return v
|
||||
|
||||
def u1(self): return self.take(1)[0]
|
||||
def u2(self): return struct.unpack(">H", self.take(2))[0]
|
||||
def i4(self): return struct.unpack(">i", self.take(4))[0]
|
||||
def i8(self): return struct.unpack(">q", self.take(8))[0]
|
||||
def f4(self): return struct.unpack(">f", self.take(4))[0]
|
||||
def f8(self): return struct.unpack(">d", self.take(8))[0]
|
||||
|
||||
def str_(self):
|
||||
end = self.b.index(0, self.p)
|
||||
s = self.b[self.p:end].decode("ascii")
|
||||
self.p = end + 1
|
||||
return s
|
||||
|
||||
|
||||
# --- schema ---------------------------------------------------------------------------
|
||||
|
||||
class Slot:
|
||||
__slots__ = ("name", "type", "flags", "default", "facets")
|
||||
|
||||
def __init__(self, elem):
|
||||
self.name = elem.get("name")
|
||||
self.type = elem.get("type")
|
||||
self.flags = elem.get("flags", "")
|
||||
self.default = elem.get("default")
|
||||
self.facets = {f.get("name"): f.get("val") for f in elem.iter()
|
||||
if f is not elem and f.get("name") is not None}
|
||||
|
||||
|
||||
def _load_kits(parts):
|
||||
"""[(kitName, checksumHex)] in kit-id order -> (types by kit, types by qname)."""
|
||||
kits, types_by_q = [], {}
|
||||
for name, cks in parts:
|
||||
path = schema.find_manifest_path(name, cks)
|
||||
if path is None:
|
||||
raise SabError(f"no manifest for kit '{name}' in Sedona home")
|
||||
mroot = ET.parse(path).getroot()
|
||||
tlist = sorted(mroot.findall("type"), key=lambda t: int(t.get("id")))
|
||||
for t in tlist:
|
||||
types_by_q[f"{name}::{t.get('name')}"] = t
|
||||
kits.append(tlist)
|
||||
return kits, types_by_q
|
||||
|
||||
|
||||
def _resolve_slots(qname, types_by_q, cache):
|
||||
"""The type's slots, base chain first, indexed by slot id."""
|
||||
if qname in cache:
|
||||
return cache[qname]
|
||||
t = types_by_q[qname]
|
||||
base = t.get("base")
|
||||
out = list(_resolve_slots(base, types_by_q, cache)) if base else []
|
||||
by_name = {s.name: i for i, s in enumerate(out)}
|
||||
for elem in sorted(t.findall("slot"), key=lambda s: int(s.get("id"))):
|
||||
slot = Slot(elem)
|
||||
if slot.name in by_name: # an action override keeps the inherited id
|
||||
out[by_name[slot.name]] = slot
|
||||
else:
|
||||
by_name[slot.name] = len(out)
|
||||
out.append(slot)
|
||||
cache[qname] = out
|
||||
return out
|
||||
|
||||
|
||||
# --- values ---------------------------------------------------------------------------
|
||||
|
||||
def _int(raw): # Integer.decode: supports 0x
|
||||
if raw in (None, "", "null"):
|
||||
return 0
|
||||
return int(raw, 0) if raw.lower().startswith(("0x", "-0x")) else int(raw)
|
||||
|
||||
|
||||
def _flt(raw):
|
||||
if raw == "null":
|
||||
return float("nan") # explicit null only
|
||||
if raw in (None, ""):
|
||||
return 0.0 # Value.defaultForType -> zero
|
||||
return float(raw)
|
||||
|
||||
|
||||
def _decode_value(r, vtype, flags):
|
||||
if "s" in flags: # asStr Buf: u2 len (incl NUL)
|
||||
n = r.u2()
|
||||
return r.take(n).split(b"\0")[0].decode("ascii", errors="replace")
|
||||
if vtype == "bool":
|
||||
v = r.u1()
|
||||
return None if v == 2 else (v == 1)
|
||||
if vtype == "byte": return r.u1()
|
||||
if vtype == "short": return r.u2()
|
||||
if vtype == "int": return r.i4()
|
||||
if vtype == "long": return r.i8()
|
||||
if vtype == "float": return r.f4()
|
||||
if vtype == "double": return r.f8()
|
||||
if vtype in ("sys::Buf", "buf"):
|
||||
return bytes(r.take(r.u2()))
|
||||
raise SabError(f"unhandled slot type {vtype}")
|
||||
|
||||
|
||||
def _default_value(slot):
|
||||
"""The manifest default, or Value.defaultForType's zero."""
|
||||
vtype, flags, raw = slot.type, slot.flags, slot.default
|
||||
if "s" in flags:
|
||||
return raw or ""
|
||||
if vtype == "bool":
|
||||
return None if raw == "null" else raw in ("true", "1")
|
||||
if vtype in ("byte", "short", "int", "long"):
|
||||
return _int(raw)
|
||||
if vtype == "float":
|
||||
# the default is a 32-bit float too: 0.1 as a double is not the file's 0.1
|
||||
return _to_f32(_flt(raw))
|
||||
if vtype == "double":
|
||||
return _flt(raw)
|
||||
if vtype in ("sys::Buf", "buf"):
|
||||
return base64.b64decode(raw) if raw else b""
|
||||
raise SabError(f"unhandled slot type {vtype}")
|
||||
|
||||
|
||||
def _same(a, b):
|
||||
if isinstance(a, float) and isinstance(b, float):
|
||||
return (math.isnan(a) and math.isnan(b)) or a == b
|
||||
return type(a) is type(b) and a == b
|
||||
|
||||
|
||||
# --- Java number formatting -----------------------------------------------------------
|
||||
|
||||
_F32_MAX = 3.4028234663852886e38
|
||||
_F32_INF_EDGE = 3.4028235677973366e38 # midpoint to 2**128: below rounds to F32_MAX
|
||||
|
||||
|
||||
def _to_f32(v):
|
||||
try:
|
||||
return struct.unpack(">f", struct.pack(">f", v))[0]
|
||||
except OverflowError:
|
||||
if abs(v) < _F32_INF_EDGE:
|
||||
return math.copysign(_F32_MAX, v)
|
||||
return math.copysign(math.inf, v)
|
||||
|
||||
|
||||
def _shortest(a, single):
|
||||
"""Fewest digits that still round-trip -- what Java's toString picks."""
|
||||
for prec in range(0, 17):
|
||||
s = "%.*e" % (prec, a)
|
||||
v = float(s)
|
||||
if single:
|
||||
v = _to_f32(v)
|
||||
if v == a:
|
||||
break
|
||||
mant, exp = s.split("e")
|
||||
digits = mant.replace(".", "").rstrip("0") or "0"
|
||||
return digits, int(exp)
|
||||
|
||||
|
||||
def _java_float(v, single=True):
|
||||
"""Float/Double.toString: always a decimal point, E-form outside [1e-3, 1e7)."""
|
||||
if math.isnan(v):
|
||||
return "NaN"
|
||||
if math.isinf(v):
|
||||
return "-Infinity" if v < 0 else "Infinity"
|
||||
neg = math.copysign(1.0, v) < 0
|
||||
a = abs(v)
|
||||
if a == 0.0:
|
||||
return "-0.0" if neg else "0.0"
|
||||
digits, exp = _shortest(a, single)
|
||||
if -3 <= exp < 7:
|
||||
if exp >= 0:
|
||||
head = digits[:exp + 1].ljust(exp + 1, "0")
|
||||
tail = digits[exp + 1:] or "0"
|
||||
else:
|
||||
head, tail = "0", "0" * (-exp - 1) + digits
|
||||
out = head + "." + tail
|
||||
else:
|
||||
out = digits[0] + "." + (digits[1:] or "0") + "E" + str(exp)
|
||||
return "-" + out if neg else out
|
||||
|
||||
|
||||
def _encode_string(val, slot):
|
||||
"""Value.encodeString()."""
|
||||
vtype, flags = slot.type, slot.flags
|
||||
if "s" in flags:
|
||||
return _literal(val)
|
||||
if vtype == "bool":
|
||||
return "null" if val is None else ("true" if val else "false")
|
||||
if vtype == "float":
|
||||
return "null" if math.isnan(val) else _java_float(val, True)
|
||||
if vtype == "double":
|
||||
return "null" if math.isnan(val) else _java_float(val, False)
|
||||
if vtype in ("sys::Buf", "buf"):
|
||||
return base64.b64encode(val).decode("ascii")
|
||||
return str(val)
|
||||
|
||||
|
||||
def _literal(s):
|
||||
"""TextUtil.toLiteral."""
|
||||
return "".join({"\0": "\\0", "\n": "\\n", "\r": "\\r", "\t": "\\t",
|
||||
"\\": "\\\\", '"': '\\"', "$": "\\$"}.get(c, c) for c in s)
|
||||
|
||||
|
||||
def _safe(s):
|
||||
"""XWriter.safe: escape markup, quotes, and anything outside printable ASCII."""
|
||||
out = []
|
||||
for c in s:
|
||||
o = ord(c)
|
||||
if o < 0x20 or o > 0x7e or c in "'\"":
|
||||
out.append("&#x%x;" % o)
|
||||
elif c == "<":
|
||||
out.append("<")
|
||||
elif c == ">":
|
||||
out.append(">")
|
||||
elif c == "&":
|
||||
out.append("&")
|
||||
else:
|
||||
out.append(c)
|
||||
return "".join(out)
|
||||
|
||||
|
||||
# --- the app --------------------------------------------------------------------------
|
||||
|
||||
class SabApp:
|
||||
"""A decoded app.sab: kits, components by id, and links.
|
||||
|
||||
Each component is a dict: id, type, name, parent, kids, and props -- a list of
|
||||
(Slot, value) for every config prop, defaults included.
|
||||
"""
|
||||
|
||||
def __init__(self, data):
|
||||
r = _Reader(data)
|
||||
if r.i4() != 0x73617070:
|
||||
raise SabError("not a .sab (bad magic)")
|
||||
ver = r.i4()
|
||||
if ver != 0x0003:
|
||||
raise SabError(f"unsupported sab version 0x{ver:x}")
|
||||
|
||||
self.kit_parts = []
|
||||
for _ in range(r.u1()):
|
||||
name = r.str_()
|
||||
self.kit_parts.append((name, f"{r.i4() & 0xFFFFFFFF:08x}"))
|
||||
kits, self._types_by_q = _load_kits(self.kit_parts)
|
||||
self._cache = {}
|
||||
r.u2() # maxId, recomputed on the way out
|
||||
|
||||
qname_of = {}
|
||||
for ki, ((kname, _), tlist) in enumerate(zip(self.kit_parts, kits)):
|
||||
for t in tlist:
|
||||
qname_of[(ki, int(t.get("id")))] = f"{kname}::{t.get('name')}"
|
||||
|
||||
by_id = {}
|
||||
while True:
|
||||
cid = r.u2()
|
||||
if cid == 0xFFFF:
|
||||
break
|
||||
ki, ti = r.u1(), r.u1()
|
||||
if (ki, ti) not in qname_of:
|
||||
raise SabError(f"comp {cid}: no type {ti} in kit {ki}")
|
||||
qname = qname_of[(ki, ti)]
|
||||
c = {"id": cid, "type": qname, "name": r.str_(), "kids": [], "props": []}
|
||||
c["parentId"], c["childId"], c["siblingId"] = r.u2(), r.u2(), r.u2()
|
||||
for slot in self.slots(qname):
|
||||
if "a" in slot.flags or "c" not in slot.flags:
|
||||
continue # config props only, in slot-id order
|
||||
c["props"].append((slot, _decode_value(r, slot.type, slot.flags)))
|
||||
if r.u1() != ord(";"):
|
||||
raise SabError(f"corrupted component {cid} {c['name']}")
|
||||
if cid in by_id:
|
||||
raise SabError(f"duplicate id: {cid}")
|
||||
by_id[cid] = c
|
||||
|
||||
def must(cid, what, owner):
|
||||
if cid not in by_id:
|
||||
raise SabError(f"comp {owner['id']} {owner['name']}: missing {what} {cid}")
|
||||
return by_id[cid]
|
||||
|
||||
for c in by_id.values():
|
||||
c["parent"] = must(c["parentId"], "parent", c) if c["parentId"] != 0xFFFF else None
|
||||
if c["childId"] != 0xFFFF:
|
||||
kid = must(c["childId"], "child", c)
|
||||
while True:
|
||||
c["kids"].append(kid)
|
||||
if kid["siblingId"] == 0xFFFF:
|
||||
break
|
||||
kid = must(kid["siblingId"], "nextSibling", kid)
|
||||
|
||||
self.links = []
|
||||
while True:
|
||||
fc = r.u2()
|
||||
if fc == 0xFFFF:
|
||||
break
|
||||
fs, tc, ts = r.u1(), r.u2(), r.u1()
|
||||
for cid in (fc, tc):
|
||||
if cid not in by_id:
|
||||
raise SabError(f"link references unknown comp id {cid}")
|
||||
self.links.append((by_id[fc], fs, by_id[tc], ts))
|
||||
if r.u1() != ord("."):
|
||||
raise SabError("invalid app end marker")
|
||||
|
||||
if 0 not in by_id:
|
||||
raise SabError("no app component (id 0)")
|
||||
self.comps = by_id
|
||||
self.app = by_id[0]
|
||||
|
||||
def slots(self, qname):
|
||||
return _resolve_slots(qname, self._types_by_q, self._cache)
|
||||
|
||||
def kit_names(self):
|
||||
return [name for name, _ in self.kit_parts]
|
||||
|
||||
def prop(self, comp, name, default=None):
|
||||
for slot, val in comp["props"]:
|
||||
if slot.name == name:
|
||||
return val
|
||||
return default
|
||||
|
||||
def slot(self, comp, name):
|
||||
return next((s for s in self.slots(comp["type"]) if s.name == name), None)
|
||||
|
||||
# --- writing app.sax ---
|
||||
|
||||
def to_sax(self):
|
||||
out = []
|
||||
w = out.append
|
||||
w("<?xml version='1.0'?>\n")
|
||||
w("<sedonaApp>\n")
|
||||
w("<schema>\n")
|
||||
for name, cks in self.kit_parts:
|
||||
w(f' <kit name="{_safe(name)}" checksum="{cks}" />\n')
|
||||
w("</schema>\n")
|
||||
w("<app>\n")
|
||||
self._write_props(w, self.app, 2)
|
||||
for kid in self.app["kids"]:
|
||||
self._write_comp(w, kid, 2)
|
||||
w("</app>\n")
|
||||
w("<links>\n")
|
||||
for fc, fs, tc, ts in self.links:
|
||||
w(' <link from="%s.%s" to="%s.%s"/>\n' % (
|
||||
_safe(_path_of(fc)), _safe(self._slot_name(fc, fs)),
|
||||
_safe(_path_of(tc)), _safe(self._slot_name(tc, ts))))
|
||||
w("</links>\n")
|
||||
w("</sedonaApp>\n")
|
||||
return "".join(out)
|
||||
|
||||
def _slot_name(self, comp, slot_id):
|
||||
slots = self.slots(comp["type"])
|
||||
if slot_id >= len(slots):
|
||||
raise SabError(f"{comp['type']} has no slot id {slot_id}")
|
||||
return slots[slot_id].name
|
||||
|
||||
def _write_props(self, w, c, indent):
|
||||
"""Only props whose value differs from the slot default get written."""
|
||||
n = 0
|
||||
for slot, val in c["props"]:
|
||||
if _same(val, _default_value(slot)):
|
||||
continue
|
||||
w('%s<prop name="%s" val="%s"/>\n' % (
|
||||
" " * indent, _safe(slot.name), _safe(_encode_string(val, slot))))
|
||||
n += 1
|
||||
return n
|
||||
|
||||
def _write_comp(self, w, c, indent):
|
||||
pad = " " * indent
|
||||
w(f"{pad}<!-- {_path_of(c)} -->\n")
|
||||
w(f'{pad}<comp name="{_safe(c["name"])}" id="{c["id"]}" type="{_safe(c["type"])}"')
|
||||
body = []
|
||||
n = self._write_props(body.append, c, indent + 2)
|
||||
for kid in c["kids"]:
|
||||
self._write_comp(body.append, kid, indent + 2)
|
||||
if not n and not c["kids"]:
|
||||
w("/>\n")
|
||||
else:
|
||||
w(">\n")
|
||||
w("".join(body))
|
||||
w(f"{pad}</comp>\n")
|
||||
|
||||
|
||||
def _path_of(c):
|
||||
parts = []
|
||||
while c["parent"] is not None:
|
||||
parts.append(c["name"])
|
||||
c = c["parent"]
|
||||
return "/" + "/".join(reversed(parts)) if parts else "/"
|
||||
@@ -381,6 +381,11 @@ class SoxClient:
|
||||
# event callback: called on main thread via queue
|
||||
self.on_event = None # fn(comp_id, what_char)
|
||||
|
||||
# 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
|
||||
|
||||
# ── Connect / Disconnect ────────────────────────────────────
|
||||
|
||||
def connect(self):
|
||||
@@ -902,6 +907,83 @@ class SoxClient:
|
||||
def _get_slots(self, kit_id, type_id_val):
|
||||
return self.slot_registry.get((kit_id, type_id_val))
|
||||
|
||||
# ── File transfer ───────────────────────────────────────────
|
||||
|
||||
def get_file(self, uri, progress=None):
|
||||
"""Reads a file off the device ('app.sab', 'kits.scode', 'images/x.bin').
|
||||
|
||||
Ported from FileTransfer.getFile: an 'f' request opens the transfer, the
|
||||
device then pushes the file as unsolicited 'k' chunks, and a 'z' closes it.
|
||||
A file the device does not have comes back as a SoxException from the open.
|
||||
progress(done_bytes, total_bytes) is called from this thread as chunks land.
|
||||
"""
|
||||
if self.is_closed:
|
||||
raise SoxException("SoxClient closed")
|
||||
with self._xfer_lock:
|
||||
chunk_size = max(16, self._session.ideal_max - 18)
|
||||
req = _prepare_request(ord('f'))
|
||||
req.write_str("g")
|
||||
req.write_str(uri)
|
||||
req.write_i4(0) # fileSize: unknown on a get
|
||||
req.write_u2(chunk_size)
|
||||
req.write_u1(0) # end of headers
|
||||
|
||||
# Set before the open goes out: the first chunks can beat its reply.
|
||||
xfer = _FileGet()
|
||||
self._xfer = xfer
|
||||
opened = closed = False
|
||||
try:
|
||||
try:
|
||||
res = self.request(req)
|
||||
except SoxException as e:
|
||||
# A P-ION on firmware 1.32 never answers a file get -- it drops off
|
||||
# the network for ~20 s, as a restart would -- even for a file it
|
||||
# does not have.
|
||||
raise SoxException(
|
||||
f"The device did not answer the request for {uri} ({e}). Its "
|
||||
f"firmware may not support reading files over Sox; some devices "
|
||||
f"stop responding, or restart, when asked.") from e
|
||||
res.check_response('F')
|
||||
opened = True
|
||||
file_size = res.i4()
|
||||
chunk_size = res.u2()
|
||||
while res.remaining > 0: # response headers -- nothing we use
|
||||
if not res.str():
|
||||
break
|
||||
res.str()
|
||||
num_chunks = -(-file_size // chunk_size) if chunk_size else 0
|
||||
|
||||
timeout = getattr(self._session, "receive_timeout", 30.0) or 30.0
|
||||
while not xfer.done(num_chunks):
|
||||
if self.is_closed:
|
||||
raise SoxException(f"Session closed while reading {uri}")
|
||||
if time.monotonic() - xfer.last_receive > timeout:
|
||||
raise SoxException(f"Timed out reading {uri}")
|
||||
xfer.wait(0.5)
|
||||
if progress:
|
||||
progress(min(file_size, xfer.count() * chunk_size), file_size)
|
||||
|
||||
close = self.request(_prepare_request(ord('z')))
|
||||
closed = True
|
||||
close.check_response('Z')
|
||||
finally:
|
||||
self._xfer = None
|
||||
# Left open, the transfer keeps the device answering "busy" to the next.
|
||||
if opened and not closed and not self.is_closed:
|
||||
try:
|
||||
self.request(_prepare_request(ord('z')))
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
data = bytearray(file_size)
|
||||
for num, chunk in xfer.chunks.items():
|
||||
if num < num_chunks:
|
||||
start = num * chunk_size
|
||||
data[start:start + len(chunk)] = chunk
|
||||
if progress:
|
||||
progress(file_size, file_size)
|
||||
return bytes(data)
|
||||
|
||||
# ── Networking ──────────────────────────────────────────────
|
||||
|
||||
def request(self, req):
|
||||
@@ -986,6 +1068,39 @@ class SoxClient:
|
||||
pass
|
||||
|
||||
|
||||
class _FileGet:
|
||||
"""The chunks of the get in progress, filled by the receiver thread."""
|
||||
|
||||
def __init__(self):
|
||||
self.chunks = {}
|
||||
self.last_receive = time.monotonic()
|
||||
self._event = threading.Event()
|
||||
self._lock = threading.Lock()
|
||||
|
||||
def receive_chunk(self, msg):
|
||||
msg.u1() # 'k'
|
||||
msg.u1() # replyNum
|
||||
num = msg.u2()
|
||||
size = msg.u2()
|
||||
data = msg.read_bytes(size)
|
||||
with self._lock:
|
||||
self.chunks[num] = data
|
||||
self.last_receive = time.monotonic()
|
||||
self._event.set()
|
||||
|
||||
def count(self):
|
||||
with self._lock:
|
||||
return len(self.chunks)
|
||||
|
||||
def done(self, num_chunks):
|
||||
with self._lock:
|
||||
return all(n in self.chunks for n in range(num_chunks))
|
||||
|
||||
def wait(self, seconds):
|
||||
self._event.wait(seconds)
|
||||
self._event.clear()
|
||||
|
||||
|
||||
# ────────────────────────────────────────────────────────────────
|
||||
# SoxReceiver — daemon thread
|
||||
# ────────────────────────────────────────────────────────────────
|
||||
@@ -1013,8 +1128,12 @@ class _SoxReceiver(threading.Thread):
|
||||
cmd = payload[0]
|
||||
if cmd == ord('e'):
|
||||
client._receive_event(msg)
|
||||
elif cmd == ord('k') or cmd == ord('z'):
|
||||
pass # file transfer — not needed for editor
|
||||
elif cmd == ord('k'):
|
||||
xfer = client._xfer
|
||||
if xfer is not None:
|
||||
xfer.receive_chunk(msg)
|
||||
elif cmd == ord('z'):
|
||||
pass # close of a put -- the editor only gets files
|
||||
else:
|
||||
client._receive_response(msg)
|
||||
except DaspException:
|
||||
|
||||
Reference in New Issue
Block a user