Add x2wincur to reverse the conversion (#2)

.github/workflows/build.yml

@@ -21,7 +21,7 @@ jobs:
           python -m pip install --upgrade pip
           pip install flake8 flake8-import-order mypy wheel coverage
           pip install -r requirements.txt
-          sudo apt-get install x11-apps
+          sudo apt-get install x11-apps dmz-cursor-theme
       - name: Lint with flake8
         run: flake8 .
       - name: Typecheck with mypy
@@ -32,15 +32,20 @@ jobs:
         run: pip install dist/*.whl
       - name: Test with sample/crosshair.cur
         run: |
-          coverage run -m win2xcur.main sample/crosshair.cur -o /tmp
+          coverage run -m win2xcur.main.win2xcur sample/crosshair.cur -o /tmp
           ls -l /tmp/crosshair
       - name: Test with animated cursors
         run: |
           wget http://www.anicursor.com/waiting.zip
           mkdir ani output
           unzip waiting.zip -d ani
-          coverage run -m win2xcur.main -s ani/*.ani -o output
+          coverage run -a -m win2xcur.main.win2xcur -s ani/*.ani -o output
           ls -l output/*
+      - name: Test with dmz-cursor-theme
+        run: |
+          mkdir dmz-white
+          coverage run -a -m win2xcur.main.x2wincur /usr/share/icons/DMZ-White/cursors/* -o dmz-white
+          ls -l dmz-white/*
       - name: Generating coverage report
         run: coverage xml
       - uses: codecov/codecov-action@v1

setup.py

@@ -13,7 +13,8 @@ setup(
 
     entry_points={
         'console_scripts': [
-            'win2xcur = win2xcur.main:main',
+            'win2xcur = win2xcur.main.win2xcur:main',
+            'x2wincur = win2xcur.main.x2wincur:main',
         ],
     },
 

win2xcur/cursor.py

@@ -4,6 +4,9 @@ from wand.sequence import SingleImage
 
 
 class CursorImage:
+    image: SingleImage
+    hotspot: Tuple[int, int]
+
     def __init__(self, image: SingleImage, hotspot: Tuple[int, int]) -> None:
         self.image = image
         self.hotspot = hotspot
@@ -13,6 +16,9 @@ class CursorImage:
 
 
 class CursorFrame:
+    images: List[CursorImage]
+    delay: int
+
     def __init__(self, images: List[CursorImage], delay: int = 0) -> None:
         self.images = images
         self.delay = delay

win2xcur/main/win2xcur.py

@@ -9,7 +9,8 @@ from typing import BinaryIO
 
 from win2xcur import shadow
 from win2xcur.parser import open_blob
-from win2xcur.writer.x11 import check_xcursorgen, to_x11
+from win2xcur.writer import to_x11
+from win2xcur.writer.x11 import check_xcursorgen
 
 
 def main() -> None:

win2xcur/main/x2wincur.py

@@ -0,0 +1,44 @@
+import argparse
+import os
+import sys
+import traceback
+from multiprocessing import cpu_count
+from multiprocessing.pool import ThreadPool
+from threading import Lock
+from typing import BinaryIO
+
+from win2xcur.parser import open_blob
+from win2xcur.writer import to_smart
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description='Converts Windows cursors to X11 cursors.')
+    parser.add_argument('files', type=argparse.FileType('rb'), nargs='+',
+                        help='X11 cursor files to convert (no extension)')
+    parser.add_argument('-o', '--output', '--output-dir', default=os.curdir,
+                        help='Directory to store converted cursor files.')
+
+    args = parser.parse_args()
+    print_lock = Lock()
+
+    def process(file: BinaryIO) -> None:
+        name = file.name
+        blob = file.read()
+        try:
+            cursor = open_blob(blob)
+        except Exception:
+            with print_lock:
+                print('Error occurred while processing %s:' % (name,), file=sys.stderr)
+                traceback.print_exc()
+        else:
+            ext, result = to_smart(cursor.frames)
+            output = os.path.join(args.output, os.path.basename(name) + ext)
+            with open(output, 'wb') as f:
+                f.write(result)
+
+    with ThreadPool(cpu_count()) as pool:
+        pool.map(process, args.files)
+
+
+if __name__ == '__main__':
+    main()

win2xcur/parser/__init__.py

@@ -3,8 +3,11 @@ from typing import List, Type
 from win2xcur.parser.ani import ANIParser
 from win2xcur.parser.base import BaseParser
 from win2xcur.parser.cur import CURParser
+from win2xcur.parser.xcursor import XCursorParser
 
-PARSERS: List[Type[BaseParser]] = [CURParser, ANIParser]
+__all__ = ['ANIParser', 'CURParser', 'XCursorParser', 'PARSERS', 'open_blob']
+
+PARSERS: List[Type[BaseParser]] = [CURParser, ANIParser, XCursorParser]
 
 
 def open_blob(blob: bytes) -> BaseParser:

win2xcur/parser/ani.py

@@ -10,7 +10,12 @@ from win2xcur.parser.cur import CURParser
 class ANIParser(BaseParser):
     SIGNATURE = b'RIFF'
     ANI_TYPE = b'ACON'
+    HEADER_CHUNK = b'anih'
+    LIST_CHUNK = b'LIST'
+    SEQ_CHUNK = b'seq '
+    RATE_CHUNK = b'rate'
     FRAME_TYPE = b'fram'
+    ICON_CHUNK = b'icon'
     RIFF_HEADER = struct.Struct('<4sI4s')
     CHUNK_HEADER = struct.Struct('<4sI')
     ANIH_HEADER = struct.Struct('<IIIIIIIII')
@@ -23,7 +28,10 @@ class ANIParser(BaseParser):
         signature: bytes
         size: int
         subtype: bytes
+        try:
             signature, size, subtype = cls.RIFF_HEADER.unpack(blob[:cls.RIFF_HEADER.size])
+        except struct.error:
+            return False
         return signature == cls.SIGNATURE and size == len(blob) - 8 and subtype == cls.ANI_TYPE
 
     def __init__(self, blob: bytes) -> None:
@@ -49,7 +57,7 @@ class ANIParser(BaseParser):
         return name, size, offset
 
     def _parse(self, offset: int) -> List[CursorFrame]:
-        _, size, offset = self._read_chunk(offset, expected=[b'anih'])
+        _, size, offset = self._read_chunk(offset, expected=[self.HEADER_CHUNK])
 
         if size != self.ANIH_HEADER.size:
             raise ValueError('Unexpected anih header size %d, expected %d' % (size, self.ANIH_HEADER.size))
@@ -57,6 +65,9 @@ class ANIParser(BaseParser):
         size, frame_count, step_count, width, height, bit_count, planes, display_rate, flags = self.ANIH_HEADER.unpack(
             self.blob[offset:offset + self.ANIH_HEADER.size])
 
+        if size != self.ANIH_HEADER.size:
+            raise ValueError('Unexpected size in anih header %d, expected %d' % (size, self.ANIH_HEADER.size))
+
         if not flags & self.ICON_FLAG:
             raise NotImplementedError('Raw BMP images not supported.')
 
@@ -67,8 +78,8 @@ class ANIParser(BaseParser):
         delays = [display_rate for _ in range(step_count)]
 
         while offset < len(self.blob):
-            name, size, offset = self._read_chunk(offset, expected=[b'LIST', b'seq ', b'rate'])
+            name, size, offset = self._read_chunk(offset, expected=[self.LIST_CHUNK, self.SEQ_CHUNK, self.RATE_CHUNK])
-            if name == b'LIST':
+            if name == self.LIST_CHUNK:
                 list_end = offset + size
                 if self.blob[offset:offset + 4] != self.FRAME_TYPE:
                     raise ValueError('Unexpected RIFF list type: %r, expected %r' %
@@ -76,18 +87,20 @@ class ANIParser(BaseParser):
                 offset += 4
 
                 for i in range(frame_count):
-                    _, size, offset = self._read_chunk(offset, expected=[b'icon'])
+                    _, size, offset = self._read_chunk(offset, expected=[self.ICON_CHUNK])
                     frames.append(CURParser(self.blob[offset:offset + size]).frames[0])
                     offset += size
+                    if offset & 1:
+                        offset += 1
 
                 if offset != list_end:
                     raise ValueError('Wrong RIFF list size: %r, expected %r' % (offset, list_end))
-            elif name == b'seq ':
+            elif name == self.SEQ_CHUNK:
                 order = [i for i, in self.UNSIGNED.iter_unpack(self.blob[offset:offset + size])]
                 if len(order) != step_count:
                     raise ValueError('Wrong animation sequence size: %r, expected %r' % (len(order), step_count))
                 offset += size
-            elif name == b'rate':
+            elif name == self.RATE_CHUNK:
                 delays = [i for i, in self.UNSIGNED.iter_unpack(self.blob[offset:offset + size])]
                 if len(delays) != step_count:
                     raise ValueError('Wrong animation rate size: %r, expected %r' % (len(delays), step_count))

win2xcur/parser/cur.py

@@ -9,6 +9,7 @@ from win2xcur.parser.base import BaseParser
 
 class CURParser(BaseParser):
     MAGIC = b'\0\0\02\0'
+    ICO_TYPE_CUR = 2
     ICON_DIR = struct.Struct('<HHH')
     ICON_DIR_ENTRY = struct.Struct('<BBBBHHII')
 
@@ -27,7 +28,7 @@ class CURParser(BaseParser):
     def _parse_header(self) -> List[Tuple[int, int]]:
         reserved, ico_type, image_count = self.ICON_DIR.unpack(self.blob[:self.ICON_DIR.size])
         assert reserved == 0
-        assert ico_type == 2
+        assert ico_type == self.ICO_TYPE_CUR
         assert image_count == len(self._image.sequence)
 
         offset = self.ICON_DIR.size

win2xcur/parser/xcursor.py

@@ -0,0 +1,98 @@
+import struct
+from collections import defaultdict
+from typing import Any, Dict, List, Tuple, cast
+
+from win2xcur.cursor import CursorFrame, CursorImage
+from win2xcur.parser.base import BaseParser
+from win2xcur.utils import image_from_pixels
+
+
+class XCursorParser(BaseParser):
+    MAGIC = b'Xcur'
+    VERSION = 0x1_0000
+    FILE_HEADER = struct.Struct('<4sIII')
+    TOC_CHUNK = struct.Struct('<III')
+    CHUNK_IMAGE = 0xFFFD0002
+    IMAGE_HEADER = struct.Struct('<IIIIIIIII')
+
+    @classmethod
+    def can_parse(cls, blob: bytes) -> bool:
+        return blob[:len(cls.MAGIC)] == cls.MAGIC
+
+    def __init__(self, blob: bytes) -> None:
+        super().__init__(blob)
+        self.frames = self._parse()
+
+    def _unpack(self, struct_cls: struct.Struct, offset: int) -> Tuple[Any, ...]:
+        return struct_cls.unpack(self.blob[offset:offset + struct_cls.size])
+
+    def _parse(self) -> List[CursorFrame]:
+        magic, header_size, version, toc_size = self._unpack(self.FILE_HEADER, 0)
+        assert magic == self.MAGIC
+
+        if version != self.VERSION:
+            raise ValueError('Unsupported Xcursor version %r' % version)
+
+        offset = self.FILE_HEADER.size
+        chunks: List[Tuple[int, int, int]] = []
+        for i in range(toc_size):
+            chunk_type, chunk_subtype, position = self._unpack(self.TOC_CHUNK, offset)
+            chunks.append((chunk_type, chunk_subtype, position))
+            offset += self.TOC_CHUNK.size
+
+        images_by_size: Dict[int, List[Tuple[CursorImage, int]]] = defaultdict(list)
+
+        for chunk_type, chunk_subtype, position in chunks:
+            if chunk_type != self.CHUNK_IMAGE:
+                continue
+
+            size, actual_type, nominal_size, version, width, height, x_offset, y_offset, delay = \
+                self._unpack(self.IMAGE_HEADER, position)
+
+            if size != self.IMAGE_HEADER.size:
+                raise ValueError('Unexpected size: %r, expected %r' % (size, self.IMAGE_HEADER.size))
+
+            if actual_type != chunk_type:
+                raise ValueError('Unexpected chunk type: %r, expected %r' % (actual_type, chunk_type))
+
+            if nominal_size != chunk_subtype:
+                raise ValueError('Unexpected nominal size: %r, expected %r' % (nominal_size, chunk_subtype))
+
+            if width > 0x7FFF:
+                raise ValueError('Image width too large: %r' % width)
+
+            if height > 0x7FFF:
+                raise ValueError('Image height too large: %r' % height)
+
+            if x_offset > width:
+                raise ValueError('Hotspot x-coordinate too large: %r' % x_offset)
+
+            if y_offset > height:
+                raise ValueError('Hotspot x-coordinate too large: %r' % y_offset)
+
+            image_start = position + self.IMAGE_HEADER.size
+            image_size = width * height * 4
+            blob = self.blob[image_start:image_start + image_size]
+            if len(blob) != image_size:
+                raise ValueError('Invalid image at %d: expected %d bytes, got %d bytes' %
+                                 (image_size, image_size, len(blob)))
+
+            images_by_size[nominal_size].append(
+                (CursorImage(image_from_pixels(blob, width, height, 'ARGB', 'char'), (x_offset, y_offset)), delay)
+            )
+
+        if len(set(map(len, images_by_size.values()))) != 1:
+            raise ValueError('win2xcur does not support animations where each size has different number of frames')
+
+        result = []
+        for sequence in cast(Any, zip(*images_by_size.values())):
+            images: Tuple[CursorImage, ...]
+            delays: Tuple[int, ...]
+            images, delays = cast(Any, zip(*sequence))
+
+            if len(set(delays)) != 1:
+                raise ValueError('win2xcur does not support animations where each size has a different frame delay')
+
+            result.append(CursorFrame(list(images), delays[0]))
+
+        return result

win2xcur/utils.py

@@ -0,0 +1,20 @@
+import ctypes
+
+from wand.api import library
+from wand.image import Image
+from wand.sequence import SingleImage
+
+MagickImportImagePixels = library['MagickImportImagePixels']
+MagickImportImagePixels.argtypes = (
+    ctypes.c_void_p, ctypes.c_ssize_t, ctypes.c_ssize_t, ctypes.c_size_t,
+    ctypes.c_size_t, ctypes.c_char_p, ctypes.c_int, ctypes.c_void_p
+)
+StorageType = ('undefined', 'char', 'double', 'float',
+               'integer', 'long', 'quantum', 'short')
+
+
+def image_from_pixels(blob: bytes, width: int, height: int, pixel_format: str, pixel_size: str) -> SingleImage:
+    image = Image(width=width, height=height)
+    MagickImportImagePixels(image.wand, 0, 0, width, height, pixel_format.encode('ascii'),
+                            StorageType.index(pixel_size), blob)
+    return image.sequence[0]

win2xcur/writer/__init__.py

@@ -1,5 +1,10 @@
+from win2xcur.writer.windows import to_ani, to_cur, to_smart
 from win2xcur.writer.x11 import to_x11
 
+__all__ = ['to_ani', 'to_cur', 'to_smart', 'to_x11']
+
 CONVERTERS = {
     'x11': (to_x11, ''),
+    'ani': (to_ani, '.ani'),
+    'cur': (to_cur, '.cur'),
 }

win2xcur/writer/windows.py

@@ -0,0 +1,71 @@
+from io import BytesIO
+from itertools import chain
+from typing import List, Tuple
+
+from win2xcur.cursor import CursorFrame
+from win2xcur.parser import ANIParser, CURParser
+
+
+def to_cur(frame: CursorFrame) -> bytes:
+    header = CURParser.ICON_DIR.pack(0, CURParser.ICO_TYPE_CUR, len(frame))
+    directory: List[bytes] = []
+    image_data: List[bytes] = []
+    offset = CURParser.ICON_DIR.size + len(frame) * CURParser.ICON_DIR_ENTRY.size
+
+    for image in frame:
+        clone = image.image.clone()
+        if clone.width > 256 or clone.height > 256:
+            raise ValueError(f'Image too big for CUR format: {clone.width}x{clone.height}')
+        blob = clone.make_blob('png')
+        image_data.append(blob)
+        x_offset, y_offset = image.hotspot
+        directory.append(CURParser.ICON_DIR_ENTRY.pack(
+            clone.height & 0xFF, clone.height & 0xFF, 0, 0, x_offset, y_offset, len(blob), offset
+        ))
+        offset += len(blob)
+
+    return b''.join(chain([header], directory, image_data))
+
+
+def get_ani_cur_list(frames: List[CursorFrame]) -> bytes:
+    io = BytesIO()
+    for frame in frames:
+        cur_file = to_cur(frame)
+        io.write(ANIParser.CHUNK_HEADER.pack(ANIParser.ICON_CHUNK, len(cur_file)))
+        io.write(cur_file)
+        if len(cur_file) & 1:
+            io.write(b'\0')
+    return io.getvalue()
+
+
+def get_ani_rate_chunk(frames: List[CursorFrame]) -> bytes:
+    io = BytesIO()
+    io.write(ANIParser.CHUNK_HEADER.pack(ANIParser.RATE_CHUNK, ANIParser.UNSIGNED.size * len(frames)))
+    for frame in frames:
+        io.write(ANIParser.UNSIGNED.pack(int(round(frame.delay * 60))))
+    return io.getvalue()
+
+
+def to_ani(frames: List[CursorFrame]) -> bytes:
+    ani_header = ANIParser.ANIH_HEADER.pack(
+        ANIParser.ANIH_HEADER.size, len(frames), len(frames), 0, 0, 32, 1, 1, ANIParser.ICON_FLAG
+    )
+
+    cur_list = get_ani_cur_list(frames)
+    chunks = [
+        ANIParser.CHUNK_HEADER.pack(ANIParser.HEADER_CHUNK, len(ani_header)),
+        ani_header,
+        ANIParser.RIFF_HEADER.pack(ANIParser.LIST_CHUNK, len(cur_list) + 4, ANIParser.FRAME_TYPE),
+        cur_list,
+        get_ani_rate_chunk(frames),
+    ]
+    body = b''.join(chunks)
+    riff_header: bytes = ANIParser.RIFF_HEADER.pack(ANIParser.SIGNATURE, len(body) + 4, ANIParser.ANI_TYPE)
+    return riff_header + body
+
+
+def to_smart(frames: List[CursorFrame]) -> Tuple[str, bytes]:
+    if len(frames) == 1:
+        return '.cur', to_cur(frames[0])
+    else:
+        return '.ani', to_ani(frames)