[swugenerator] [PATCH 1/2] generator: Implement xz support

[Ernestas Kulik]

---
swugenerator/generator.py | 16 +++++++++++++---
1 file changed, 13 insertions(+), 3 deletions(-)

diff --git a/swugenerator/generator.py b/swugenerator/generator.py
index 4a4ce8e..eb3a8f8 100644
--- a/swugenerator/generator.py
+++ b/swugenerator/generator.py
@@ -83,17 +83,27 @@ class SWUGenerator:

new.newfilename = entry["filename"]

if "compressed" in entry and not self.nocompress:
cmp = entry["compressed"]
- if cmp not in ("zlib", "zstd"):
+ if cmp not in ("xz", "zlib", "zstd"):
logging.critical("Wrong compression algorithm: %s", cmp)
sys.exit(1)

new_path = os.path.join(self.temp.name, new.newfilename) + "." + cmp
new.newfilename = new.newfilename + "." + cmp
- if cmp == "zlib":
+ if cmp == "xz":
+ cmd = [
+ "xz",
+ "-f",
+ "-k",
+ "-c",
+ new.fullfilename,
+ ">",
+ new_path,
+ ]
+ elif cmp == "zlib":
cmd = [
"gzip",
"-f",
"-9",
"-n",
@@ -101,11 +111,11 @@ class SWUGenerator:
"--rsyncable",
new.fullfilename,
">",
new_path,
]
- else:
+ elif cmp == "zstd":
cmd = [
"zstd",
"-z",
"-k",
"-T0",
--
2.47.3

[Ernestas Kulik]

Currently, when using single image in multiple handlers
(e.g. just flashing to several MTD partitions), the decompressed-size
property will only be set for the first one. Moreover, it only being set
for ubivol handlers breaks MTD flashing.
---
swugenerator/generator.py | 7 +++----
1 file changed, 3 insertions(+), 4 deletions(-)

diff --git a/swugenerator/generator.py b/swugenerator/generator.py
index eb3a8f8..95f18df 100644
--- a/swugenerator/generator.py
+++ b/swugenerator/generator.py
@@ -133,14 +133,10 @@ class SWUGenerator:
"Cannot compress %s with %s", entry["filename"], cmd
)
sys.exit(1)

new.fullfilename = new_path
-
- if entry.get("type") == "ubivol":
- entry.setdefault("properties", {}) \
- .update({ "decompressed-size": str(new.getsize()) })
# compression cannot be used with delta, because it has own compressor
elif ("type" in entry) and entry["type"] == "delta":
cmd = [
"zck",
"-u",
@@ -225,10 +221,13 @@ class SWUGenerator:
entry["filename"] = new.newfilename
if not self.nohash:
entry["sha256"] = new.getsha256()
if "encrypted" in entry and entry["encrypted"] is True:
entry["ivt"] = new.ivt
+ if entry.get("compressed") and not self.nocompress:
+ entry.setdefault("properties", {}) \
+ .update({ "decompressed-size": str(new.getsize()) })

def find_files_in_swdesc(self, first):
for n, val in first.items():
if isinstance(val, libconf.AttrDict):
self.find_files_in_swdesc(val)
--
2.47.3

[Mark Jonas]

Hi Ernestas,

I think we could improve the robustness and maintainability a little
by moving the critical error that cmp does not match a supported
compression into the else. Then, we do not need to maintain two lists
with the supported compression.

if cmp == "xz":
...
elif cmp == "zlib":
...
elif cmp == "zstd":
...
else

logging.critical("Wrong compression algorithm: %s", cmp)
sys.exit(1)

IMHO an even better and more pythonic (?) approach (attention,
untested code!) could be:

compress_cmd_args = {
"xz": ["xz", "-f", "-k", "-c"],
"zlib": ["gzip", "-f", "-9", "-n", "-c", "--rsyncable"],
"zstd": ["zstd", "-z", "-k", "-T0", "-f", "-c"],
}

if cmp not in compress_cmd_args:

logging.critical("Wrong compression algorithm: %s", cmp)
sys.exit(1)

cmd = compress_cmd_args[cmp] + [new.fullfilename, ">", new_path]

Cheers,
Mark

[Ernestas Kulik]

On Sun, 2025-12-07 at 10:17 +0100, Mark Jonas wrote:
> Hi Ernestas,

Hi, Mark,

> I think we could improve the robustness and maintainability a little
> by moving the critical error that cmp does not match a supported
> compression into the else. Then, we do not need to maintain two lists
> with the supported compression.
>
> if cmp == "xz":
>  ...
> elif cmp == "zlib":
>  ...
> elif cmp == "zstd":
>  ...
> else
>     logging.critical("Wrong compression algorithm: %s", cmp)
>     sys.exit(1)
>
> IMHO an even better and more pythonic (?) approach (attention,
> untested code!) could be:
>
> compress_cmd_args = {
>     "xz": ["xz", "-f", "-k", "-c"],
>     "zlib": ["gzip", "-f", "-9", "-n", "-c", "--rsyncable"],
>     "zstd": ["zstd", "-z", "-k", "-T0", "-f", "-c"],
> }
>
> if cmp not in compress_cmd_args:
>     logging.critical("Wrong compression algorithm: %s", cmp)
>     sys.exit(1)
>
> cmd = compress_cmd_args[cmp] + [new.fullfilename, ">", new_path]

I’ll see what can be arranged, thanks for taking a look. :)

