feat: update assemblicon all in one
after doing some exploration for assemblicon architecture in aurora-iconset project, we decide peel it off from aurora-iconset as an independent project. the git log history is stored in aurora-iconset project and I don't want to migrate them. the history of these commited code can be tracted by commit 93b48f22647265cc9efd87f52850f0511319672c..72edad6edca32333b5bd3b1668cd2837d770e4aa in SarasasChipWorkshop/aurora-iconset.
This commit is contained in:
@@ -1,2 +0,0 @@
|
|||||||
def hello() -> str:
|
|
||||||
return "Hello from assemblicon!"
|
|
||||||
|
|||||||
+98
-35
@@ -1,45 +1,108 @@
|
|||||||
import argparse
|
import enum
|
||||||
from typing import Optional, Callable
|
from typing import Optional
|
||||||
from dataclasses import dataclass
|
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
from argparse import ArgumentParser, Namespace
|
||||||
|
from dataclasses import dataclass
|
||||||
|
from .lowlevel.artrdr import SvgRenderKind
|
||||||
|
|
||||||
|
|
||||||
|
class CliSvgRenderKind(enum.StrEnum):
|
||||||
|
Inkscape = "inkscape"
|
||||||
|
ReSvg = "resvg"
|
||||||
|
|
||||||
|
def to_assemblicon_type(self) -> SvgRenderKind:
|
||||||
|
match self:
|
||||||
|
case CliSvgRenderKind.Inkscape:
|
||||||
|
return SvgRenderKind.Inkscape
|
||||||
|
case CliSvgRenderKind.ReSvg:
|
||||||
|
return SvgRenderKind.ReSvg
|
||||||
|
|
||||||
|
|
||||||
@dataclass(frozen=True)
|
@dataclass(frozen=True)
|
||||||
class Cli:
|
class StdOpts:
|
||||||
input_dir: Path
|
"""Assemblicon standard command line options"""
|
||||||
output_dir: Path
|
|
||||||
|
|
||||||
def parse() -> Cli:
|
input_directory: Optional[Path]
|
||||||
parser = argparse.ArgumentParser(
|
"""The path to input directory. None if you doesn't request it."""
|
||||||
prog="SarasaCW OMRF Packer",
|
output_directory: Optional[Path]
|
||||||
description=(
|
"""The path to output directory. None if you doesn't request it."""
|
||||||
"Packaging tool of Sarasas Chip Workshop Oh My Rust FFI (OMRF). "
|
selection: Optional[list[str]]
|
||||||
"Creates a CMake-style distributable layout and generates CMake "
|
"""The list holding all name of jobs for building."""
|
||||||
"and pkg-config files for Rust FFI build artifacts."
|
svg_render: CliSvgRenderKind
|
||||||
),
|
"""The render kind for SVG file."""
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass(frozen=True)
|
||||||
|
class StdOptsSetup:
|
||||||
|
require_input: bool
|
||||||
|
"""Whether add input path option in argument parser."""
|
||||||
|
require_output: bool
|
||||||
|
"""Whether add outputput path option in argument parser."""
|
||||||
|
|
||||||
|
|
||||||
|
def create_argparser(setup: StdOptsSetup) -> ArgumentParser:
|
||||||
|
"""
|
||||||
|
Create Assemblicon standard argument parser.
|
||||||
|
|
||||||
|
This argument parser are filled properly with the requirement of Assemblicon itself.
|
||||||
|
It can be directly used or modified with extra options.
|
||||||
|
|
||||||
|
:param cfg: The setup of argument parser.
|
||||||
|
:return: The built argument parser.
|
||||||
|
"""
|
||||||
|
parser = ArgumentParser(
|
||||||
|
prog="Assemblicon",
|
||||||
|
description="The Programmatic Icon Assembly Workflow.",
|
||||||
|
)
|
||||||
|
if setup.require_input:
|
||||||
|
parser.add_argument(
|
||||||
|
"-i",
|
||||||
|
"--input",
|
||||||
|
dest="input",
|
||||||
|
action="store",
|
||||||
|
type=Path,
|
||||||
|
required=True,
|
||||||
|
help="The path to input directory.",
|
||||||
|
metavar="DIR",
|
||||||
|
)
|
||||||
|
if setup.require_output:
|
||||||
|
parser.add_argument(
|
||||||
|
"-o",
|
||||||
|
"--output",
|
||||||
|
dest="output",
|
||||||
|
action="store",
|
||||||
|
type=Path,
|
||||||
|
required=True,
|
||||||
|
help="The path to output directory.",
|
||||||
|
metavar="DIR",
|
||||||
|
)
|
||||||
|
parser.add_argument(
|
||||||
|
"-t",
|
||||||
|
"--select",
|
||||||
|
dest="selection",
|
||||||
|
action="extend",
|
||||||
|
nargs="+",
|
||||||
|
type=str,
|
||||||
|
help="Only build given name represented jobs.",
|
||||||
|
metavar="RENDER",
|
||||||
)
|
)
|
||||||
parser.add_argument(
|
parser.add_argument(
|
||||||
"-i",
|
"-s",
|
||||||
"--input",
|
"--svg-render",
|
||||||
dest="input_dir",
|
dest="svg_render",
|
||||||
action="store",
|
action="store",
|
||||||
type=Path,
|
type=CliSvgRenderKind,
|
||||||
required=False,
|
default=CliSvgRenderKind.Inkscape,
|
||||||
help="Directory where distributable files are installed (creates a CMake layout: bin/, include/, lib/ ...)",
|
help="The render used for rendering SVG into PNG.",
|
||||||
metavar="DIR",
|
metavar="RENDER",
|
||||||
)
|
)
|
||||||
parser.add_argument(
|
return parser
|
||||||
"-o",
|
|
||||||
"--output",
|
|
||||||
dest="output_dir",
|
def parse_standard_opts(cfg: StdOptsSetup, ns: Namespace) -> StdOpts:
|
||||||
action="store",
|
return StdOpts(
|
||||||
type=Path,
|
input_directory=ns.input if cfg.require_input else None,
|
||||||
required=False,
|
output_directory=ns.output if cfg.require_output else None,
|
||||||
help="Path of the zip archive created from the dist-dir contents",
|
selection=ns.selection,
|
||||||
metavar="DIR",
|
svg_render=ns.svg_render,
|
||||||
)
|
|
||||||
args = parser.parse_args()
|
|
||||||
return Cli(
|
|
||||||
dist_dir=args.dist_dir,
|
|
||||||
dist_zip=args.dist_zip,
|
|
||||||
)
|
)
|
||||||
|
|||||||
@@ -1,38 +0,0 @@
|
|||||||
import tempfile
|
|
||||||
from pathlib import Path
|
|
||||||
from types import TracebackType
|
|
||||||
|
|
||||||
|
|
||||||
class Context:
|
|
||||||
|
|
||||||
__input_dir: Path
|
|
||||||
__output_dir: Path
|
|
||||||
__temp_dir: tempfile.TemporaryDirectory[str]
|
|
||||||
|
|
||||||
def __init__(self, input_dir: Path, output_dir: Path) -> None:
|
|
||||||
self.__input_dir = input_dir
|
|
||||||
self.__output_dir = output_dir
|
|
||||||
self.__temp_dir = tempfile.TemporaryDirectory()
|
|
||||||
|
|
||||||
def input_artifact(self, *args: str) -> Path:
|
|
||||||
return self.__input_dir / Path(*args)
|
|
||||||
|
|
||||||
def output_artifact(self, *args: str) -> Path:
|
|
||||||
return self.__output_dir / Path(*args)
|
|
||||||
|
|
||||||
def temporary_artifact(self, *args: str) -> Path:
|
|
||||||
return Path(self.__temp_dir.name) / "user" / Path(*args)
|
|
||||||
|
|
||||||
def _intern_temporary_artifact(self, *args: str) -> Path:
|
|
||||||
return Path(self.__temp_dir.name) / "assemblicon" / Path(*args)
|
|
||||||
|
|
||||||
def __enter__(self) -> "Context":
|
|
||||||
return self
|
|
||||||
|
|
||||||
def __exit__(
|
|
||||||
self,
|
|
||||||
exc_type: type[BaseException] | None,
|
|
||||||
exc_value: BaseException | None,
|
|
||||||
traceback: TracebackType | None,
|
|
||||||
) -> None:
|
|
||||||
self.__temp_dir.cleanup()
|
|
||||||
@@ -0,0 +1,123 @@
|
|||||||
|
import math
|
||||||
|
from dataclasses import dataclass
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class Point:
|
||||||
|
"""A point (or vector) on 2D plane with integer coordinates."""
|
||||||
|
|
||||||
|
x: int
|
||||||
|
y: int
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def from_tuple(t: tuple[int, int]) -> "Point":
|
||||||
|
return Point(t[0], t[1])
|
||||||
|
|
||||||
|
def __add__(self, other: "Point") -> "Point":
|
||||||
|
if not isinstance(other, Point):
|
||||||
|
return NotImplemented
|
||||||
|
return Point(self.x + other.x, self.y + other.y)
|
||||||
|
|
||||||
|
def __sub__(self, other: "Point") -> "Point":
|
||||||
|
if not isinstance(other, Point):
|
||||||
|
return NotImplemented
|
||||||
|
return Point(self.x - other.x, self.y - other.y)
|
||||||
|
|
||||||
|
def __neg__(self) -> "Point":
|
||||||
|
return Point(-self.x, -self.y)
|
||||||
|
|
||||||
|
def __mul__(self, other: int | float) -> "Point":
|
||||||
|
if not isinstance(other, (int, float)):
|
||||||
|
return NotImplemented
|
||||||
|
return Point(math.floor(self.x * other), math.floor(self.y * other))
|
||||||
|
|
||||||
|
def __rmul__(self, other: int | float) -> "Point":
|
||||||
|
return self.__mul__(other)
|
||||||
|
|
||||||
|
def __floordiv__(self, other: int | float) -> "Point":
|
||||||
|
if not isinstance(other, (int, float)):
|
||||||
|
return NotImplemented
|
||||||
|
return Point(math.floor(self.x / other), math.floor(self.y / other))
|
||||||
|
|
||||||
|
def to_tuple(self) -> tuple[int, int]:
|
||||||
|
"""
|
||||||
|
Create the tuple representing this point by ``(x, y)``
|
||||||
|
which can be used by Pillow.
|
||||||
|
"""
|
||||||
|
return (self.x, self.y)
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class Rectangle:
|
||||||
|
"""
|
||||||
|
A rectangle framed by two corner points.
|
||||||
|
|
||||||
|
pos1 is always assumed to be the top-left corner and pos2 the
|
||||||
|
bottom-right one. This relation is a pure convention and is never
|
||||||
|
validated: constructing a Rectangle with swapped or inverted corners
|
||||||
|
produces a meaningless-but-silent result.
|
||||||
|
"""
|
||||||
|
|
||||||
|
pos1: Point
|
||||||
|
pos2: Point
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def from_tuple(t: tuple[int, int, int, int]) -> "Rectangle":
|
||||||
|
return Rectangle(Point(t[0], t[1]), Point(t[2], t[3]))
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def from_min_max_rect(pos1: Point, pos2: Point) -> "Rectangle":
|
||||||
|
return Rectangle(pos1, pos2)
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def from_pos_size_rect(pos: Point, width: int, height: int) -> "Rectangle":
|
||||||
|
return Rectangle(pos, pos + Point(width, height))
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def from_margin_rect(
|
||||||
|
width: int, height: int, top: int, right: int, bottom: int, left: int
|
||||||
|
) -> "Rectangle":
|
||||||
|
"""
|
||||||
|
Create the rectangle with the size of this region's parent, and the margin between this region and its parent.
|
||||||
|
|
||||||
|
:param width: The width of parent.
|
||||||
|
:param height: The height of parent.
|
||||||
|
:param top: The distance from this rectangle to the top border of parent.
|
||||||
|
:param right: The distance from this rectangle to the right border of parent.
|
||||||
|
:param bottom: The distance from this rectangle to the bottom border of parent.
|
||||||
|
:param left: The distance from this rectangle to the top border of parent.
|
||||||
|
"""
|
||||||
|
return Rectangle(Point(left, top), Point(width - right, height - bottom))
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def from_padding_rect(
|
||||||
|
pos: Point, top: int, right: int, bottom: int, left: int
|
||||||
|
) -> "Rectangle":
|
||||||
|
"""
|
||||||
|
Create the rectangle with the central point in this region's parent, and the padding between this point and region's border.
|
||||||
|
|
||||||
|
:param x: The X factor of reference point.
|
||||||
|
:param y: The Y factor of reference point.
|
||||||
|
:param top: The distance from reference point to this rectangle's top border.
|
||||||
|
:param right: The distance from reference point to this rectangle's right border.
|
||||||
|
:param bottom: The distance from reference point to this rectangle's bottom border.
|
||||||
|
:param left: The distance from reference point to this rectangle's left border.
|
||||||
|
"""
|
||||||
|
return Rectangle(
|
||||||
|
Point(pos.x - left, pos.y - top), Point(pos.x + right, pos.y + bottom)
|
||||||
|
)
|
||||||
|
|
||||||
|
@property
|
||||||
|
def width(self) -> int:
|
||||||
|
return self.pos2.x - self.pos1.x
|
||||||
|
|
||||||
|
@property
|
||||||
|
def height(self) -> int:
|
||||||
|
return self.pos2.y - self.pos1.y
|
||||||
|
|
||||||
|
def to_tuple(self) -> tuple[int, int, int, int]:
|
||||||
|
"""
|
||||||
|
Create the tuple representing this rectangle by ``(x1, y2 x2, y2)``
|
||||||
|
which can be used by Pillow.
|
||||||
|
"""
|
||||||
|
return (self.pos1.x, self.pos1.y, self.pos2.x, self.pos2.y)
|
||||||
@@ -0,0 +1,40 @@
|
|||||||
|
import enum
|
||||||
|
|
||||||
|
|
||||||
|
class AddressDomain(enum.IntEnum):
|
||||||
|
Input = enum.auto()
|
||||||
|
Temporary = enum.auto()
|
||||||
|
Output = enum.auto()
|
||||||
|
|
||||||
|
FdOutput = enum.auto()
|
||||||
|
WinOutput = enum.auto()
|
||||||
|
MacOutput = enum.auto()
|
||||||
|
|
||||||
|
|
||||||
|
class Address:
|
||||||
|
"""The unique identity of an artifact slot (source/sink)."""
|
||||||
|
|
||||||
|
__domain: AddressDomain
|
||||||
|
__subdirectories: tuple[str, ...]
|
||||||
|
|
||||||
|
def __init__(self, domain: AddressDomain, *args: str) -> None:
|
||||||
|
self.__domain = domain
|
||||||
|
self.__subdirectories = tuple(args)
|
||||||
|
|
||||||
|
def __hash__(self) -> int:
|
||||||
|
return hash((self.__domain.value,) + self.__subdirectories)
|
||||||
|
|
||||||
|
def __eq__(self, other: object) -> bool:
|
||||||
|
if self is other:
|
||||||
|
return True
|
||||||
|
|
||||||
|
if not isinstance(other, Address):
|
||||||
|
return NotImplemented
|
||||||
|
|
||||||
|
return (
|
||||||
|
self.__domain == other.__domain
|
||||||
|
and self.__subdirectories == other.__subdirectories
|
||||||
|
)
|
||||||
|
|
||||||
|
def __repr__(self) -> str:
|
||||||
|
return f"Address({self.__domain.name}, {'/'.join(self.__subdirectories)})"
|
||||||
@@ -0,0 +1,191 @@
|
|||||||
|
from collections import deque
|
||||||
|
from dataclasses import dataclass
|
||||||
|
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:
|
||||||
|
"""
|
||||||
|
The class represents a type-erased job.
|
||||||
|
Precise typing lives only in register's signature.
|
||||||
|
"""
|
||||||
|
|
||||||
|
name: str
|
||||||
|
"""The name of this job."""
|
||||||
|
ios: tuple[Source | Sink, ...]
|
||||||
|
"""The sources and sinks of this job."""
|
||||||
|
executor: Callable[..., None]
|
||||||
|
"""The executor of this job."""
|
||||||
|
|
||||||
|
|
||||||
|
class Context(Generic[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."""
|
||||||
|
|
||||||
|
def __init__(self, env: Environment[P]) -> None:
|
||||||
|
self.__jobs = {}
|
||||||
|
self.__env = env
|
||||||
|
|
||||||
|
def run(self, selected: Iterable[str] | None) -> None:
|
||||||
|
"""
|
||||||
|
Execute builder with given selections.
|
||||||
|
|
||||||
|
The selected jobs, together with all their transitive dependencies,
|
||||||
|
are resolved and topologically sorted (Kahn's algorithm), then
|
||||||
|
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 0: build the producer table over all registered jobs,
|
||||||
|
# mapping every produced address to its producing job name.
|
||||||
|
# An address produced more than once is rejected right here,
|
||||||
|
# even if it is never actually required: the producers may be
|
||||||
|
# several jobs, or a single job declaring several identical
|
||||||
|
# sinks.
|
||||||
|
producers: dict[Address, str] = {}
|
||||||
|
for name, job in self.__jobs.items():
|
||||||
|
for item in job.ios:
|
||||||
|
if not isinstance(item, Sink):
|
||||||
|
continue
|
||||||
|
addr = item.address
|
||||||
|
prev = producers.get(addr)
|
||||||
|
if prev is not None and prev == name:
|
||||||
|
raise ValueError(
|
||||||
|
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)
|
||||||
|
# dependencies are not built yet.
|
||||||
|
# "dependents": key is a job name, value is the list of jobs in
|
||||||
|
# "needed" that directly depend on the key job.
|
||||||
|
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:
|
||||||
|
# "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(
|
||||||
|
name for name in self.__jobs if name in needed and in_degree[name] == 0
|
||||||
|
)
|
||||||
|
order: list[str] = []
|
||||||
|
while queue:
|
||||||
|
name = queue.popleft()
|
||||||
|
order.append(name)
|
||||||
|
for dependent in dependents[name]:
|
||||||
|
in_degree[dependent] -= 1
|
||||||
|
if in_degree[dependent] == 0:
|
||||||
|
queue.append(dependent)
|
||||||
|
if len(order) != len(needed):
|
||||||
|
cyclic = sorted(needed - set(order))
|
||||||
|
raise ValueError(f"Dependency cycle detected among jobs: {cyclic}")
|
||||||
|
|
||||||
|
# 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].ios)
|
||||||
|
|
||||||
|
def register[*Ts](
|
||||||
|
self,
|
||||||
|
name: str,
|
||||||
|
ios: tuple[*Ts],
|
||||||
|
executor: Executor[Environment[P], *Ts],
|
||||||
|
) -> None:
|
||||||
|
"""
|
||||||
|
Register a new job with given name, dependencies and executor.
|
||||||
|
|
||||||
|
:param name: The name of 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(
|
||||||
|
f"All items in the ios of job {name!r} must be instances "
|
||||||
|
f"of Source or Sink, but got {type(item)}."
|
||||||
|
)
|
||||||
|
|
||||||
|
self.__jobs[name] = Job(name, cast(tuple[Source | Sink, ...], ios), executor)
|
||||||
@@ -0,0 +1,193 @@
|
|||||||
|
import tempfile
|
||||||
|
import os
|
||||||
|
from typing import Optional
|
||||||
|
from types import TracebackType
|
||||||
|
from dataclasses import dataclass
|
||||||
|
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 .utils import FdContext, WinCategory, MacCategory, FD_THUMBNAIL_HWS
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass(frozen=True)
|
||||||
|
class Prologue(LowPrologue):
|
||||||
|
svg_render: SvgRenderKind
|
||||||
|
"""The render kind for SVG"""
|
||||||
|
temp_dir_initializer: Callable[[Path], None] | None
|
||||||
|
"""
|
||||||
|
The extra initializer for user temporary directory.
|
||||||
|
The only argument in function is the path to user temporary directory.
|
||||||
|
``None`` for no extra initializer.
|
||||||
|
"""
|
||||||
|
|
||||||
|
|
||||||
|
_TEMP_USER_DIR: str = "user"
|
||||||
|
_TEMP_LIB_DIR: str = "assemblicon"
|
||||||
|
_TEMP_LIB_ALLOC_DIR: str = "alloc"
|
||||||
|
_TEMP_LIB_STABLE_DIR: str = "stable"
|
||||||
|
|
||||||
|
_ART_WIN: str = "windows"
|
||||||
|
_ART_MAC: str = "macos"
|
||||||
|
|
||||||
|
P = TypeVar("P", bound=Prologue)
|
||||||
|
|
||||||
|
|
||||||
|
class Environment(Generic[P]):
|
||||||
|
__prologue: P
|
||||||
|
"""The prologue used for environment."""
|
||||||
|
__lowenv: LowEnv
|
||||||
|
"""The lowlevel environment managing temporary directory."""
|
||||||
|
|
||||||
|
def __init__(self, prologue: P) -> None:
|
||||||
|
self.__prologue = prologue
|
||||||
|
self.__lowenv = LowEnv(self.__prologue)
|
||||||
|
|
||||||
|
# construct basic layout of temporary directory
|
||||||
|
self.__temporary_artifact(_TEMP_USER_DIR).mkdir(parents=True, exist_ok=True)
|
||||||
|
self.__temporary_artifact(_TEMP_LIB_DIR, _TEMP_LIB_ALLOC_DIR).mkdir(
|
||||||
|
parents=True, exist_ok=True
|
||||||
|
)
|
||||||
|
self.__temporary_artifact(_TEMP_LIB_DIR, _TEMP_LIB_STABLE_DIR).mkdir(
|
||||||
|
parents=True, exist_ok=True
|
||||||
|
)
|
||||||
|
# and output directory
|
||||||
|
for hw in FD_THUMBNAIL_HWS:
|
||||||
|
for kind in FdContext:
|
||||||
|
self.__output_artifact(f"{hw}x{hw}", str(kind)).mkdir(
|
||||||
|
parents=True, exist_ok=True
|
||||||
|
)
|
||||||
|
for kind in WinCategory:
|
||||||
|
self.__output_artifact(_ART_WIN, str(kind)).mkdir(
|
||||||
|
parents=True, exist_ok=True
|
||||||
|
)
|
||||||
|
for kind in MacCategory:
|
||||||
|
self.__output_artifact(_ART_MAC, str(kind)).mkdir(
|
||||||
|
parents=True, exist_ok=True
|
||||||
|
)
|
||||||
|
|
||||||
|
# construct user custom temporary layout of possible
|
||||||
|
if self.__prologue.temp_dir_initializer is not None:
|
||||||
|
temp_dir = self.user_temporary_artifact()
|
||||||
|
self.__prologue.temp_dir_initializer(temp_dir)
|
||||||
|
|
||||||
|
# region: With Visitor Requirements
|
||||||
|
|
||||||
|
def __enter__(self) -> "Environment":
|
||||||
|
self.__lowenv.__enter__()
|
||||||
|
return self
|
||||||
|
|
||||||
|
def __exit__(
|
||||||
|
self,
|
||||||
|
exc_type: type[BaseException] | None,
|
||||||
|
exc_value: BaseException | None,
|
||||||
|
traceback: TracebackType | None,
|
||||||
|
) -> None:
|
||||||
|
self.__lowenv.__exit__(exc_type, exc_value, traceback)
|
||||||
|
|
||||||
|
# endregion
|
||||||
|
|
||||||
|
# region: Prologue Visitor
|
||||||
|
|
||||||
|
@property
|
||||||
|
def prologue(self) -> Prologue:
|
||||||
|
return self.__prologue
|
||||||
|
|
||||||
|
@property
|
||||||
|
def user_prologue(self) -> P:
|
||||||
|
return self.__prologue
|
||||||
|
|
||||||
|
# endregion
|
||||||
|
|
||||||
|
# region: Input Artifact
|
||||||
|
|
||||||
|
def input_artifact(self, *args: str) -> Path:
|
||||||
|
"""
|
||||||
|
Get the path to input artifact.
|
||||||
|
"""
|
||||||
|
return self.__lowenv.input_artifact(*args)
|
||||||
|
|
||||||
|
# endregion
|
||||||
|
|
||||||
|
# region: Temporary Artifact
|
||||||
|
|
||||||
|
def __temporary_artifact(self, *args: str) -> Path:
|
||||||
|
return self.__lowenv.temporary_artifact(*args)
|
||||||
|
|
||||||
|
def user_temporary_artifact(self, *args: str) -> Path:
|
||||||
|
return self.__temporary_artifact(_TEMP_USER_DIR, *args)
|
||||||
|
|
||||||
|
def __lib_temporary_artifact(self, *args: str) -> Path:
|
||||||
|
return self.__temporary_artifact(_TEMP_LIB_DIR, *args)
|
||||||
|
|
||||||
|
def __allocate_lib_temporary_artifact(self, suffix: Optional[str] = None) -> Path:
|
||||||
|
p = self.__lib_temporary_artifact(_TEMP_LIB_ALLOC_DIR)
|
||||||
|
(fd, pf) = tempfile.mkstemp(suffix=suffix, prefix=None, dir=str(p), text=False)
|
||||||
|
os.close(fd)
|
||||||
|
return Path(pf)
|
||||||
|
|
||||||
|
def __fetch_const_lib_temporary_artifact(self, name: str) -> Path:
|
||||||
|
return self.__lib_temporary_artifact(_TEMP_LIB_STABLE_DIR, name)
|
||||||
|
|
||||||
|
def allocate_lib_render_temporary_artifact(self) -> Path:
|
||||||
|
"""
|
||||||
|
Allocated a path pointing to temporary render result.
|
||||||
|
|
||||||
|
:return: The path to temporary PNG file.
|
||||||
|
Please note that this file is already presneted in file system,
|
||||||
|
so caller need to overwrite it.
|
||||||
|
"""
|
||||||
|
return self.__allocate_lib_temporary_artifact(".png")
|
||||||
|
|
||||||
|
def fetch_lib_blender_script_temporary_artifact(self) -> Path:
|
||||||
|
"""
|
||||||
|
Allocated a path pointing to temporary script instructing Blender render.
|
||||||
|
|
||||||
|
:return: The path to temporary Python file.
|
||||||
|
Please note that this file may be presneted in file system,
|
||||||
|
so caller may need to overwrite it.
|
||||||
|
All Blender render process share the same script file (this).
|
||||||
|
"""
|
||||||
|
# TODO: Strip this "share" behavior to enable multi-threads build ability.
|
||||||
|
return self.__fetch_const_lib_temporary_artifact("blender.py")
|
||||||
|
|
||||||
|
# endregion
|
||||||
|
|
||||||
|
# region: Output Artifact
|
||||||
|
|
||||||
|
def __output_artifact(self, *args: str) -> Path:
|
||||||
|
"""
|
||||||
|
Get the path to output artifact.
|
||||||
|
"""
|
||||||
|
return self.__lowenv.output_artifact(*args)
|
||||||
|
|
||||||
|
def fd_output_artifact(self, hw: int, kind: FdContext, name: str) -> Path:
|
||||||
|
"""
|
||||||
|
Get the path for saving FreeDesktop icon with given image size, FreeDesktop icon and name.
|
||||||
|
|
||||||
|
:param hw: The height or width of this FreeDesktop icon.
|
||||||
|
Required because FreeDesktop need to save icon with different size variant.
|
||||||
|
:param kind: The FreeDesktop kind of this icon.
|
||||||
|
:param name: The name of saved artifact. The file extension ".png" is not required.
|
||||||
|
"""
|
||||||
|
return self.__output_artifact(f"{hw}x{hw}", str(kind), f"{name}.png")
|
||||||
|
|
||||||
|
def win_output_artifact(self, kind: WinCategory, name: str) -> Path:
|
||||||
|
"""
|
||||||
|
Get the path for saving Windows-only icon.
|
||||||
|
|
||||||
|
:param kind: The kind of this icon.
|
||||||
|
:param name: The name of saved artifact. The file extension ".ico" is not required.
|
||||||
|
"""
|
||||||
|
return self.__output_artifact(_ART_WIN, str(kind), f"{name}.ico")
|
||||||
|
|
||||||
|
def mac_output_artifact(self, kind: MacCategory, name: str) -> Path:
|
||||||
|
"""
|
||||||
|
Get the path for saving macOS-only icon.
|
||||||
|
|
||||||
|
:param kind: The kind of this icon.
|
||||||
|
:param name: The name of saved artifact. The file extension ".icns" is not required.
|
||||||
|
"""
|
||||||
|
return self.__output_artifact(_ART_MAC, str(kind), f"{name}.icns")
|
||||||
|
|
||||||
|
# endregion
|
||||||
@@ -0,0 +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",
|
||||||
|
"FdArtSink",
|
||||||
|
"WinArtSink",
|
||||||
|
"MacArtSink"
|
||||||
|
]
|
||||||
@@ -0,0 +1,12 @@
|
|||||||
|
from abc import ABC, abstractmethod
|
||||||
|
from ..address import Address
|
||||||
|
|
||||||
|
|
||||||
|
class Sink(ABC):
|
||||||
|
"""
|
||||||
|
Abstract base class for all sinks.
|
||||||
|
"""
|
||||||
|
|
||||||
|
@property
|
||||||
|
@abstractmethod
|
||||||
|
def address(self) -> Address: ...
|
||||||
@@ -0,0 +1,42 @@
|
|||||||
|
from dataclasses import dataclass
|
||||||
|
from .common import Sink
|
||||||
|
from ..address import Address, AddressDomain
|
||||||
|
from ..environment import Environment
|
||||||
|
from ..utils import Artifact, FdContext, FD_THUMBNAIL_HWS, check_vart
|
||||||
|
from ...lowlevel import artio, artrdr
|
||||||
|
from ...logger import HIGHLEVEL_LOGGER as LOGGER
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass(frozen=True)
|
||||||
|
class FdArtSinkHint:
|
||||||
|
kind: FdContext
|
||||||
|
name: str
|
||||||
|
|
||||||
|
|
||||||
|
class FdArtSink(Sink):
|
||||||
|
__kind: FdContext
|
||||||
|
__name: str
|
||||||
|
|
||||||
|
def __init__(self, kind: FdContext, name: str) -> None:
|
||||||
|
self.__kind = kind
|
||||||
|
self.__name = name
|
||||||
|
|
||||||
|
@property
|
||||||
|
def address(self) -> Address:
|
||||||
|
return Address(AddressDomain.FdOutput, str(self.__kind), self.__name)
|
||||||
|
|
||||||
|
def push(self, env: Environment, art: Artifact) -> None:
|
||||||
|
LOGGER.info(
|
||||||
|
'Saving FreeDesktop artifact with kind "%s" and name "%s".',
|
||||||
|
self.__kind,
|
||||||
|
self.__name,
|
||||||
|
)
|
||||||
|
check_vart(art)
|
||||||
|
for hw, thumbnail in zip(
|
||||||
|
FD_THUMBNAIL_HWS, artrdr.render_thumbnail(art, FD_THUMBNAIL_HWS)
|
||||||
|
):
|
||||||
|
dst = env.fd_output_artifact(hw, self.__kind, self.__name)
|
||||||
|
artio.save_artifact(thumbnail, dst)
|
||||||
|
|
||||||
|
def push_hint(self) -> FdArtSinkHint:
|
||||||
|
return FdArtSinkHint(self.__kind, self.__name)
|
||||||
@@ -0,0 +1,38 @@
|
|||||||
|
from dataclasses import dataclass
|
||||||
|
from .common import Sink
|
||||||
|
from ..address import Address, AddressDomain
|
||||||
|
from ..environment import Environment
|
||||||
|
from ..utils import Artifact, MacCategory, check_vart
|
||||||
|
from ...lowlevel import artrdr
|
||||||
|
from ...logger import HIGHLEVEL_LOGGER as LOGGER
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass(frozen=True)
|
||||||
|
class MacArtSinkHint:
|
||||||
|
kind: MacCategory
|
||||||
|
name: str
|
||||||
|
|
||||||
|
|
||||||
|
class MacArtSink(Sink):
|
||||||
|
__kind: MacCategory
|
||||||
|
__name: str
|
||||||
|
|
||||||
|
def __init__(self, kind: MacCategory, name: str) -> None:
|
||||||
|
self.__kind = kind
|
||||||
|
self.__name = name
|
||||||
|
|
||||||
|
@property
|
||||||
|
def address(self) -> Address:
|
||||||
|
return Address(AddressDomain.MacOutput, str(self.__kind), self.__name)
|
||||||
|
|
||||||
|
def push(self, env: Environment, art: Artifact) -> None:
|
||||||
|
LOGGER.info(
|
||||||
|
'Saving macOS-only artifact with kind "%s" and name "%s".',
|
||||||
|
self.__kind,
|
||||||
|
self.__name,
|
||||||
|
)
|
||||||
|
check_vart(art)
|
||||||
|
artrdr.render_icns(art, env.mac_output_artifact(self.__kind, self.__name))
|
||||||
|
|
||||||
|
def push_hint(self) -> MacArtSinkHint:
|
||||||
|
return MacArtSinkHint(self.__kind, self.__name)
|
||||||
@@ -0,0 +1,43 @@
|
|||||||
|
from pathlib import Path
|
||||||
|
from .common import Sink
|
||||||
|
from ..address import Address, AddressDomain
|
||||||
|
from ..environment import Environment
|
||||||
|
from ..utils import Artifact
|
||||||
|
from ...lowlevel import artio
|
||||||
|
from ...logger import HIGHLEVEL_LOGGER as LOGGER
|
||||||
|
|
||||||
|
|
||||||
|
class TempArtSink(Sink):
|
||||||
|
"""
|
||||||
|
The sink to artifact (PNG-only) located in temporary directory.
|
||||||
|
|
||||||
|
All existence of sub-directory should be ensured on your own.
|
||||||
|
"""
|
||||||
|
|
||||||
|
__comps: tuple[str, ...]
|
||||||
|
|
||||||
|
def __init__(self, *args: str) -> None:
|
||||||
|
self.__comps = tuple(args)
|
||||||
|
|
||||||
|
@property
|
||||||
|
def address(self) -> Address:
|
||||||
|
return Address(AddressDomain.Temporary, *self.__comps)
|
||||||
|
|
||||||
|
def push(self, env: Environment, art: Artifact) -> None:
|
||||||
|
p = self.push_hint(env)
|
||||||
|
LOGGER.info("Saving temporary artifact: %s", p)
|
||||||
|
artio.save_artifact(art, p)
|
||||||
|
|
||||||
|
def repull(self, env: Environment) -> Artifact:
|
||||||
|
"""
|
||||||
|
Re-pull the artifact from this sink, after the artifact has
|
||||||
|
been pushed to it (usually via a direct rendering).
|
||||||
|
Only meaningful for artifacts produced by the job declaring
|
||||||
|
this sink.
|
||||||
|
"""
|
||||||
|
p = self.push_hint(env)
|
||||||
|
LOGGER.info("Re-pulling temporary artifact: %s", p)
|
||||||
|
return artio.load_artifact(p)
|
||||||
|
|
||||||
|
def push_hint(self, env: Environment) -> Path:
|
||||||
|
return env.user_temporary_artifact(*self.__comps)
|
||||||
@@ -0,0 +1,38 @@
|
|||||||
|
from dataclasses import dataclass
|
||||||
|
from .common import Sink
|
||||||
|
from ..address import Address, AddressDomain
|
||||||
|
from ..environment import Environment
|
||||||
|
from ..utils import Artifact, WinCategory, check_vart
|
||||||
|
from ...lowlevel import artrdr
|
||||||
|
from ...logger import HIGHLEVEL_LOGGER as LOGGER
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass(frozen=True)
|
||||||
|
class WinArtSinkHint:
|
||||||
|
kind: WinCategory
|
||||||
|
name: str
|
||||||
|
|
||||||
|
|
||||||
|
class WinArtSink(Sink):
|
||||||
|
__kind: WinCategory
|
||||||
|
__name: str
|
||||||
|
|
||||||
|
def __init__(self, kind: WinCategory, name: str) -> None:
|
||||||
|
self.__kind = kind
|
||||||
|
self.__name = name
|
||||||
|
|
||||||
|
@property
|
||||||
|
def address(self) -> Address:
|
||||||
|
return Address(AddressDomain.WinOutput, str(self.__kind), self.__name)
|
||||||
|
|
||||||
|
def push(self, env: Environment, art: Artifact) -> None:
|
||||||
|
LOGGER.info(
|
||||||
|
'Saving Windows-only artifact with kind "%s" and name "%s".',
|
||||||
|
self.__kind,
|
||||||
|
self.__name,
|
||||||
|
)
|
||||||
|
check_vart(art)
|
||||||
|
artrdr.render_ico(art, env.win_output_artifact(self.__kind, self.__name))
|
||||||
|
|
||||||
|
def push_hint(self) -> WinArtSinkHint:
|
||||||
|
return WinArtSinkHint(self.__kind, self.__name)
|
||||||
@@ -0,0 +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",
|
||||||
|
"NewArtSource",
|
||||||
|
"BitmapSource",
|
||||||
|
"SvgSource",
|
||||||
|
"BlenderSource",
|
||||||
|
"TempArtSource",
|
||||||
|
"FdArtSource",
|
||||||
|
"WinArtSource",
|
||||||
|
"MacArtSource",
|
||||||
|
]
|
||||||
@@ -0,0 +1,24 @@
|
|||||||
|
from .common import Source
|
||||||
|
from ..address import Address
|
||||||
|
from ..environment import Environment
|
||||||
|
from ..utils import Artifact
|
||||||
|
from ...lowlevel import artio
|
||||||
|
from ...logger import HIGHLEVEL_LOGGER as LOGGER
|
||||||
|
|
||||||
|
|
||||||
|
class BitmapSource(Source):
|
||||||
|
"""The source to bimap artifact (PNG, JPEG or BMP) located in input directory."""
|
||||||
|
|
||||||
|
__comps: tuple[str, ...]
|
||||||
|
|
||||||
|
def __init__(self, *args: str) -> None:
|
||||||
|
self.__comps = tuple(args)
|
||||||
|
|
||||||
|
@property
|
||||||
|
def address(self) -> Address | None:
|
||||||
|
return None
|
||||||
|
|
||||||
|
def pull(self, env: Environment) -> Artifact:
|
||||||
|
p = env.input_artifact(*self.__comps)
|
||||||
|
LOGGER.info("Loading input artifact: %s", p)
|
||||||
|
return artio.load_input_artifact(p)
|
||||||
@@ -0,0 +1,58 @@
|
|||||||
|
from pathlib import Path
|
||||||
|
from typing import Optional
|
||||||
|
from .common import Source
|
||||||
|
from ..address import Address
|
||||||
|
from ..environment import Environment
|
||||||
|
from ..sink.tempart_sink import TempArtSink
|
||||||
|
from ..utils import Artifact, VART_HW
|
||||||
|
from ...lowlevel import artio, artrdr
|
||||||
|
from ...lowlevel.artrdr import BlenderConfig
|
||||||
|
from ...logger import HIGHLEVEL_LOGGER as LOGGER
|
||||||
|
|
||||||
|
|
||||||
|
DEFAULT_BLD_CONFIG = artrdr.BlenderConfig(VART_HW, VART_HW)
|
||||||
|
|
||||||
|
|
||||||
|
class BlenderSource(Source):
|
||||||
|
"""The source to Blender composition located in input directory."""
|
||||||
|
|
||||||
|
__comps: tuple[str, ...]
|
||||||
|
|
||||||
|
def __init__(self, *args: str) -> None:
|
||||||
|
self.__comps = tuple(args)
|
||||||
|
|
||||||
|
@property
|
||||||
|
def address(self) -> Address | None:
|
||||||
|
return None
|
||||||
|
|
||||||
|
def pull(
|
||||||
|
self, env: Environment, cfg: BlenderConfig = DEFAULT_BLD_CONFIG
|
||||||
|
) -> Artifact:
|
||||||
|
dst = self.__render(env, None, cfg)
|
||||||
|
return artio.load_artifact(dst)
|
||||||
|
|
||||||
|
def pull_to_temp(
|
||||||
|
self,
|
||||||
|
env: Environment,
|
||||||
|
temp_sink: TempArtSink,
|
||||||
|
cfg: BlenderConfig = DEFAULT_BLD_CONFIG,
|
||||||
|
) -> None:
|
||||||
|
self.__render(env, temp_sink.push_hint(env), cfg)
|
||||||
|
|
||||||
|
def __render(
|
||||||
|
self, env: Environment, dst: Optional[Path], cfg: BlenderConfig
|
||||||
|
) -> Path:
|
||||||
|
"""
|
||||||
|
Internal used function for rendering Blender composition.
|
||||||
|
|
||||||
|
:return: The path to destination storing render result.
|
||||||
|
"""
|
||||||
|
src = env.input_artifact(*self.__comps)
|
||||||
|
temp_script = env.fetch_lib_blender_script_temporary_artifact()
|
||||||
|
if dst is None:
|
||||||
|
LOGGER.info("Rendering temporary Blender composition: %s", src)
|
||||||
|
dst = env.allocate_lib_render_temporary_artifact()
|
||||||
|
else:
|
||||||
|
LOGGER.info("Rendering Blender composition: %s -> %s", src, dst)
|
||||||
|
artrdr.render_blender(src, dst, temp_script, cfg)
|
||||||
|
return dst
|
||||||
@@ -0,0 +1,20 @@
|
|||||||
|
from abc import ABC, abstractmethod
|
||||||
|
from ..address import Address
|
||||||
|
|
||||||
|
|
||||||
|
class Source(ABC):
|
||||||
|
"""
|
||||||
|
Abstract base class for all sources.
|
||||||
|
"""
|
||||||
|
|
||||||
|
@property
|
||||||
|
@abstractmethod
|
||||||
|
def address(self) -> Address | None:
|
||||||
|
"""
|
||||||
|
Get the address representation of this source.
|
||||||
|
|
||||||
|
:return: The address representing this source.
|
||||||
|
``None`` if there is no producer for this source.
|
||||||
|
It means that this source do not depend anything.
|
||||||
|
"""
|
||||||
|
...
|
||||||
@@ -0,0 +1,35 @@
|
|||||||
|
from .common import Source
|
||||||
|
from ..address import Address, AddressDomain
|
||||||
|
from ..environment import Environment
|
||||||
|
from ..sink.fdart_sink import FdArtSink
|
||||||
|
from ..utils import FdContext, FD_THUMBNAIL_HWS
|
||||||
|
from ...lowlevel import artio
|
||||||
|
from ...logger import HIGHLEVEL_LOGGER as LOGGER
|
||||||
|
|
||||||
|
|
||||||
|
class FdArtSource(Source):
|
||||||
|
__kind: FdContext
|
||||||
|
__name: str
|
||||||
|
|
||||||
|
def __init__(self, kind: FdContext, name: str) -> None:
|
||||||
|
self.__kind = kind
|
||||||
|
self.__name = name
|
||||||
|
|
||||||
|
@property
|
||||||
|
def address(self) -> Address:
|
||||||
|
return Address(AddressDomain.FdOutput, str(self.__kind), self.__name)
|
||||||
|
|
||||||
|
def link_to(self, env: Environment, fd_sink: FdArtSink) -> None:
|
||||||
|
sink_hint = fd_sink.push_hint()
|
||||||
|
LOGGER.info(
|
||||||
|
'Duplicating FreeDesktop artifact from kind "%s" name "%s" to kind "%s" name "%s".',
|
||||||
|
self.__kind,
|
||||||
|
self.__name,
|
||||||
|
sink_hint.kind,
|
||||||
|
sink_hint.name,
|
||||||
|
)
|
||||||
|
for hw in FD_THUMBNAIL_HWS:
|
||||||
|
artio.link_artifact(
|
||||||
|
env.fd_output_artifact(hw, self.__kind, self.__name),
|
||||||
|
env.fd_output_artifact(hw, sink_hint.kind, sink_hint.name),
|
||||||
|
)
|
||||||
@@ -0,0 +1,34 @@
|
|||||||
|
from .common import Source
|
||||||
|
from ..address import Address, AddressDomain
|
||||||
|
from ..environment import Environment
|
||||||
|
from ..sink.macart_sink import MacArtSink
|
||||||
|
from ..utils import MacCategory
|
||||||
|
from ...lowlevel import artio
|
||||||
|
from ...logger import HIGHLEVEL_LOGGER as LOGGER
|
||||||
|
|
||||||
|
|
||||||
|
class MacArtSource(Source):
|
||||||
|
__kind: MacCategory
|
||||||
|
__name: str
|
||||||
|
|
||||||
|
def __init__(self, kind: MacCategory, name: str) -> None:
|
||||||
|
self.__kind = kind
|
||||||
|
self.__name = name
|
||||||
|
|
||||||
|
@property
|
||||||
|
def address(self) -> Address:
|
||||||
|
return Address(AddressDomain.MacOutput, str(self.__kind), self.__name)
|
||||||
|
|
||||||
|
def link_to(self, env: Environment, mac_sink: MacArtSink) -> None:
|
||||||
|
sink_hint = mac_sink.push_hint()
|
||||||
|
LOGGER.info(
|
||||||
|
'Duplicating macOS-only artifact from kind "%s" name "%s" to kind "%s" name "%s".',
|
||||||
|
self.__kind,
|
||||||
|
self.__name,
|
||||||
|
sink_hint.kind,
|
||||||
|
sink_hint.name,
|
||||||
|
)
|
||||||
|
artio.link_artifact(
|
||||||
|
env.mac_output_artifact(self.__kind, self.__name),
|
||||||
|
env.mac_output_artifact(sink_hint.kind, sink_hint.name),
|
||||||
|
)
|
||||||
@@ -0,0 +1,16 @@
|
|||||||
|
from .common import Source
|
||||||
|
from ..address import Address
|
||||||
|
from ..utils import Artifact, VART_HW
|
||||||
|
from ...lowlevel import artio
|
||||||
|
|
||||||
|
|
||||||
|
class NewArtSource(Source):
|
||||||
|
def __init__(self) -> None:
|
||||||
|
pass
|
||||||
|
|
||||||
|
@property
|
||||||
|
def address(self) -> Address | None:
|
||||||
|
return None
|
||||||
|
|
||||||
|
def pull(self) -> Artifact:
|
||||||
|
return artio.new_artifact((VART_HW, VART_HW))
|
||||||
@@ -0,0 +1,52 @@
|
|||||||
|
from pathlib import Path
|
||||||
|
from typing import Optional
|
||||||
|
from .common import Source
|
||||||
|
from ..address import Address
|
||||||
|
from ..environment import Environment
|
||||||
|
from ..sink.tempart_sink import TempArtSink
|
||||||
|
from ..utils import Artifact, VART_HW
|
||||||
|
from ...lowlevel import artio, artrdr
|
||||||
|
from ...lowlevel.artrdr import SvgConfig
|
||||||
|
from ...logger import HIGHLEVEL_LOGGER as LOGGER
|
||||||
|
|
||||||
|
DEFAULT_SVG_CONFIG = SvgConfig(VART_HW, VART_HW)
|
||||||
|
|
||||||
|
|
||||||
|
class SvgSource(Source):
|
||||||
|
"""The source to SVG located in input directory."""
|
||||||
|
|
||||||
|
__comps: tuple[str, ...]
|
||||||
|
|
||||||
|
def __init__(self, *args: str) -> None:
|
||||||
|
self.__comps = tuple(args)
|
||||||
|
|
||||||
|
@property
|
||||||
|
def address(self) -> Address | None:
|
||||||
|
return None
|
||||||
|
|
||||||
|
def pull(self, env: Environment, cfg: SvgConfig = DEFAULT_SVG_CONFIG) -> Artifact:
|
||||||
|
dst = self.__render(env, None, cfg)
|
||||||
|
return artio.load_artifact(dst)
|
||||||
|
|
||||||
|
def pull_to_temp(
|
||||||
|
self,
|
||||||
|
env: Environment,
|
||||||
|
temp_sink: TempArtSink,
|
||||||
|
cfg: SvgConfig = DEFAULT_SVG_CONFIG,
|
||||||
|
) -> None:
|
||||||
|
self.__render(env, temp_sink.push_hint(env), cfg)
|
||||||
|
|
||||||
|
def __render(self, env: Environment, dst: Optional[Path], cfg: SvgConfig) -> Path:
|
||||||
|
"""
|
||||||
|
Internal used function for rendering SVG.
|
||||||
|
|
||||||
|
:return: The path to destination storing render result.
|
||||||
|
"""
|
||||||
|
src = env.input_artifact(*self.__comps)
|
||||||
|
if dst is None:
|
||||||
|
LOGGER.info("Rendering temporary SVG: %s", src)
|
||||||
|
dst = env.allocate_lib_render_temporary_artifact()
|
||||||
|
else:
|
||||||
|
LOGGER.info("Rendering SVG: %s -> %s", src, dst)
|
||||||
|
artrdr.render_svg(src, dst, env.prologue.svg_render, cfg)
|
||||||
|
return dst
|
||||||
@@ -0,0 +1,24 @@
|
|||||||
|
from .common import Source
|
||||||
|
from ..address import Address, AddressDomain
|
||||||
|
from ..environment import Environment
|
||||||
|
from ..utils import Artifact
|
||||||
|
from ...lowlevel import artio
|
||||||
|
from ...logger import HIGHLEVEL_LOGGER as LOGGER
|
||||||
|
|
||||||
|
|
||||||
|
class TempArtSource(Source):
|
||||||
|
"""The source to artifact (PNG-only) located in temporary directory."""
|
||||||
|
|
||||||
|
__comps: tuple[str, ...]
|
||||||
|
|
||||||
|
def __init__(self, *args: str) -> None:
|
||||||
|
self.__comps = tuple(args)
|
||||||
|
|
||||||
|
@property
|
||||||
|
def address(self) -> Address | None:
|
||||||
|
return Address(AddressDomain.Temporary, *self.__comps)
|
||||||
|
|
||||||
|
def pull(self, env: Environment) -> Artifact:
|
||||||
|
p = env.user_temporary_artifact(*self.__comps)
|
||||||
|
LOGGER.info("Loading temporary artifact: %s", p)
|
||||||
|
return artio.load_artifact(p)
|
||||||
@@ -0,0 +1,34 @@
|
|||||||
|
from .common import Source
|
||||||
|
from ..address import Address, AddressDomain
|
||||||
|
from ..environment import Environment
|
||||||
|
from ..sink.winart_sink import WinArtSink
|
||||||
|
from ..utils import WinCategory
|
||||||
|
from ...lowlevel import artio
|
||||||
|
from ...logger import HIGHLEVEL_LOGGER as LOGGER
|
||||||
|
|
||||||
|
|
||||||
|
class WinArtSource(Source):
|
||||||
|
__kind: WinCategory
|
||||||
|
__name: str
|
||||||
|
|
||||||
|
def __init__(self, kind: WinCategory, name: str) -> None:
|
||||||
|
self.__kind = kind
|
||||||
|
self.__name = name
|
||||||
|
|
||||||
|
@property
|
||||||
|
def address(self) -> Address:
|
||||||
|
return Address(AddressDomain.WinOutput, str(self.__kind), self.__name)
|
||||||
|
|
||||||
|
def link_to(self, env: Environment, win_sink: WinArtSink) -> None:
|
||||||
|
sink_hint = win_sink.push_hint()
|
||||||
|
LOGGER.info(
|
||||||
|
'Duplicating Windows-only artifact from kind "%s" name "%s" to kind "%s" name "%s".',
|
||||||
|
self.__kind,
|
||||||
|
self.__name,
|
||||||
|
sink_hint.kind,
|
||||||
|
sink_hint.name,
|
||||||
|
)
|
||||||
|
artio.link_artifact(
|
||||||
|
env.win_output_artifact(self.__kind, self.__name),
|
||||||
|
env.win_output_artifact(sink_hint.kind, sink_hint.name),
|
||||||
|
)
|
||||||
@@ -0,0 +1,46 @@
|
|||||||
|
import enum
|
||||||
|
from ..lowlevel.utils import Artifact
|
||||||
|
|
||||||
|
VART_HW: int = 1024
|
||||||
|
"""The height or width of Vanilla Artifact."""
|
||||||
|
|
||||||
|
|
||||||
|
def check_vart(art: Artifact) -> None:
|
||||||
|
if art.size != (VART_HW, VART_HW):
|
||||||
|
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)
|
||||||
@@ -0,0 +1,24 @@
|
|||||||
|
import logging
|
||||||
|
|
||||||
|
|
||||||
|
class LoggerFactory:
|
||||||
|
def __init__(self) -> None:
|
||||||
|
# setup log format and level
|
||||||
|
logging.basicConfig(
|
||||||
|
format="[%(levelname)s] [%(name)s] %(message)s", level=logging.INFO
|
||||||
|
)
|
||||||
|
|
||||||
|
def get_app_logger(self) -> logging.Logger:
|
||||||
|
return logging.getLogger("App")
|
||||||
|
|
||||||
|
def get_lowlevel_logger(self) -> logging.Logger:
|
||||||
|
return logging.getLogger("Lowlevel")
|
||||||
|
|
||||||
|
def get_highlevel_logger(self) -> logging.Logger:
|
||||||
|
return logging.getLogger("Highlevel")
|
||||||
|
|
||||||
|
|
||||||
|
LOGGER_FACTORY = LoggerFactory()
|
||||||
|
LOWLEVEL_LOGGER = LOGGER_FACTORY.get_lowlevel_logger()
|
||||||
|
HIGHLEVEL_LOGGER = LOGGER_FACTORY.get_highlevel_logger()
|
||||||
|
APP_LOGGER = LOGGER_FACTORY.get_app_logger()
|
||||||
@@ -0,0 +1,153 @@
|
|||||||
|
import shutil
|
||||||
|
import os
|
||||||
|
import sys
|
||||||
|
import enum
|
||||||
|
from pathlib import Path
|
||||||
|
import PIL.Image
|
||||||
|
from .utils import Artifact
|
||||||
|
from ..logger import LOWLEVEL_LOGGER as LOGGER
|
||||||
|
|
||||||
|
|
||||||
|
_LINK_MODE_ENV: str = "ASSEMBLICON_LINK_MODE"
|
||||||
|
"""The environment variable selecting how artifacts are linked."""
|
||||||
|
|
||||||
|
|
||||||
|
class LinkMode(enum.StrEnum):
|
||||||
|
"""All valid values of the ``ASSEMBLICON_LINK_MODE`` environment variable."""
|
||||||
|
|
||||||
|
Copy = "copy"
|
||||||
|
Symlink = "symlink"
|
||||||
|
|
||||||
|
|
||||||
|
def new_artifact(size: tuple[int, int]) -> Artifact:
|
||||||
|
"""
|
||||||
|
Create new RGBA empty (#ffffff00) artifact with given size.
|
||||||
|
"""
|
||||||
|
return PIL.Image.new("RGBA", size, "#ffffff00")
|
||||||
|
|
||||||
|
|
||||||
|
def load_input_artifact(p: Path) -> Artifact:
|
||||||
|
"""
|
||||||
|
Load bitmap artifact (PNG, JPEG or BMP) which usually is located in input directory.
|
||||||
|
"""
|
||||||
|
LOGGER.debug("Loading input artifact: %s", p)
|
||||||
|
return PIL.Image.open(p, "r", ["PNG", "JPEG", "BMP"])
|
||||||
|
|
||||||
|
|
||||||
|
def load_artifact(p: Path) -> Artifact:
|
||||||
|
"""
|
||||||
|
Load bitmap artifact (PNG-only) saved previously as the intermediary.
|
||||||
|
"""
|
||||||
|
LOGGER.debug("Loading temporary or output artifact: %s", p)
|
||||||
|
return PIL.Image.open(p, "r", ["PNG"])
|
||||||
|
|
||||||
|
|
||||||
|
def save_artifact(art: Artifact, p: Path) -> None:
|
||||||
|
LOGGER.debug("Saving artifact: %s", p)
|
||||||
|
art.save(p, format="PNG")
|
||||||
|
|
||||||
|
|
||||||
|
def dup_artifact(src: Path, dst: Path) -> None:
|
||||||
|
LOGGER.debug("Duplicating artifact: %s -> %s", src, dst)
|
||||||
|
shutil.copyfile(src, dst)
|
||||||
|
|
||||||
|
|
||||||
|
def link_artifact(src: Path, dst: Path) -> None:
|
||||||
|
"""
|
||||||
|
Make given artifact reachable from another path, either via a
|
||||||
|
relative symlink or a content copy.
|
||||||
|
|
||||||
|
The linking mode is controlled by the ``ASSEMBLICON_LINK_MODE``
|
||||||
|
environment variable: ``symlink`` (default) creates a symlink with a
|
||||||
|
path relative to the destination; ``copy`` duplicates the artifact
|
||||||
|
content. An existing destination is only removed after it is proven
|
||||||
|
to be a regular file, or a symlink whose target is a file path
|
||||||
|
(the target is allowed to be missing), and only when it shares the
|
||||||
|
same filename suffix as the source. Symlink creation failure falls
|
||||||
|
back to copying.
|
||||||
|
|
||||||
|
:param src: The path to source artifact file.
|
||||||
|
:param dst: The path to destination which should reach the artifact.
|
||||||
|
:raises ValueError: If the link mode is unknown, the destination
|
||||||
|
exists but is not safely deletable (see above), its suffix
|
||||||
|
differs from the source one, or no common directory exists
|
||||||
|
between source and destination.
|
||||||
|
"""
|
||||||
|
try:
|
||||||
|
mode = LinkMode(os.getenv(_LINK_MODE_ENV, LinkMode.Symlink))
|
||||||
|
except ValueError:
|
||||||
|
raise ValueError(
|
||||||
|
f"Invalid link mode for env {_LINK_MODE_ENV}. "
|
||||||
|
f"Valid modes are: {', '.join(m.value for m in LinkMode)}."
|
||||||
|
)
|
||||||
|
|
||||||
|
LOGGER.debug("Linking artifact: %s -> %s (mode: %s)", src, dst, mode)
|
||||||
|
|
||||||
|
# SAFETY: the code below deletes an existing destination file. These
|
||||||
|
# checks are the only guards against accidental data deletion, so
|
||||||
|
# never relax them. A destination is deletable only when it is a
|
||||||
|
# regular file, or a symlink whose target is a file path (the target
|
||||||
|
# itself is allowed to be missing, i.e. a broken symlink is
|
||||||
|
# deletable). Additionally, the source and the destination must
|
||||||
|
# share the same filename suffix.
|
||||||
|
if os.path.lexists(dst):
|
||||||
|
if dst.is_symlink():
|
||||||
|
# a symlink is deletable unless it provably points to a
|
||||||
|
# non-file, e.g. an existing directory
|
||||||
|
if dst.exists() and not dst.is_file():
|
||||||
|
raise ValueError(
|
||||||
|
f'Refusing to remove existing destination "{dst}": '
|
||||||
|
f"it is a symlink not pointing to a file."
|
||||||
|
)
|
||||||
|
elif not dst.is_file():
|
||||||
|
raise ValueError(
|
||||||
|
f'Refusing to remove existing destination "{dst}": '
|
||||||
|
f"it is not a regular file."
|
||||||
|
)
|
||||||
|
if src.suffix != dst.suffix:
|
||||||
|
raise ValueError(
|
||||||
|
f'Refusing to remove existing destination "{dst}": its '
|
||||||
|
f'suffix "{dst.suffix}" differs from source suffix "{src.suffix}".'
|
||||||
|
)
|
||||||
|
dst.unlink()
|
||||||
|
|
||||||
|
if mode == LinkMode.Copy:
|
||||||
|
shutil.copyfile(src, dst)
|
||||||
|
return
|
||||||
|
|
||||||
|
# symlinks are always created with a path relative to the
|
||||||
|
# destination, so that the whole artifact directory stays relocatable
|
||||||
|
try:
|
||||||
|
link_target = src.relative_to(dst.parent, walk_up=True)
|
||||||
|
except ValueError:
|
||||||
|
raise ValueError(
|
||||||
|
f'No common directory between "{src}" and "{dst}", '
|
||||||
|
f"can not create relative symlink."
|
||||||
|
)
|
||||||
|
|
||||||
|
# SAFETY: the symlink target must always be stored with POSIX
|
||||||
|
# separators ("/"). On Windows, "relative_to" yields a
|
||||||
|
# backslash-separated path, and "symlink_to" stores the target string
|
||||||
|
# verbatim into the NTFS reparse point, so a backslash form would
|
||||||
|
# also leak verbatim into tar archives. Whether an unpacker accepts
|
||||||
|
# backslash targets is not guaranteed, and we deliberately do not
|
||||||
|
# rely on the packer normalizing them. Converting to POSIX form here
|
||||||
|
# is the single point guaranteeing that archived symlinks stay
|
||||||
|
# portable, regardless of the platform the build ran on or the
|
||||||
|
# packer used.
|
||||||
|
try:
|
||||||
|
dst.symlink_to(link_target.as_posix())
|
||||||
|
except OSError as e:
|
||||||
|
LOGGER.warning(
|
||||||
|
"Fail to create symlink for artifact %s: %s. Falling back to copy.",
|
||||||
|
dst,
|
||||||
|
e,
|
||||||
|
)
|
||||||
|
if sys.platform == "win32":
|
||||||
|
LOGGER.warning(
|
||||||
|
"Symlink creation on Windows usually requires administrator "
|
||||||
|
"privileges. Consider rerunning as administrator, enabling "
|
||||||
|
"Developer Mode, or setting %s=copy.",
|
||||||
|
_LINK_MODE_ENV,
|
||||||
|
)
|
||||||
|
shutil.copyfile(src, dst)
|
||||||
@@ -0,0 +1,136 @@
|
|||||||
|
import enum
|
||||||
|
import math
|
||||||
|
import PIL.Image
|
||||||
|
from .utils import Artifact
|
||||||
|
from ..geometry import Point
|
||||||
|
|
||||||
|
|
||||||
|
class HorizontalAnchor(enum.IntEnum):
|
||||||
|
Left = enum.auto()
|
||||||
|
Center = enum.auto()
|
||||||
|
Right = enum.auto()
|
||||||
|
|
||||||
|
|
||||||
|
class VerticalAnchor(enum.IntEnum):
|
||||||
|
Top = enum.auto()
|
||||||
|
Center = enum.auto()
|
||||||
|
Bottom = enum.auto()
|
||||||
|
|
||||||
|
|
||||||
|
def anchor_to_offset(
|
||||||
|
oversize_width: int,
|
||||||
|
oversize_height: int,
|
||||||
|
horizontal_anchor: HorizontalAnchor,
|
||||||
|
vertical_anchor: VerticalAnchor,
|
||||||
|
) -> Point:
|
||||||
|
x: int
|
||||||
|
match horizontal_anchor:
|
||||||
|
case HorizontalAnchor.Left:
|
||||||
|
x = 0
|
||||||
|
case HorizontalAnchor.Center:
|
||||||
|
x = oversize_width // 2
|
||||||
|
case HorizontalAnchor.Right:
|
||||||
|
x = oversize_width
|
||||||
|
|
||||||
|
y: int
|
||||||
|
match vertical_anchor:
|
||||||
|
case VerticalAnchor.Top:
|
||||||
|
y = 0
|
||||||
|
case VerticalAnchor.Center:
|
||||||
|
y = oversize_height // 2
|
||||||
|
case VerticalAnchor.Bottom:
|
||||||
|
y = oversize_height
|
||||||
|
|
||||||
|
return Point(x, y)
|
||||||
|
|
||||||
|
|
||||||
|
def aspect_ratio_scale_and_crop(
|
||||||
|
art: Artifact,
|
||||||
|
target_width: int,
|
||||||
|
target_height: int,
|
||||||
|
scale_resample: int,
|
||||||
|
crop_horizontal_anchor: HorizontalAnchor,
|
||||||
|
crop_vertical_anchor: VerticalAnchor,
|
||||||
|
) -> Artifact:
|
||||||
|
if target_width <= 0:
|
||||||
|
raise ValueError("Target width must be a positive integer")
|
||||||
|
if target_height <= 0:
|
||||||
|
raise ValueError("Target height must be positive integer")
|
||||||
|
|
||||||
|
scale = max(target_width / art.width, target_height / art.height)
|
||||||
|
# use math.ceil to ensure that resized image at least has required size
|
||||||
|
# (floating point value error may cause this)
|
||||||
|
scaled_width = math.ceil(art.width * scale)
|
||||||
|
scaled_height = math.ceil(art.height * scale)
|
||||||
|
scaled = art.resize((scaled_width, scaled_height), scale_resample)
|
||||||
|
|
||||||
|
crop = anchor_to_offset(
|
||||||
|
scaled_width - target_width,
|
||||||
|
scaled_height - target_height,
|
||||||
|
crop_horizontal_anchor,
|
||||||
|
crop_vertical_anchor,
|
||||||
|
)
|
||||||
|
(crop_x, crop_y) = crop.to_tuple()
|
||||||
|
return scaled.crop((crop_x, crop_y, crop_x + target_width, crop_y + target_height))
|
||||||
|
|
||||||
|
|
||||||
|
def default_art_resize(art: Artifact, width: int, height: int) -> Artifact:
|
||||||
|
"""
|
||||||
|
Resize given artifact with default resample (LANCZOS).
|
||||||
|
"""
|
||||||
|
return art.resize((width, height), PIL.Image.Resampling.LANCZOS)
|
||||||
|
|
||||||
|
|
||||||
|
def art_anchor_to_offset(
|
||||||
|
art: Artifact,
|
||||||
|
horizontal_anchor: HorizontalAnchor,
|
||||||
|
vertical_anchor: VerticalAnchor,
|
||||||
|
) -> Point:
|
||||||
|
"""
|
||||||
|
Resolve the anchor point position inside given artifact.
|
||||||
|
|
||||||
|
:param art: The artifact whose anchor point should be resolved.
|
||||||
|
:param horizontal_anchor: The anchor selection at horizontal direction.
|
||||||
|
:param vertical_anchor: The anchor selection at vertical direction.
|
||||||
|
:return: The (x, y) coordinate of the anchor point. For Left/Top
|
||||||
|
anchors this is the corresponding edge; for Right/Bottom anchors
|
||||||
|
this is just outside the artifact (the exact corner coordinate).
|
||||||
|
"""
|
||||||
|
return anchor_to_offset(*art.size, horizontal_anchor, vertical_anchor)
|
||||||
|
|
||||||
|
|
||||||
|
def align_paste(
|
||||||
|
art: Artifact,
|
||||||
|
clipboard: Artifact,
|
||||||
|
art_pos: Point,
|
||||||
|
clipboard_pos: Point,
|
||||||
|
) -> None:
|
||||||
|
"""
|
||||||
|
Paste clipboard onto given artifact with two anchor points aligned.
|
||||||
|
|
||||||
|
The clipboard is pasted in the way that the point indicated by
|
||||||
|
clipboard_pos on the clipboard exactly overlaps the point indicated
|
||||||
|
by art_pos on the artifact. The art is modified in place. Positions
|
||||||
|
resolving to negative paste origins are allowed and clipped by PIL
|
||||||
|
just like a normal paste call.
|
||||||
|
|
||||||
|
:param art: The artifact pasted onto. Modified in place.
|
||||||
|
:param clipboard: The artifact pasted as the clipboard.
|
||||||
|
:param art_pos: The (x, y) point on the art to be aligned, usually
|
||||||
|
resolved by anchor_to_pos.
|
||||||
|
:param clipboard_pos: The (x, y) point on the clipboard to be
|
||||||
|
aligned, usually resolved by anchor_to_pos.
|
||||||
|
"""
|
||||||
|
art.paste(clipboard, (art_pos - clipboard_pos).to_tuple())
|
||||||
|
|
||||||
|
|
||||||
|
def align_alpha_composite(
|
||||||
|
art: Artifact,
|
||||||
|
clipboard: Artifact,
|
||||||
|
art_pos: Point,
|
||||||
|
clipboard_pos: Point,
|
||||||
|
) -> None:
|
||||||
|
"""
|
||||||
|
Alpha-supported ``align_paste``.
|
||||||
|
"""
|
||||||
|
art.alpha_composite(clipboard, (art_pos - clipboard_pos).to_tuple())
|
||||||
@@ -0,0 +1,232 @@
|
|||||||
|
import os
|
||||||
|
import enum
|
||||||
|
import subprocess
|
||||||
|
from typing import Iterable, Iterator
|
||||||
|
from pathlib import Path
|
||||||
|
from dataclasses import dataclass
|
||||||
|
import PIL.Image
|
||||||
|
from .utils import Artifact
|
||||||
|
from ..logger import LOWLEVEL_LOGGER as LOGGER
|
||||||
|
|
||||||
|
|
||||||
|
class SvgRenderKind(enum.IntEnum):
|
||||||
|
Inkscape = enum.auto()
|
||||||
|
ReSvg = enum.auto()
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class SvgConfig:
|
||||||
|
width: int
|
||||||
|
height: int
|
||||||
|
|
||||||
|
def __post_init__(self) -> None:
|
||||||
|
if self.width <= 0:
|
||||||
|
raise ValueError(
|
||||||
|
f"The image width in Inkscape config must be a positive integer"
|
||||||
|
)
|
||||||
|
if self.height <= 0:
|
||||||
|
raise ValueError(
|
||||||
|
f"The image height in Inkscape config must be a positive integer"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def render_svg(src: Path, dst: Path, kind: SvgRenderKind, cfg: SvgConfig) -> None:
|
||||||
|
"""
|
||||||
|
Render given SVG file to PNG file.
|
||||||
|
|
||||||
|
:param src: The path to SVG file to be rendered
|
||||||
|
:param dst: The path to rendered PNG file.
|
||||||
|
:param kind: The selected SVG render kind.
|
||||||
|
:param cfg: The configuration when rendering.
|
||||||
|
"""
|
||||||
|
LOGGER.debug(
|
||||||
|
"Rendering SVG artifact with render %s: %s -> %s", kind.name, src, dst
|
||||||
|
)
|
||||||
|
|
||||||
|
# YYC MARK:
|
||||||
|
# Inkscape will silently quit with return code 0
|
||||||
|
# if given source file doesn't exist in file system.
|
||||||
|
# So we need to check whether it is existing and popularize it to all render.
|
||||||
|
if not src.is_file():
|
||||||
|
raise ValueError(f"Source SVG file is not presented in file system: {src}")
|
||||||
|
|
||||||
|
match kind:
|
||||||
|
case SvgRenderKind.Inkscape:
|
||||||
|
inkscape_bin = os.getenv("ASSEMBLICON_INKSCAPE", "inkscape")
|
||||||
|
cmd = [
|
||||||
|
inkscape_bin,
|
||||||
|
"--export-area-page",
|
||||||
|
"--export-width",
|
||||||
|
str(cfg.width),
|
||||||
|
"--export-height",
|
||||||
|
str(cfg.height),
|
||||||
|
"--export-type",
|
||||||
|
"png",
|
||||||
|
"--export-png-color-mode",
|
||||||
|
"RGBA_8",
|
||||||
|
"--export-background-opacity",
|
||||||
|
"0.0",
|
||||||
|
"--export-png-use-dithering",
|
||||||
|
"false",
|
||||||
|
"--export-filename",
|
||||||
|
str(dst),
|
||||||
|
str(src),
|
||||||
|
]
|
||||||
|
case SvgRenderKind.ReSvg:
|
||||||
|
resvg_bin = os.getenv("ASSEMBLICON_RESVG", "resvg")
|
||||||
|
cmd = [
|
||||||
|
resvg_bin,
|
||||||
|
"--width",
|
||||||
|
str(cfg.width),
|
||||||
|
"--height",
|
||||||
|
str(cfg.height),
|
||||||
|
str(src),
|
||||||
|
str(dst),
|
||||||
|
]
|
||||||
|
|
||||||
|
proc = subprocess.run(cmd, capture_output=False)
|
||||||
|
if proc.returncode != 0:
|
||||||
|
raise RuntimeError(
|
||||||
|
f"Fail to execute SVG render {kind.name}. Return code is {proc.returncode}"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class BlenderConfig:
|
||||||
|
width: int
|
||||||
|
height: int
|
||||||
|
|
||||||
|
def __post_init__(self) -> None:
|
||||||
|
if self.width <= 0:
|
||||||
|
raise ValueError(
|
||||||
|
f"The image width in Blender config must be a positive integer"
|
||||||
|
)
|
||||||
|
if self.height <= 0:
|
||||||
|
raise ValueError(
|
||||||
|
f"The image height in Blender config must be a positive integer"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
_BLENDER_RENDER_SCRIPT = """\
|
||||||
|
import sys
|
||||||
|
import traceback
|
||||||
|
|
||||||
|
try:
|
||||||
|
import bpy
|
||||||
|
|
||||||
|
render = bpy.context.scene.render
|
||||||
|
render.resolution_x = {width}
|
||||||
|
render.resolution_y = {height}
|
||||||
|
render.resolution_percentage = 100
|
||||||
|
render.image_settings.file_format = "PNG"
|
||||||
|
render.image_settings.color_mode = "RGBA"
|
||||||
|
render.image_settings.color_depth = "8"
|
||||||
|
render.film_transparent = True
|
||||||
|
render.use_file_extension = False
|
||||||
|
render.filepath = {dst}
|
||||||
|
bpy.ops.render.render(write_still=True)
|
||||||
|
except BaseException:
|
||||||
|
traceback.print_exc()
|
||||||
|
sys.exit(1)
|
||||||
|
"""
|
||||||
|
|
||||||
|
|
||||||
|
def render_blender(src: Path, dst: Path, temp_script: Path, cfg: BlenderConfig) -> None:
|
||||||
|
"""
|
||||||
|
Render given Blender composition into PNG file.
|
||||||
|
|
||||||
|
:param src: The path to Blender composition file to be rendered
|
||||||
|
:param dst: The path to rendered PNG file.
|
||||||
|
:param temp_script: The path to temporary Python script instructing Blender rendering.
|
||||||
|
The caller must make sure that this file is immutable during calling this function.
|
||||||
|
:param cfg: The configuration when rendering.
|
||||||
|
"""
|
||||||
|
LOGGER.debug("Rendering Blender artifact: %s -> %s", src, dst)
|
||||||
|
|
||||||
|
if not src.is_file():
|
||||||
|
raise ValueError(
|
||||||
|
f"Source Blender composition file is not presented in file system: {src}"
|
||||||
|
)
|
||||||
|
|
||||||
|
script = _BLENDER_RENDER_SCRIPT.format(
|
||||||
|
width=cfg.width, height=cfg.height, dst=repr(str(dst))
|
||||||
|
)
|
||||||
|
temp_script.write_text(script, encoding="utf-8")
|
||||||
|
|
||||||
|
blender_bin = os.getenv("ASSEMBLICON_BLENDER", "blender")
|
||||||
|
cmd = [
|
||||||
|
blender_bin,
|
||||||
|
"--background",
|
||||||
|
str(src),
|
||||||
|
"--python",
|
||||||
|
str(temp_script),
|
||||||
|
]
|
||||||
|
proc = subprocess.run(cmd, capture_output=False)
|
||||||
|
if proc.returncode != 0:
|
||||||
|
raise RuntimeError(f"Fail to execute Blender. Return code is {proc.returncode}")
|
||||||
|
if not dst.exists():
|
||||||
|
raise RuntimeError(
|
||||||
|
f"Blender exited successfully but output file {dst} is missing"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def render_thumbnail(art: Artifact, hws: Iterable[int]) -> Iterator[Artifact]:
|
||||||
|
"""
|
||||||
|
Render given artifact as multiple resolution thumbnails.
|
||||||
|
|
||||||
|
:param art: The artifact to be rendered.
|
||||||
|
:param hws: The int iterable holding all resolution of thumbnails (square thumbnail only)
|
||||||
|
:return: A generator outputing thumbnail one by one with the order of given resolution.
|
||||||
|
"""
|
||||||
|
LOGGER.debug("Rendering thumbnail...")
|
||||||
|
|
||||||
|
# check whether given artifact is square rectangle.
|
||||||
|
if art.width != art.height:
|
||||||
|
raise ValueError(f'Thumbnail render only accept square rectangle.')
|
||||||
|
# scale artifact up to the maximum resolution hws if it is too small.
|
||||||
|
max_hws = max(hws)
|
||||||
|
if art.width < max_hws:
|
||||||
|
art = art.resize((max_hws, max_hws), PIL.Image.Resampling.LANCZOS)
|
||||||
|
|
||||||
|
# render thumbnail
|
||||||
|
for hw in hws:
|
||||||
|
LOGGER.debug("Rendering thumbnail %dx%d.", hw, hw)
|
||||||
|
thumbnail = art.copy()
|
||||||
|
thumbnail.thumbnail((hw, hw), PIL.Image.Resampling.LANCZOS)
|
||||||
|
yield thumbnail
|
||||||
|
|
||||||
|
|
||||||
|
def render_ico(art: Artifact, dst: Path) -> None:
|
||||||
|
"""
|
||||||
|
Render given artifact as Windows ICO in given path.
|
||||||
|
|
||||||
|
:param art: The artifact to be rendered.
|
||||||
|
:param dst: The path to rendered result.
|
||||||
|
"""
|
||||||
|
LOGGER.debug("Rendering .ICO artifact: %s", dst)
|
||||||
|
|
||||||
|
sizes = [16, 32, 48, 64, 128, 256]
|
||||||
|
# provide every frame explicitly so that we never depend on
|
||||||
|
# undocumented plugin-side resizing behavior
|
||||||
|
frames = list(render_thumbnail(art, sizes))
|
||||||
|
art.save(
|
||||||
|
dst,
|
||||||
|
format="ICO",
|
||||||
|
sizes=[(size, size) for size in sizes],
|
||||||
|
append_images=frames,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def render_icns(art: Artifact, dst: Path) -> None:
|
||||||
|
"""
|
||||||
|
Render given artifact as macOS ICNS in given path.
|
||||||
|
|
||||||
|
:param art: The artifact to be rendered.
|
||||||
|
:param dst: The path to rendered result.
|
||||||
|
"""
|
||||||
|
LOGGER.debug("Rendering .ICNS artifact: %s", dst)
|
||||||
|
|
||||||
|
# provide every smaller frame explicitly; the 1024x1024 entry is
|
||||||
|
# covered by the art itself
|
||||||
|
frames = list(render_thumbnail(art, (512, 256, 128, 64, 32)))
|
||||||
|
art.save(dst, format="ICNS", append_images=frames)
|
||||||
@@ -0,0 +1,68 @@
|
|||||||
|
import tempfile
|
||||||
|
from dataclasses import dataclass
|
||||||
|
from types import TracebackType
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass(frozen=True)
|
||||||
|
class Prologue:
|
||||||
|
input_directory: Path
|
||||||
|
"""The path to input directory."""
|
||||||
|
output_directory: Path
|
||||||
|
"""The path to output directory."""
|
||||||
|
|
||||||
|
def __post_init__(self) -> None:
|
||||||
|
if not self.input_directory.is_dir():
|
||||||
|
raise ValueError(
|
||||||
|
f"Given input path {self.input_directory} is not an existing directory."
|
||||||
|
)
|
||||||
|
if not self.output_directory.is_dir():
|
||||||
|
raise ValueError(
|
||||||
|
f"Given output path {self.output_directory} is not an existing directory."
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
class Environment:
|
||||||
|
__prologue: Prologue
|
||||||
|
"""The prologue used for environment."""
|
||||||
|
__temp_dir: tempfile.TemporaryDirectory[str]
|
||||||
|
"""Allocated temporary directory."""
|
||||||
|
|
||||||
|
def __init__(self, prologue: Prologue) -> None:
|
||||||
|
self.__prologue = prologue
|
||||||
|
self.__temp_dir = tempfile.TemporaryDirectory()
|
||||||
|
|
||||||
|
def __enter__(self) -> "Environment":
|
||||||
|
return self
|
||||||
|
|
||||||
|
def __exit__(
|
||||||
|
self,
|
||||||
|
exc_type: type[BaseException] | None,
|
||||||
|
exc_value: BaseException | None,
|
||||||
|
traceback: TracebackType | None,
|
||||||
|
) -> None:
|
||||||
|
self.__temp_dir.cleanup()
|
||||||
|
|
||||||
|
def input_artifact(self, *args: str) -> Path:
|
||||||
|
"""
|
||||||
|
Get the path to input artifact.
|
||||||
|
|
||||||
|
:return: The built absolute path to input artifact.
|
||||||
|
"""
|
||||||
|
return self.__prologue.input_directory / Path(*args)
|
||||||
|
|
||||||
|
def temporary_artifact(self, *args: str) -> Path:
|
||||||
|
"""
|
||||||
|
Get the path to temporary artifact.
|
||||||
|
|
||||||
|
:return: The built absolute path to temporary artifact.
|
||||||
|
"""
|
||||||
|
return Path(self.__temp_dir.name) / Path(*args)
|
||||||
|
|
||||||
|
def output_artifact(self, *args: str) -> Path:
|
||||||
|
"""
|
||||||
|
Get the path to output artifact.
|
||||||
|
|
||||||
|
:return: The built absolute path to output artifact.
|
||||||
|
"""
|
||||||
|
return self.__prologue.output_directory / Path(*args)
|
||||||
@@ -0,0 +1,3 @@
|
|||||||
|
from PIL.Image import Image
|
||||||
|
|
||||||
|
type Artifact = Image
|
||||||
@@ -1,121 +0,0 @@
|
|||||||
import logging
|
|
||||||
import os
|
|
||||||
import subprocess
|
|
||||||
from pathlib import Path
|
|
||||||
from dataclasses import dataclass
|
|
||||||
|
|
||||||
from .context import Context
|
|
||||||
|
|
||||||
|
|
||||||
@dataclass
|
|
||||||
class InkscapeConfig:
|
|
||||||
width: int
|
|
||||||
height: int
|
|
||||||
|
|
||||||
def __post_init__(self) -> None:
|
|
||||||
if self.width <= 0:
|
|
||||||
raise ValueError(
|
|
||||||
f"The image width in Inkscape config must be a positive integer"
|
|
||||||
)
|
|
||||||
if self.height <= 0:
|
|
||||||
raise ValueError(
|
|
||||||
f"The image height in Inkscape config must be a positive integer"
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
def inkscape_render(src: Path, dst: Path, cfg: InkscapeConfig) -> None:
|
|
||||||
logging.info("Rendering Inkscape artifact: %s -> %s", src, dst)
|
|
||||||
|
|
||||||
inkscape_bin = os.getenv("ASSEMBLICON_INKSCAPE", "inkscape")
|
|
||||||
cmd = [
|
|
||||||
inkscape_bin,
|
|
||||||
"--export-area-page",
|
|
||||||
"--export-width",
|
|
||||||
str(cfg.width),
|
|
||||||
"--export-height",
|
|
||||||
str(cfg.height),
|
|
||||||
"--export-type",
|
|
||||||
"png",
|
|
||||||
"--export-png-color-mode",
|
|
||||||
"RGBA_8",
|
|
||||||
"--export-background-opacity",
|
|
||||||
"0.0",
|
|
||||||
"--export-png-use-dithering",
|
|
||||||
"false",
|
|
||||||
"--export-filename",
|
|
||||||
str(dst),
|
|
||||||
str(src),
|
|
||||||
]
|
|
||||||
proc = subprocess.run(cmd, capture_output=False)
|
|
||||||
if proc.returncode != 0:
|
|
||||||
raise RuntimeError(
|
|
||||||
f"Fail to execute Inkscape. Return code is {proc.returncode}"
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
@dataclass
|
|
||||||
class BlenderConfig:
|
|
||||||
width: int
|
|
||||||
height: int
|
|
||||||
|
|
||||||
def __post_init__(self) -> None:
|
|
||||||
if self.width <= 0:
|
|
||||||
raise ValueError(
|
|
||||||
f"The image width in Blender config must be a positive integer"
|
|
||||||
)
|
|
||||||
if self.height <= 0:
|
|
||||||
raise ValueError(
|
|
||||||
f"The image height in Blender config must be a positive integer"
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
_BLENDER_RENDER_SCRIPT = """\
|
|
||||||
import sys
|
|
||||||
import traceback
|
|
||||||
|
|
||||||
try:
|
|
||||||
import bpy
|
|
||||||
|
|
||||||
render = bpy.context.scene.render
|
|
||||||
render.resolution_x = {width}
|
|
||||||
render.resolution_y = {height}
|
|
||||||
render.resolution_percentage = 100
|
|
||||||
render.image_settings.file_format = "PNG"
|
|
||||||
render.image_settings.color_mode = "RGBA"
|
|
||||||
render.image_settings.color_depth = "8"
|
|
||||||
render.film_transparent = True
|
|
||||||
render.use_file_extension = False
|
|
||||||
render.filepath = {dst}
|
|
||||||
bpy.ops.render.render(write_still=True)
|
|
||||||
except BaseException:
|
|
||||||
traceback.print_exc()
|
|
||||||
sys.exit(1)
|
|
||||||
"""
|
|
||||||
|
|
||||||
|
|
||||||
def blender_render(ctx: Context, src: Path, dst: Path, cfg: BlenderConfig) -> None:
|
|
||||||
logging.info("Rendering Blender artifact: %s -> %s", src, dst)
|
|
||||||
|
|
||||||
script = _BLENDER_RENDER_SCRIPT.format(
|
|
||||||
width=cfg.width, height=cfg.height, dst=repr(str(dst))
|
|
||||||
)
|
|
||||||
script_path = ctx._intern_temporary_artifact("blender_render.py")
|
|
||||||
script_path.write_text(script, encoding="utf-8")
|
|
||||||
|
|
||||||
blender_bin = os.getenv("ASSEMBLICON_BLENDER", "blender")
|
|
||||||
cmd = [
|
|
||||||
blender_bin,
|
|
||||||
"--background",
|
|
||||||
str(src),
|
|
||||||
"--python",
|
|
||||||
str(script_path),
|
|
||||||
]
|
|
||||||
proc = subprocess.run(cmd, capture_output=False)
|
|
||||||
if proc.returncode != 0:
|
|
||||||
raise RuntimeError(
|
|
||||||
f"Fail to execute Blender. Return code is {proc.returncode}"
|
|
||||||
)
|
|
||||||
if not dst.exists():
|
|
||||||
raise RuntimeError(
|
|
||||||
f"Blender exited successfully but output file {dst} is missing"
|
|
||||||
)
|
|
||||||
@@ -1,4 +0,0 @@
|
|||||||
import logging
|
|
||||||
|
|
||||||
def setup_logging() -> None:
|
|
||||||
logging.basicConfig(format="[%(levelname)s] %(message)s", level=logging.INFO)
|
|
||||||
Reference in New Issue
Block a user