feat: refactor assemblicon (2/3) context job topology

tool/src/aurora_iconset_builder/assemblicon/highlevel/address.py

@@ -35,3 +35,6 @@ class Address:
             self.__domain == other.__domain
             and self.__subdirectories == other.__subdirectories
         )
+
+    def __repr__(self) -> str:
+        return f"Address({self.__domain.name}, {'/'.join(self.__subdirectories)})"

tool/src/aurora_iconset_builder/assemblicon/highlevel/context.py

@@ -1,27 +1,37 @@
 from collections import deque
 from dataclasses import dataclass
-from typing import Callable, Iterable, TypeVar, Generic
+from typing import Iterable, TypeVar, Generic, Protocol, Callable, cast
 from .environment import Environment, Prologue
+from .address import Address
+from .source import Source
+from .sink import Sink
 from ..logger import HIGHLEVEL_LOGGER as LOGGER
 
 
 P = TypeVar("P", bound=Prologue)
 
 
+class Executor[*Ts](Protocol):
+    def __call__(self, *args: *Ts) -> None: ...
+
+
 @dataclass(frozen=True)
-class Job(Generic[P]):
+class Job:
-    """The class represents a job."""
+    """
+    The class represents a type-erased job.
+    Precise typing lives only in register's signature.
+    """
 
     name: str
     """The name of this job."""
-    dependencies: set[str]
+    ios: tuple[Source | Sink, ...]
-    """The dependencies of this job. Each item is a job name. Empty for no dependencies."""
+    """The sources and sinks of this job."""
-    executor: Callable[[Environment[P]], None]
+    executor: Callable[..., None]
     """The executor of this job."""
 
 
 class Context(Generic[P]):
-    __jobs: dict[str, Job[P]]
+    __jobs: dict[str, Job]
     """The dict storing all jobs. Key is job name and value is job payload."""
     __env: Environment[P]
     """The shared environment passed to every job executor."""
@@ -39,26 +49,76 @@ class Context(Generic[P]):
         executed sequentially with the shared environment. The chain is
         aborted on the first failing executor.
 
+        Job dependencies are not declared explicitly. They are inferred
+        from the job IO declarations instead: a job depends on the
+        producer of every address required by one of its sources.
+
         :param selected: Only build jobs matching these given names.
             Their dependencies are always built first. None for building
             all jobs.