[Ernestas Kulik]

While swupdate is capable of decompressing XZ archives, swugenerator
currently does not allow its use in compression. This commit adds
support for it and refactors the compressed entry processing for
maintainability.
---
swugenerator/generator.py | 78 ++++++++++++++++-----------------------
1 file changed, 32 insertions(+), 46 deletions(-)

diff --git a/swugenerator/generator.py b/swugenerator/generator.py
index 4a4ce8e..706b850 100644
--- a/swugenerator/generator.py
+++ b/swugenerator/generator.py
@@ -64,10 +64,40 @@ class SWUGenerator:
def close(self):
self.temp.cleanup()
self.cpiofile.add_trailer()
self.out.close()

+ def process_compressed_entry(self, entry, cmp, new):
+ cmds = {
+ "xz": ["xz", "-f", "-k", "-c"],
+ "zlib": ["gzip", "-f", "-9", "-n", "-c", "--rsyncable"],
+ "zstd": ["zstd", "-z", "-k", "-T0", "-f", "-c"],
+ }
+ cmd = cmds.get(cmp)
+ if not cmd:
+ logging.critical("Wrong compression algorithm: %s", cmp)
+ sys.exit(1)
+
+ new_path = os.path.join(self.temp.name, new.newfilename) + "." + cmp
+ new.newfilename = new.newfilename + "." + cmp
+
+ cmd.extend([new.fullfilename, ">", new_path])
+
+ try:
+ subprocess.run(" ".join(cmd), shell=True, check=True, text=True)
+ except subprocess.CalledProcessError:
+ logging.critical(
+ "Cannot compress %s with %s", entry["filename"], cmd
+ )
+ sys.exit(1)
+
+ new.fullfilename = new_path
+
+ if entry.get("type") == "ubivol":

+ entry.setdefault("properties", {}) \
+ .update({ "decompressed-size": str(new.getsize()) })

+
def process_entry(self, entry):
if "filename" not in entry:
return
new = None
for image in self.artifacts:
@@ -81,56 +111,12 @@ class SWUGenerator:
logging.critical("Artifact %s not found", entry["filename"])
sys.exit(22)

new.newfilename = entry["filename"]

- if "compressed" in entry and not self.nocompress:
- cmp = entry["compressed"]

- if cmp not in ("zlib", "zstd"):

- logging.critical("Wrong compression algorithm: %s", cmp)
- sys.exit(1)
-
- new_path = os.path.join(self.temp.name, new.newfilename) + "." + cmp
- new.newfilename = new.newfilename + "." + cmp

- if cmp == "zlib":

- cmd = [
- "gzip",
- "-f",
- "-9",
- "-n",
- "-c",
- "--rsyncable",
- new.fullfilename,
- ">",
- new_path,
- ]
- else:
- cmd = [
- "zstd",
- "-z",
- "-k",
- "-T0",
- "-f",
- "-c",
- new.fullfilename,
- ">",
- new_path,
- ]
-
- try:
- subprocess.run(" ".join(cmd), shell=True, check=True, text=True)
- except subprocess.CalledProcessError:
- logging.critical(
- "Cannot compress %s with %s", entry["filename"], cmd
- )
- sys.exit(1)
-
- new.fullfilename = new_path

-
- if entry.get("type") == "ubivol":
- entry.setdefault("properties", {}) \
- .update({ "decompressed-size": str(new.getsize()) })

+ if not self.nocompress and (cmp := entry.get("compressed")):
+ self.process_compressed_entry(entry, cmp, new)

# compression cannot be used with delta, because it has own compressor
elif ("type" in entry) and entry["type"] == "delta":
cmd = [
"zck",
"-u",

--
2.47.3

[Ernestas Kulik]

Currently, when using single image in multiple handlers
(e.g. just flashing to several MTD partitions), the decompressed-size
property will only be set for the first one. Moreover, it only being set
for ubivol handlers breaks MTD flashing.
---
swugenerator/generator.py | 7 +++----

1 file changed, 3 insertions(+), 4 deletions(-)

diff --git a/swugenerator/generator.py b/swugenerator/generator.py
index 706b850..6578f75 100644
--- a/swugenerator/generator.py
+++ b/swugenerator/generator.py
@@ -90,14 +90,10 @@ class SWUGenerator:

)
sys.exit(1)

new.fullfilename = new_path

- if entry.get("type") == "ubivol":
- entry.setdefault("properties", {}) \
- .update({ "decompressed-size": str(new.getsize()) })

-

def process_entry(self, entry):
if "filename" not in entry:
return
new = None
for image in self.artifacts:

@@ -201,10 +197,13 @@ class SWUGenerator:

entry["filename"] = new.newfilename
if not self.nohash:
entry["sha256"] = new.getsha256()
if "encrypted" in entry and entry["encrypted"] is True:
entry["ivt"] = new.ivt
+ if entry.get("compressed") and not self.nocompress:

+ entry.setdefault("properties", {}) \
+ .update({ "decompressed-size": str(new.getsize()) })

[Ernestas Kulik]

Currently, when using single image in multiple handlers
(e.g. just flashing to several MTD partitions), the decompressed-size
property will only be set for the first one. Moreover, it only being set
for ubivol handlers breaks MTD flashing.

Signed-off-by: Ernestas Kulik <ernes...@iconn-networks.com>

[Mark Jonas]

Reviewed-by: Mark Jonas <toe...@gmail.com>