Compare commits

...
2 Commits
Author SHA1 Message Date
ArdaandClaude Opus 5.5 08b4dec532 Tree: Reorder... for a component's children
Right-click a component (or the app) with two or more children to move them
up and down; OK applies the new order, which is the order they run in. On a
device the order goes there first via Sox reorder, checked against the
device's children, and the file is left alone if it is refused. Undoable.

Also: device host in editor.properties.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-10 23:39:06 +03:00
ArdaandClaude Opus 5.5 232eea99f4 App > Put with Files: restore a backup zip onto the connected device
Follows Ontrol's Workbench tool (BPutWithFilesTool / PutApp): the backup's
app.sax is encoded for the device's kits and put as app.sab.writing, renamed
to app.sab.stage; the platform service's deleteFile clears fonts/* and
images/*, then images, fonts (from Sedona home's bank, in the device's cut)
and the boot logo go. The zip is checked against the device first: same
platform (Orion, Rion or P-ION, from the platform id), and kits the device
can load. Falls back to the backup's app.sab, with its kits.scode if needed.

- sab_app.encode_sax: .sax -> .sab, ported from Py_sab2sax with Java's value
  parsing (fromLiteral, nullbool/nullfloat, Integer.decode); byte-identical
  to sedonac on the saved backups
- app_backup.native_image: port of OntOxImageUtil's png -> native .bin
- sox.put_file / rename_file
- Get with Files: Orion's boot logo is system/logo.jpg
- A restart that goes unanswered is taken as the device going down

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-10 23:37:45 +03:00
9 changed files with 1180 additions and 42 deletions
+480 -15
View File
@@ -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: 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/kits.scode
device/images/<name>.bin device/images/<name>.bin
device/fonts/<font file> device/fonts/<font file>
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 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 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) fontBank[XL]/<font file> 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 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. 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 import datetime
@@ -23,11 +27,11 @@ import zipfile
import app_state import app_state
import sab_app 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" APP_SAB = "app.sab"
KITS_SCODE = "kits.scode" KITS_SCODE = "kits.scode"
LOGO_URI = "system/logo.bin"
_IMAGE_SLOTS = ("imageFileName", "trueImageFileName", "falseImageFileName") _IMAGE_SLOTS = ("imageFileName", "trueImageFileName", "falseImageFileName")
_SOURCE_IMAGE_EXTENSIONS = tuple(e for e in IMAGE_EXTENSIONS if e != ".bin") _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: 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.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.image_bank = image_bank
self.font_bank = font_bank self.font_bank = font_bank
self.font_extensions = font_extensions 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 keeps a landscape (.oxfl) and a portrait (.oxfp) cut of each font, and its boot
_ORION = _Platform("ORION", "imageBankXL", "fontBankXL", (".oxfl", ".oxfp")) # logo as a jpg; Rion's is a native image (Ontrol's BGetWithFilesTool). A P-ION has no
_RION = _Platform("RION", "imageBank", "fontBank", (".bin",)) # 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): def platform_of(app):
kits = app.kit_names() kits = app.kit_names()
if ORION_KIT in kits: return next((p for p in _PLATFORMS if p.kit in kits), None)
return _ORION
if RION_KIT in kits:
return _RION
return None
def backup_folder(): def backup_folder():
@@ -98,6 +110,8 @@ def _font_stem(app, comp):
def media_files(app, platform): def media_files(app, platform):
"""The images and fonts the app names, each with its device path and bank source.""" """The images and fonts the app names, each with its device path and bank source."""
files, seen = [], set() files, seen = [], set()
if platform.screen is None:
return files
def add(uri, bank_folder, bank_name, group, required=True): def add(uri, bank_folder, bank_name, group, required=True):
if uri not in seen: if uri not in seen:
@@ -134,7 +148,7 @@ def media_files(app, platform):
add_font(f"bold {stem}", group, True) add_font(f"bold {stem}", group, True)
add_font(stem, group, not bold) 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 return files
@@ -154,7 +168,7 @@ def _bank_source(bank_folder, bank_name, is_image):
def _source_entry(media, source_path): def _source_entry(media, source_path):
"""Where the bank file goes in the zip: next to its bank's name, the logo in system/.""" """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) 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 "system/logo" + os.path.splitext(file_name)[1]
return f"{media.bank_folder}/{file_name}" 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] result.entries = [name for name, _ in entries]
report("Done", 1.0) report("Done", 1.0)
return result 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 '<name>.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)
+1 -1
View File
@@ -8,7 +8,7 @@ so later rebinding would only ever be seen by the module that did it.
import configparser import configparser
import os import os
VERSION = "0.0.0.164" VERSION = "0.0.0.166"
PROPERTIES_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "editor.properties") PROPERTIES_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "editor.properties")
DEFAULT_GRID_SIZE = 10 DEFAULT_GRID_SIZE = 10
DEFAULT_ADD_TIMESTAMP = True DEFAULT_ADD_TIMESTAMP = True
+1 -1
View File
@@ -7,7 +7,7 @@ left_pane_width = 230
trace_dasp = True trace_dasp = True
[Device] [Device]
host = 192.168.1.29 host = 192.168.1.251
port = 1876 port = 1876
user = admin user = admin
+37
View File
@@ -155,6 +155,43 @@ def live_rename_component(element, new_name):
return False return False
# ── Reorder Children ────────────────────────────────────────────
def live_reorder_children(parent_element, ordered_children):
"""In live mode, sends a Sox reorder to the device. Returns True on success.
The device takes the order only when it names every child it has, no more and no
fewer, so the list is checked against what the device last said before it is sent.
"""
ds = _session()
if ds is None:
return False
parent_id = _comp_id(parent_element)
if parent_id is None:
_fail(f"'{parent_element.get('name', '?')}' has no component id.")
return False
child_ids = []
for child in ordered_children:
cid = _comp_id(child)
if cid is None:
_fail(f"'{child.get('name', '?')}' has no component id.")
return False
child_ids.append(cid)
client = ds.sox_client
cached = client.cache_get(parent_id)
if cached is not None and set(cached.children) != set(child_ids):
_fail("The editor's list of children no longer matches the device's. "
"Reconnect to read the device again.")
return False
try:
client.reorder(parent_id, child_ids)
return True
except Exception as e:
_fail(f"The device rejected the reorder: {e}")
return False
# ── Link / Unlink ─────────────────────────────────────────────── # ── Link / Unlink ───────────────────────────────────────────────
def _link_ends(ds, from_element, from_slot_name, to_element, to_slot_name): def _link_ends(ds, from_element, from_slot_name, to_element, to_slot_name):
+286 -21
View File
@@ -19,6 +19,7 @@ import palette
import properties import properties
import sax_file import sax_file
import schema import schema
import sox
import undo import undo
import ox_page_view import ox_page_view
import user_service_view import user_service_view
@@ -615,15 +616,22 @@ def _invoke_app_action(slot_name):
if slot is None: if slot is None:
app_state.set_status(f"Action slot '{slot_name}' not found on app", warn=True) app_state.set_status(f"Action slot '{slot_name}' not found on app", warn=True)
return False return False
goes_down = slot_name in ("hibernate", "quit", "restart", "reboot")
try: try:
ds.sox_client.invoke(0, slot["id"]) ds.sox_client.invoke(0, slot["id"])
app_state.set_status(f"Invoked '{slot_name}' on device") 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: except Exception as e:
app_state.set_status(f"Invoke '{slot_name}' failed: {e}", warn=True) app_state.set_status(f"Invoke '{slot_name}' failed: {e}", warn=True)
return False return False
if slot_name == "save": if slot_name == "save":
app_state.mark_clean() app_state.mark_clean()
elif slot_name in ("hibernate", "quit", "restart", "reboot"): elif goes_down:
do_device_disconnect() do_device_disconnect()
return True return True
@@ -656,9 +664,28 @@ def do_get_with_file():
if not path: if not path:
return 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 window = app_state.window
dlg = tk.Toplevel(window) dlg = tk.Toplevel(window)
dlg.title("Get with Files") dlg.title(title)
dlg.resizable(False, False) dlg.resizable(False, False)
dlg.transient(window) dlg.transient(window)
dlg.protocol("WM_DELETE_WINDOW", lambda: None) # a transfer cannot be stopped midway 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) bar.config(value=fraction)
window.after(0, show) window.after(0, show)
def finish(result, error): def done(result, error):
dlg.grab_release() dlg.grab_release()
dlg.destroy() dlg.destroy()
if error is not None: finish(result, error)
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)
def work(): def run():
try: try:
result = app_backup.get_with_file(ds.sox_client, path, status) result = work(status)
except Exception as e: except Exception as e:
window.after(0, lambda e=e: finish(None, e)) window.after(0, lambda e=e: done(None, e))
else: 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): def _device_app_names(ds):
"""The save dialog for the zip: <Sedona home>/store/apps, the name filled in. """(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 Read off the device now -- the app component's config is not loaded until
app component's config is not loaded until something asks for it. something asks for it.
""" """
device_name, app_name = app_state.device_host, "app"
try: try:
comp = ds.sox_client.cache_get(0) comp = ds.sox_client.cache_get(0)
if comp is not None: if comp is not None:
ds.sox_client.update([comp], device_session.CONFIG) ds.sox_client.update([comp], device_session.CONFIG)
device_name = comp.values.get("deviceName") or device_name return comp.values.get("deviceName") or None, comp.values.get("appName") or None
app_name = comp.values.get("appName") or app_name
except Exception as e: except Exception as e:
app_state.set_status(f"Could not read the app's names: {e}", warn=True) 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: <Sedona home>/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() folder = app_backup.backup_folder()
try: try:
@@ -751,6 +778,109 @@ def _report_backup(result):
(messagebox.showwarning if warn else messagebox.showinfo)("Get with Files", NL.join(lines)) (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 NL = chr(10) # newline, for the multi-line message boxes
@@ -1240,6 +1370,10 @@ def on_tree_right_click(event):
if app_element is not None: if app_element is not None:
tree_root_context_menu.add_command( tree_root_context_menu.add_command(
label="Properties", command=lambda: show_properties(app_element)) label="Properties", command=lambda: show_properties(app_element))
if _has_reorderable_children(app_element):
tree_root_context_menu.add_command(
label="Reorder...",
command=lambda: open_reorder_dialog(current_element(app_element)))
tree_root_context_menu.add_separator() tree_root_context_menu.add_separator()
if app_state.live_mode and app_state.device_session and app_state.device_session.is_connected: if app_state.live_mode and app_state.device_session and app_state.device_session.is_connected:
for action_name in _APP_ACTIONS: for action_name in _APP_ACTIONS:
@@ -1263,6 +1397,12 @@ def on_tree_right_click(event):
selection = [app_state.tree_element_map[row] for row in selected_rows selection = [app_state.tree_element_map[row] for row in selected_rows
if row in app_state.tree_element_map] if row in app_state.tree_element_map]
fill_component_popup(tree_comp_context_menu, comp_element, selection) fill_component_popup(tree_comp_context_menu, comp_element, selection)
if _has_reorderable_children(comp_element):
# After Rename..., with the other entries that change the component itself.
rename_index = tree_comp_context_menu.index("Rename...")
tree_comp_context_menu.insert_command(
rename_index + 1, label="Reorder...",
command=lambda: open_reorder_dialog(current_element(comp_element)))
tree_comp_context_menu.post(event.x_root, event.y_root) tree_comp_context_menu.post(event.x_root, event.y_root)
@@ -1676,6 +1816,128 @@ def open_rename_dialog(comp):
entry.select_range(0, tk.END) entry.select_range(0, tk.END)
def _child_display_names(parent, children):
"""Names for a reorder list. On a device a child not yet read is a "..." row, so
its name is read from the device rather than shown as dots."""
names = [child.get("name", "?") for child in children]
if not (app_state.live_mode and "..." in names):
return names
ds = app_state.device_session
try:
ds.load_children(int(parent.get("id")))
except Exception:
return names
for i, child in enumerate(children):
if names[i] == "...":
cached = ds.sox_client.cache_get(int(child.get("id", -1)))
if cached is not None:
names[i] = cached.name
return names
def open_reorder_dialog(parent):
"""Lists a component's children to move up and down; OK applies the new order."""
window = app_state.window
original = [child for child in parent if child.tag == "comp"]
if len(original) < 2:
return
names = dict(zip(map(id, original), _child_display_names(parent, original)))
order = list(original)
dialog = tk.Toplevel(window)
dialog.title(f"Reorder {parent.get('name', 'app')}")
dialog.transient(window)
dialog.grab_set()
dialog.geometry(f"+{window.winfo_x() + 100}+{window.winfo_y() + 60}")
main_frame = tk.Frame(dialog, padx=10, pady=10)
main_frame.pack(fill=tk.BOTH, expand=True)
main_frame.columnconfigure(0, weight=1)
main_frame.rowconfigure(0, weight=1)
list_frame = tk.Frame(main_frame)
list_frame.grid(row=0, column=0, sticky="nsew")
listbox = tk.Listbox(list_frame, width=28, height=min(max(len(order), 10), 20),
exportselection=False, activestyle="none")
scrollbar = ttk.Scrollbar(list_frame, orient=tk.VERTICAL, command=listbox.yview)
listbox.configure(yscrollcommand=scrollbar.set)
listbox.pack(side=tk.LEFT, fill=tk.BOTH, expand=True)
scrollbar.pack(side=tk.RIGHT, fill=tk.Y)
side = tk.Frame(main_frame, padx=10)
side.grid(row=0, column=1, sticky="n")
up_button = ttk.Button(side, text="Move Up", width=14)
up_button.pack(fill=tk.X)
down_button = ttk.Button(side, text="Move Down", width=14)
down_button.pack(fill=tk.X, pady=(6, 0))
reset_button = ttk.Button(side, text="Reset", width=14)
reset_button.pack(fill=tk.X, pady=(40, 0))
btn_frame = tk.Frame(main_frame)
btn_frame.grid(row=1, column=0, columnspan=2, pady=(10, 0))
ok_button = ttk.Button(btn_frame, text="OK", width=10)
ok_button.pack(side=tk.LEFT, padx=5)
ttk.Button(btn_frame, text="Cancel", width=10,
command=dialog.destroy).pack(side=tk.LEFT, padx=5)
def selected_index():
sel = listbox.curselection()
return sel[0] if sel else None
def update_buttons(*_):
index = selected_index()
up_button.state(["!disabled"] if index not in (None, 0) else ["disabled"])
down_button.state(["!disabled"] if index not in (None, len(order) - 1)
else ["disabled"])
reset_button.state(["!disabled"] if order != original else ["disabled"])
def fill(select=None):
listbox.delete(0, tk.END)
for child in order:
listbox.insert(tk.END, names[id(child)])
if select is not None:
listbox.selection_set(select)
listbox.activate(select)
listbox.see(select)
update_buttons()
def move(step):
index = selected_index()
if index is None or not 0 <= index + step < len(order):
return
order[index], order[index + step] = order[index + step], order[index]
fill(index + step)
def reset():
# Back to the order the dialog opened with, keeping the same row selected.
index = selected_index()
picked = order[index] if index is not None else None
order[:] = original
fill(order.index(picked) if picked is not None else None)
def on_ok(*_):
dialog.destroy()
if sax_file.reorder_children(parent, order):
refresh_all_views()
up_button.configure(command=lambda: move(-1))
down_button.configure(command=lambda: move(1))
reset_button.configure(command=reset)
ok_button.configure(command=on_ok)
listbox.bind("<<ListboxSelect>>", update_buttons)
dialog.bind("<Alt-Up>", lambda _: move(-1))
dialog.bind("<Alt-Down>", lambda _: move(1))
dialog.bind("<Return>", on_ok)
dialog.bind("<Escape>", lambda _: dialog.destroy())
fill(0)
listbox.focus_set()
def _has_reorderable_children(element):
return sum(1 for child in element if child.tag == "comp") >= 2
def fill_component_popup(menu, comp, selection=None): def fill_component_popup(menu, comp, selection=None):
"""Fills a popup with the actions that apply to one component. """Fills a popup with the actions that apply to one component.
@@ -2653,6 +2915,7 @@ def _on_connected(ds, root_comp):
menu_device.entryconfig("Disconnect", state="normal") menu_device.entryconfig("Disconnect", state="normal")
menu_device.entryconfig("Download to SAX...", state="normal") menu_device.entryconfig("Download to SAX...", state="normal")
menu_app_info.entryconfig("Get with Files...", 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}") app_state.set_status(f"Connected to {app_state.device_host}:{app_state.device_port}")
sax_root = ds.build_dom_tree(root_comp) sax_root = ds.build_dom_tree(root_comp)
@@ -2720,6 +2983,7 @@ def do_device_disconnect():
menu_device.entryconfig("Disconnect", state="disabled") menu_device.entryconfig("Disconnect", state="disabled")
menu_device.entryconfig("Download to SAX...", state="disabled") menu_device.entryconfig("Download to SAX...", state="disabled")
menu_app_info.entryconfig("Get with Files...", 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") app_state.set_status("Disconnected")
sax_file.reset_to_startup_state() sax_file.reset_to_startup_state()
@@ -3008,6 +3272,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_command(label="App Info", command=show_app_info)
menu_app_info.add_separator() 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="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_bar.add_cascade(label="App", menu=menu_app_info)
menu_prefs = tk.Menu(menu_bar, tearoff=0) menu_prefs = tk.Menu(menu_bar, tearoff=0)
+214 -1
View File
@@ -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 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, (decodeAppBinary + encodeAppXml); its output matches `sedonac app.sab` byte for byte,
@@ -408,3 +409,215 @@ def _path_of(c):
parts.append(c["name"]) parts.append(c["name"])
c = c["parent"] c = c["parent"]
return "/" + "/".join(reversed(parts)) if parts else "/" 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 <schema> or <app>")
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)
+37
View File
@@ -483,6 +483,43 @@ def rename_component(comp_element, new_name):
return True return True
def reorder_children(parent, new_order):
"""Gives `parent`'s child components a new order, which is the order they run in.
Returns True when the order changed. On a device the new order goes there first,
and the file is left alone if it does not.
"""
root = app_state.xml_root_element
if root is None:
return False
old_order = [child for child in parent if child.tag == "comp"]
new_order = list(new_order)
if new_order == old_order:
return False
if (len(new_order) != len(old_order)
or {id(c) for c in new_order} != {id(c) for c in old_order}):
return False
name = parent.get("name", "app")
if app_state.live_mode:
if not live_hooks.live_reorder_children(parent, new_order):
live_hooks.warn_failure("Reorder", f"{name} was not reordered on the device")
return False
undo.apply_child_order(parent, new_order)
command_class = (undo.LiveReorderChildrenCommand if app_state.live_mode
else undo.ReorderChildrenCommand)
undo.push(command_class(parent, old_order, new_order, label=f"reorder {name}"))
app_state.mark_dirty()
refresh_treeview_from_dom(root)
wiresheet.render_current_sheet()
where = " on the device" if app_state.live_mode else ""
app_state.set_status(f"{name} reordered{where}.")
return True
def allocate_component_id(): def allocate_component_id():
"""Lowest unused id, matching how this app's ids are dense with no gaps.""" """Lowest unused id, matching how this app's ids are dense with no gaps."""
used = set() used = set()
+83 -3
View File
@@ -384,7 +384,7 @@ class SoxClient:
# file transfer: one at a time; its chunks arrive unasked, so the receiver # file transfer: one at a time; its chunks arrive unasked, so the receiver
# hands them to whichever transfer is open # hands them to whichever transfer is open
self._xfer_lock = threading.Lock() 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 ──────────────────────────────────── # ── Connect / Disconnect ────────────────────────────────────
@@ -989,6 +989,77 @@ class SoxClient:
progress(file_size, file_size) progress(file_size, file_size)
return bytes(data) 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 ────────────────────────────────────────────── # ── Networking ──────────────────────────────────────────────
def request(self, req): def request(self, req):
@@ -1111,6 +1182,13 @@ class _FileGet:
# SoxReceiver — daemon thread # 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): class _SoxReceiver(threading.Thread):
def __init__(self, client): def __init__(self, client):
super().__init__(name="SoxReceiver", daemon=True) super().__init__(name="SoxReceiver", daemon=True)
@@ -1136,10 +1214,12 @@ class _SoxReceiver(threading.Thread):
client._receive_event(msg) client._receive_event(msg)
elif cmd == ord('k'): elif cmd == ord('k'):
xfer = client._xfer xfer = client._xfer
if xfer is not None: if isinstance(xfer, _FileGet):
xfer.receive_chunk(msg) xfer.receive_chunk(msg)
elif cmd == ord('z'): 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: else:
client._receive_response(msg) client._receive_response(msg)
except DaspException: except DaspException:
+41
View File
@@ -378,3 +378,44 @@ class LiveRenameCommand(RenameComponentCommand):
raise LiveUndoError(f"{self.old_name} was not renamed to {self.new_name} " raise LiveUndoError(f"{self.old_name} was not renamed to {self.new_name} "
"on the device") "on the device")
super().redo() super().redo()
def apply_child_order(parent, ordered_children):
"""Puts `parent`'s <comp> children in the given order; its other children stay put."""
children = list(parent)
comp_slots = [i for i, child in enumerate(children) if child.tag == "comp"]
for slot, child in zip(comp_slots, ordered_children):
children[slot] = child
parent[:] = children
class ReorderChildrenCommand:
"""A new order for a component's children."""
def __init__(self, parent, old_order, new_order, label="reorder"):
self.parent = parent
self.old_order = list(old_order)
self.new_order = list(new_order)
self.label = label
def undo(self):
apply_child_order(self.parent, self.old_order)
def redo(self):
apply_child_order(self.parent, self.new_order)
class LiveReorderChildrenCommand(ReorderChildrenCommand):
"""A reorder on the device: the old order goes back there before the DOM takes it."""
def undo(self):
if not live_hooks.live_reorder_children(self.parent, self.old_order):
raise LiveUndoError(f"The old order of {self.parent.get('name', 'app')} "
"could not be put back on the device")
super().undo()
def redo(self):
if not live_hooks.live_reorder_children(self.parent, self.new_order):
raise LiveUndoError(f"{self.parent.get('name', 'app')} was not reordered "
"on the device")
super().redo()