+        :raises ValueError: If a selected job name is unknown, some
+            address is produced more than once (by one or several
+            jobs), some required address has no producer at all, or the
+            inferred dependency graph contains a cycle (including
+            self-dependencies).
         """
-        # Step 1: resolve the selection closure. Collect the selected jobs
+        # Step 0: build the producer table over all registered jobs,
-        # themselves plus all their transitive dependencies into "needed".
+        # mapping every produced address to its producing job name.
-        needed: set[str] = set()
+        # An address produced more than once is rejected right here,
-        if selected is None:
+        # even if it is never actually required: the producers may be
-            needed = set(self.__jobs.keys())
+        # several jobs, or a single job declaring several identical
-        else:
+        # sinks.
-            pending = list(selected)
+        producers: dict[Address, str] = {}
-            while pending:
+        for name, job in self.__jobs.items():
-                name = pending.pop()
+            for item in job.ios:
-                if name in needed:
+                if not isinstance(item, Sink):
                     continue
-                job = self.__jobs.get(name)
+                addr = item.address
-                if job is None:
+                prev = producers.get(addr)
-                    raise ValueError(f'Unknown job name: "{name}"')
+                if prev is not None and prev == name:
-                needed.add(name)
+                    raise ValueError(
-                pending.extend(job.dependencies)
+                        f'Job "{name}" declares multiple sinks '
+                        f"producing the same address {addr!r}."
+                    )
+                if prev is not None:
+                    raise ValueError(
+                        f"Address {addr!r} is produced by both job "
+                        f'"{prev}" and "{name}".'
+                    )
+                producers[addr] = name
+
+        # Step 1: resolve the selection closure. Collect the selected
+        # jobs themselves plus, transitively, every job producing an
+        # address one of their sources requires. A source whose address
+        # is ``None`` depends on nothing.
+        needed: set[str] = set()
+        pending = list(self.__jobs.keys()) if selected is None else list(selected)
+        while pending:
+            name = pending.pop()
+            if name in needed:
+                continue
+            job = self.__jobs.get(name)
+            if job is None:
+                raise ValueError(f'Unknown job name: "{name}"')
+            needed.add(name)
+            for item in job.ios:
+                if not isinstance(item, Source):
+                    continue
+                addr = item.address
+                if addr is None:
+                    continue
+                producer = producers.get(addr)
+                if producer is None:
+                    raise ValueError(
+                        f'Job "{name}" requires resource {addr!r} '
+                        f"but no registered job produces it."
+                    )
+                if producer == name:
+                    raise ValueError(
+                        f'Job "{name}" requires resource {addr!r} '
+                        f"which is produced by itself."
+                    )
+                pending.append(producer)
 
         # Step 2: Kahn topological sort over the sub-graph induced by "needed".
         # "in_degree": key is a job name, value is how many of its (needed)
@@ -68,7 +128,17 @@ class Context(Generic[P]):
         in_degree: dict[str, int] = {name: 0 for name in needed}
         dependents: dict[str, list[str]] = {name: [] for name in needed}
         for name in needed:
-            for dep in self.__jobs[name].dependencies:
+            # "deps": the names of jobs producing resources this job
+            # requires, deduplicated
+            deps: set[str] = set()
+            for item in self.__jobs[name].ios:
+                if not isinstance(item, Source):
+                    continue
+                addr = item.address
+                if addr is None:
+                    continue
+                deps.add(producers[addr])
+            for dep in deps:
                 dependents[dep].append(name)
                 in_degree[name] += 1
         queue = deque(
@@ -86,29 +156,35 @@ class Context(Generic[P]):
             cyclic = sorted(needed - set(order))
             raise ValueError(f"Dependency cycle detected among jobs: {cyclic}")
 
-        # Step 3: execute the build chain in resolved order.
+        # Step 3: execute the build chain in resolved order, feeding
+        # the shared environment and the declared IO items to each
+        # executor.
         for name in order:
             LOGGER.info('Running job "%s"...', name)
-            self.__jobs[name].executor(self.__env)
+            self.__jobs[name].executor(self.__env, *self.__jobs[name].ios)
 
-    def register(
+    def register[*Ts](
         self,
         name: str,
-        deps: Iterable[str],
+        ios: tuple[*Ts],
-        exec: Callable[[Environment[P]], None],
+        executor: Executor[Environment[P], *Ts],
     ) -> None:
         """
         Register a new job with given name, dependencies and executor.
 
         :param name: The name of job.
