The save and file-open requests used their own fixed resend intervals (5 s / 15 s) and the hello asked for 256-byte packets. Ontrol's Sedona Java stack does neither and works with the P-ION: the standard backoff (1, 2, 4, 8 s) sends a busy device a few copies with the same seqNum, which its receive window drops. Both are removed. Also fixes a race in SendWindow.send(): the packet entered the window before its payload and sent_time were set, so housekeeping could resend it with an empty payload, and the device would then drop the real one as a duplicate. Tested on a P-ION at 192.168.1.252, App > Get with File: save 3.2 s (2 resends), app.sab + kits.scode in 11.3 s with 500-byte chunks (was 14.6 s at 256), device stayed up, files identical to a 256-byte backup. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
262 lines
9.4 KiB
Python
262 lines
9.4 KiB
Python
"""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}"
|
|
|
|
|
|
def save_app(client):
|
|
"""Invokes App.save on the device and waits for it to finish.
|
|
|
|
Ontrol's Niagara driver does this before every app.sab get.
|
|
"""
|
|
comp = client.cache_get(0) or client.load(0)
|
|
slots = client.slot_registry.get((comp.kit_id, comp.type_id)) or []
|
|
slot = next((s for s in slots if s["name"] == "save" and s.get("is_action")), None)
|
|
client.invoke(0, slot["id"] if slot else 0)
|
|
|
|
|
|
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`.
|
|
|
|
The app is saved on the device first, so app.sab holds what is running. Runs off
|
|
the main thread; status(text, fraction) reports progress. Returns a BackupResult;
|
|
a failure to save, or 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)
|
|
|
|
report("Saving the app on the device ...", 0.0)
|
|
try:
|
|
save_app(client)
|
|
except Exception as e:
|
|
raise RuntimeError(f"Saving the app on the device: {e}") from e
|
|
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
|