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