-        :param deps: An iterable whose each item is the name of dependencies of this job.
+        :param ios: A tuple declares ths sources and sinks of this job.
+            These sources and sinks will be directly passed to executor.
         :param exec: The executor of this job.
         """
+        # check job name collision
         if name in self.__jobs:
             raise ValueError(f'Can not register 2 jobs with same name: "{name}"')
+        # check the type of sources and sinks
+        for item in ios:
+            if not isinstance(item, (Source, Sink)):
+                raise TypeError(
+                    "passes source or sink instance must be the subclass of Source or Sink"
+                )
 
-        job_deps = set(deps)
+        self.__jobs[name] = Job(name, cast(tuple[Source | Sink, ...], ios), executor)
-        if name in job_deps:
-            raise ValueError(f'Self-dependency is not allowed in job: "{name}"')
-
-        self.__jobs[name] = Job(name, job_deps, exec)

tool/src/aurora_iconset_builder/assemblicon/highlevel/environment.py

@@ -7,7 +7,7 @@ from pathlib import Path
 from typing import TypeVar, Generic, Callable
 from ..lowlevel.environment import Prologue as LowPrologue, Environment as LowEnv
 from ..lowlevel.artrdr import SvgRenderKind
-from .platform import FdContext, WinCategory, MacCategory, FD_THUMBNAIL_HWS
+from .utils import FdContext, WinCategory, MacCategory, FD_THUMBNAIL_HWS
 
 
 @dataclass(frozen=True)

tool/src/aurora_iconset_builder/assemblicon/highlevel/platform.py

@@ -1,35 +0,0 @@
-import enum
-
-
-class FdContext(enum.StrEnum):
-    Actions = "actions"
-    Animations = "animations"
-    Applications = "apps"
-    Categories = "categories"
-    Devices = "devices"
-    Emblems = "emblems"
-    Emotes = "emotes"
-    International = "intl"
-    MimeTypes = "mimetypes"
-    Places = "places"
-    Status = "status"
-
-
-class WinCategory(enum.StrEnum):
-    Application = "apps"
-    """Icons for applications"""
-    Extension = "exts"
-    """Icons for file extensions"""
-
-
-class MacCategory(enum.StrEnum):
-    Application = "apps"
-    """Icons for applications"""
-    Extension = "exts"
-    """Icons for file extensions"""
-
-
-# TODO: For easy testing, we only build 32x32 and 48x48.
-# Enable full generation after testing.
-FD_THUMBNAIL_HWS: tuple[int, ...] = (32, 48)
-# FD_THUMBNAIL_HWS: tuple[int, ...] = (16, 32, 48, 64, 128, 256)

tool/src/aurora_iconset_builder/assemblicon/highlevel/sink/__init__.py

@@ -1,7 +1,13 @@
 from .common import Sink
 from .tempart_sink import TempArtSink
+from .fdart_sink import FdArtSink
+from .winart_sink import WinArtSink
+from .macart_sink import MacArtSink
 
 __all__ = [
     "Sink",
-    "TempArtSink"
+    "TempArtSink",
+    "FdArtSink",
+    "WinArtSink",
+    "MacArtSink"
 ]

tool/src/aurora_iconset_builder/assemblicon/highlevel/sink/fdart_sink.py

@@ -2,8 +2,7 @@ from dataclasses import dataclass
 from .common import Sink
 from ..address import Address, AddressDomain
 from ..environment import Environment
-from ..platform import FdContext, FD_THUMBNAIL_HWS
+from ..utils import Artifact, FdContext, FD_THUMBNAIL_HWS
-from ..utils import Artifact
 from ...lowlevel import artio, artrdr
 from ...logger import HIGHLEVEL_LOGGER as LOGGER
 

tool/src/aurora_iconset_builder/assemblicon/highlevel/sink/macart_sink.py

@@ -2,8 +2,7 @@ from dataclasses import dataclass
 from .common import Sink
 from ..address import Address, AddressDomain
 from ..environment import Environment
-from ..platform import MacCategory
+from ..utils import Artifact, MacCategory
-from ..utils import Artifact
 from ...lowlevel import artrdr
 from ...logger import HIGHLEVEL_LOGGER as LOGGER
 

tool/src/aurora_iconset_builder/assemblicon/highlevel/sink/winart_sink.py

@@ -2,8 +2,7 @@ from dataclasses import dataclass
 from .common import Sink
 from ..address import Address, AddressDomain
 from ..environment import Environment
-from ..platform import WinCategory
+from ..utils import Artifact, WinCategory
-from ..utils import Artifact
 from ...lowlevel import artrdr
 from ...logger import HIGHLEVEL_LOGGER as LOGGER
 

tool/src/aurora_iconset_builder/assemblicon/highlevel/source/__init__.py

@@ -1,7 +1,22 @@
 from .common import Source
+from .newart_source import NewArtSource
+from .bitmap_source import BitmapSource
+from .svg_source import SvgSource
+from .blender_source import BlenderSource
 from .tempart_source import TempArtSource
+from .fdart_source import FdArtSource
+from .winart_source import WinArtSource
+from .macart_source import MacArtSource
+
 
 __all__ = [
     "Source",
-    "TempArtSource"
+    "NewArtSource",
+    "BitmapSource",
+    "SvgSource",
+    "BlenderSource",
+    "TempArtSource",
+    "FdArtSource",
+    "WinArtSource",
+    "MacArtSource",
 ]

tool/src/aurora_iconset_builder/assemblicon/highlevel/source/bitmap_source.py

@@ -6,7 +6,7 @@ from ...lowlevel import artio
 from ...logger import HIGHLEVEL_LOGGER as LOGGER
 
 
-class InArtSource(Source):
+class BitmapSource(Source):
     """The source to bimap artifact (PNG, JPEG or BMP) located in input directory."""
 
     __comps: tuple[str, ...]

tool/src/aurora_iconset_builder/assemblicon/highlevel/source/fdart_source.py

@@ -2,12 +2,12 @@ from .common import Source
 from ..address import Address, AddressDomain
 from ..environment import Environment
 from ..sink.fdart_sink import FdArtSink
-from ..platform import FdContext, FD_THUMBNAIL_HWS
+from ..utils import FdContext, FD_THUMBNAIL_HWS
 from ...lowlevel import artio
 from ...logger import HIGHLEVEL_LOGGER as LOGGER
 
 
-class FdOutArtSource(Source):
+class FdArtSource(Source):
     __kind: FdContext
     __name: str
 

tool/src/aurora_iconset_builder/assemblicon/highlevel/source/macart_source.py

@@ -2,12 +2,12 @@ from .common import Source
 from ..address import Address, AddressDomain
 from ..environment import Environment
 from ..sink.macart_sink import MacArtSink
-from ..platform import MacCategory
+from ..utils import MacCategory
 from ...lowlevel import artio
 from ...logger import HIGHLEVEL_LOGGER as LOGGER
 
 
-class MacOutArtSource(Source):
+class MacArtSource(Source):
     __kind: MacCategory
     __name: str
 

tool/src/aurora_iconset_builder/assemblicon/highlevel/source/winart_source.py

@@ -2,12 +2,12 @@ from .common import Source
 from ..address import Address, AddressDomain
 from ..environment import Environment
 from ..sink.winart_sink import WinArtSink
-from ..platform import WinCategory
+from ..utils import WinCategory
 from ...lowlevel import artio
 from ...logger import HIGHLEVEL_LOGGER as LOGGER
 
 
-class WinOutArtSource(Source):
+class WinArtSource(Source):
     __kind: WinCategory
     __name: str
 

tool/src/aurora_iconset_builder/assemblicon/highlevel/utils.py

@@ -1,3 +1,4 @@
+import enum
 from ..lowlevel.utils import Artifact
 
 VART_HW: int = 1024
@@ -9,3 +10,37 @@ def check_vart(art: Artifact) -> None:
         raise ValueError(
             f"The size of general highlevel used artifact must be {VART_HW}x{VART_HW}, got {art.width}x{art.height} instead"
         )
+
+
+class FdContext(enum.StrEnum):
+    Actions = "actions"
+    Animations = "animations"
+    Applications = "apps"
+    Categories = "categories"
+    Devices = "devices"
+    Emblems = "emblems"
+    Emotes = "emotes"
+    International = "intl"
+    MimeTypes = "mimetypes"
+    Places = "places"
+    Status = "status"
+
+
+class WinCategory(enum.StrEnum):
+    Application = "apps"
+    """Icons for applications"""
+    Extension = "exts"
+    """Icons for file extensions"""
+
+
+class MacCategory(enum.StrEnum):
+    Application = "apps"
+    """Icons for applications"""
+    Extension = "exts"
+    """Icons for file extensions"""
+
+
+# TODO: For easy testing, we only build 32x32 and 48x48.
+# Enable full generation after testing.
+FD_THUMBNAIL_HWS: tuple[int, ...] = (32, 48)
+# FD_THUMBNAIL_HWS: tuple[int, ...] = (16, 32, 48, 64, 128, 256)