Skip to content

Model Card for DS-GAN

paper venue license base dataset vendor-parity hub

This package ports DS-GAN, the content-aware poster layout generator from PosterLayout, into a 🤗transformers-style package.

Model Details

Model Description

DS-GAN is a content-aware poster layout generator that predicts text, logo, and underlay element geometry from an RGB poster canvas plus saliency maps. The package wraps the released generator as DSGANModel, DSGANProcessor, and DSGANPipeline; public outputs use normalized center xywh boxes in [0, 1], dataset-local integer labels, a valid-element mask, and id2label.

  • Developed by: HsiaoYuan Hsu, Xiangteng He, Yuxin Peng, Hao Kong, and Qing Zhang.
  • Shared by: creative-graphic-design.
  • Model type: content-aware; task: single-task; conditioning: none.
  • Language(s) (NLP): not applicable.
  • License: upstream license review needed before publishing converted weights.

Model Sources

Supported Checkpoints

Checkpoint Hub ID Status
PKU PosterLayout DS-GAN Epoch 300 creative-graphic-design/ds-gan-pku-posterlayout not-published

Uses

Direct Use

Use this package for research inference, conversion checks, and agreement checks against the original implementation for content-aware poster layout generation.

Conditional generation accepts an RGB content image plus one saliency map, or PFPNet and BASNet saliency maps that are merged before resizing:

from ds_gan import DSGANPipeline

pipe = DSGANPipeline.from_pretrained(
    "creative-graphic-design/ds-gan-pku-posterlayout"
)
out = pipe(images=image, saliency=saliency, seed=0)
print(out.bbox, out.labels, out.mask)

Model/processor usage is also available for lower-level checks:

from ds_gan import DSGANModel, DSGANProcessor

model = DSGANModel.from_pretrained(
    "creative-graphic-design/ds-gan-pku-posterlayout"
).eval()
processor = DSGANProcessor.from_pretrained(
    "creative-graphic-design/ds-gan-pku-posterlayout"
)
encoded = processor(images=image, saliency=saliency)
out = model(**encoded)
decoded = processor.decode(class_probs=out.class_probs, bbox=out.bbox)
print(decoded.bbox, decoded.labels, decoded.mask)

Downstream Use

Generated layouts may feed rendering, design tooling, layout evaluation, or downstream poster composition systems after task-specific validation.

Out-of-Scope Use

Do not treat generated layouts as production accessibility annotations, OCR output, semantic scene understanding, or license-cleared design assets without separate review.

Bias, Risks, and Limitations

The converted behavior follows the upstream checkpoint, saliency preprocessing path, and PosterLayout data release. Dataset coverage, visual style, label vocabulary, and layout quality inherit the limits of those sources. The upstream repository does not include a top-level redistribution license.

Recommendations

Re-run the agreement-check suite against the original implementation before publishing converted checkpoints or comparing new results. Review generated layouts before downstream use and evaluate separately for each target poster domain.

How to Get Started with the Model

Install the package directly from this repository. The command includes shared packages when they are not published on PyPI.

pip install \
  "laygen @ git+https://github.com/creative-graphic-design/design-generators.git#subdirectory=lib/laygen" \
  "posgen @ git+https://github.com/creative-graphic-design/design-generators.git#subdirectory=lib/posgen" \
  "ds-gan @ git+https://github.com/creative-graphic-design/design-generators.git#subdirectory=models/ds-gan"

Clone this repository, install the workspace member, and run the download and conversion steps in REPRODUCING.md. Those steps create .cache/ds-gan/converted/ds-gan-pku-posterlayout.

git clone https://github.com/creative-graphic-design/design-generators.git
cd design-generators
uv sync --package ds-gan
uv run --package ds-gan python
import torch

from ds_gan import DSGANPipeline

path = ".cache/ds-gan/converted/ds-gan-pku-posterlayout"
# After Hub publication: from_pretrained("creative-graphic-design/ds-gan-pku-posterlayout")
pipe = DSGANPipeline.from_pretrained(path, local_files_only=True)
out = pipe(pixel_values=torch.zeros(1, 4, 350, 240), seed=0)

print(out.bbox)
print(out.labels)
print(out.mask)

Training Details

Training Data

Dataset Dataset ID Notes
PKU-PosterLayout creative-graphic-design/PKU-PosterLayout PKU release agreement; box_elem stores pixel ltrb boxes and cls_elem includes text, logo, underlay, and INVALID

Training Procedure

This package ports released behavior and does not retrain the method in this repository.

Preprocessing

Inputs and outputs are normalized to the public layout schema at package boundaries. RGB canvases and saliency maps are resized to (350, 240). PFPNet and BASNet saliency are merged by native-resolution pixelwise max before resize. The original class id 0 represents no object and becomes mask=False.

Training Hyperparameters

  • Training regime: original upstream training; not rerun in this repository.

Speeds, Sizes, Times

Training-time and carbon measurements are unknown.

Evaluation

Testing Data, Factors & Metrics

Testing Data

Agreement checks use local-only generated fixtures from the released DS-GAN checkpoint and PosterLayout test canvases. Large generated tensors, images, weights, and downloaded artifacts are not committed.

Factors

Agreement checks are scoped to the PKU PosterLayout DS-GAN generator, fixed seed 0, content-image conditioning, and 905 test canvases.

Metrics

Metrics are exact tensor equality for model outputs, exact processor pixel-value equality against the regenerated original-implementation fixture, and exact public postprocessing equality for bbox, labels, and mask.

Parity Results

Check Cases Match criterion Result
Original-implementation DS-GAN class probabilities 905 canvases x 32 slots x 4 classes rtol=0, atol=0, max_abs=0.0, mismatched=0 passed
Original-implementation DS-GAN boxes 905 canvases x 32 slots x 4 values rtol=0, atol=0, max_abs=0.0, mismatched=0 passed
Processor pixel_values 905 canvases x 4 x 350 x 240 pixels max_abs=0.0, mismatched=0 against regenerated fixture passed
Public postprocessing bbox, labels, and mask for 905 decoded layouts exact equality against decoded original-implementation fixture passed

Reproducibility

See REPRODUCING.md for the commands that download original-implementation assets, generate reference outputs, run agreement checks, convert checkpoints, and smoke-test local loading.

Environmental Impact

No new model training is performed by this conversion package. Conversion and agreement-check costs depend on local hardware and the selected checkpoint.

Technical Specifications

Model Architecture and Objective

DS-GAN loads the released PosterLayout generator checkpoint. A four-channel ResNet-FPN image encoder initializes a bidirectional LSTM over an initial layout tensor shaped (B, 32, 2, 4). The generator predicts original-implementation class probabilities over no_object, text, logo, and underlay, plus normalized center xywh boxes.

Compute Infrastructure

Agreement-check commands are intended for one explicitly selected GPU when the original implementation requires CUDA.

Hardware

CPU is sufficient for import and smoke tests. CUDA is recommended for heavyweight agreement checks against the original implementation and released checkpoint.

Software

Use uv run --package ds-gan ... from the repository root so workspace dependency sources and extras resolve correctly.

License

Repository wrapper code is Apache-2.0. The upstream PosterLayout repository does not include a top-level redistribution license; converted weights and Hub model repositories are held until license and release terms are reviewed.

Citation

@inproceedings{hsu2023posterlayout,
  title = {PosterLayout: A New Benchmark and Approach for Content-Aware Visual-Textual Presentation Layout},
  author = {Hsu, HsiaoYuan and He, Xiangteng and Peng, Yuxin and Kong, Hao and Zhang, Qing},
  booktitle = {Proceedings of the IEEE/CVF Conference on Computer Vision and Pattern Recognition},
  year = {2023},
  pages = {6018--6026}
}

API Reference

Transformers-style DS-GAN components for PosterLayout generation.

DSGANConfig

Bases: PretrainedConfig

Store DS-GAN architecture and dataset metadata.

Parameters:

Name Type Description Default
dataset_name DatasetName | str

Dataset key. DS-GAN currently supports PKU PosterLayout.

pku_posterlayout
backbone str

Timm ResNet backbone name used by the reference checkpoint.

'resnet50'
max_elem int

Maximum number of generated elements.

32
in_channels int

CNN-LSTM input channels after flattening class and box planes.

8
out_channels int

CNN-LSTM convolution output channels.

32
hidden_size int | None

Bidirectional LSTM hidden size.

None
num_layers int

Number of LSTM layers.

4
output_size int

Combined class and box output width in the internal model.

8
image_size tuple[int, int] | list[int]

Processor/model input size as (height, width).

(350, 240)
reference_canvas_size tuple[int, int] | list[int]

Reference normalization canvas as (width, height).

(513, 750)
backbone_feature_size int

Flattened ResNet-FPN spatial size. The reference default is 22 * 15 = 330 for image_size=(350, 240).

330
model_num_classes int

Internal class channels including 0 = no object.

4
id2label Id2LabelMapping | None

Public zero-based semantic label mapping.

None
model_subfolder str

Pipeline subfolder for the model component.

'model'
processor_subfolder str

Pipeline subfolder for the processor component.

'processor'

Examples:

>>> DSGANConfig().max_elem
32
Source code in models/ds-gan/src/ds_gan/configuration_ds_gan.py
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
class DSGANConfig(PretrainedConfig):
    """Store DS-GAN architecture and dataset metadata.

    Args:
        dataset_name: Dataset key. DS-GAN currently supports PKU PosterLayout.
        backbone: Timm ResNet backbone name used by the reference checkpoint.
        max_elem: Maximum number of generated elements.
        in_channels: CNN-LSTM input channels after flattening class and box planes.
        out_channels: CNN-LSTM convolution output channels.
        hidden_size: Bidirectional LSTM hidden size.
        num_layers: Number of LSTM layers.
        output_size: Combined class and box output width in the internal model.
        image_size: Processor/model input size as ``(height, width)``.
        reference_canvas_size: Reference normalization canvas as ``(width, height)``.
        backbone_feature_size: Flattened ResNet-FPN spatial size. The reference
            default is ``22 * 15 = 330`` for ``image_size=(350, 240)``.
        model_num_classes: Internal class channels including ``0 = no object``.
        id2label: Public zero-based semantic label mapping.
        model_subfolder: Pipeline subfolder for the model component.
        processor_subfolder: Pipeline subfolder for the processor component.

    Examples:
        >>> DSGANConfig().max_elem
        32
    """

    model_type = "ds_gan"

    def __init__(
        self,
        dataset_name: DatasetName | str = DatasetName.pku_posterlayout,
        backbone: str = "resnet50",
        max_elem: int = 32,
        in_channels: int = 8,
        out_channels: int = 32,
        hidden_size: int | None = None,
        num_layers: int = 4,
        output_size: int = 8,
        image_size: tuple[int, int] | list[int] = (350, 240),
        reference_canvas_size: tuple[int, int] | list[int] = (513, 750),
        backbone_feature_size: int = 330,
        model_num_classes: int = 4,
        id2label: Id2LabelMapping | None = None,
        label2id: dict[str, int] | None = None,
        model_subfolder: str = "model",
        processor_subfolder: str = "processor",
        condition_types: list[str] | tuple[str, ...] | None = None,
        architectures: list[str] | None = None,
        model_type: str | None = None,
        transformers_version: str | None = None,
        torch_dtype: str | None = None,
        dtype: str | None = None,
        name_or_path: str = "",
        _commit_hash: str | None = None,
        **kwargs: str | int | float | bool | None,
    ) -> None:
        """Initialize DS-GAN configuration."""
        _ = (model_type, transformers_version)
        dataset = normalize_dataset_name(dataset_name)
        if dataset is not DatasetName.pku_posterlayout:
            raise ValueError(f"Unsupported DS-GAN dataset_name: {dataset_name}")

        public_id2label = {
            int(k): v for k, v in (id2label or _semantic_pku_id2label()).items()
        }
        public_label2id = label2id or {v: k for k, v in public_id2label.items()}
        super().__init__(
            id2label=public_id2label,
            label2id=public_label2id,
            architectures=architectures or ["DSGANModel"],
            torch_dtype=torch_dtype,  # ty: ignore[unknown-argument]
            dtype=dtype,
            name_or_path=name_or_path,  # ty: ignore[unknown-argument]
            _commit_hash=_commit_hash,  # ty: ignore[unknown-argument]
            **kwargs,  # ty: ignore[invalid-argument-type]
        )
        self.dataset_name = str(dataset)
        self.backbone = backbone
        self.max_elem = max_elem
        self.in_channels = in_channels
        self.out_channels = out_channels
        self.hidden_size = hidden_size if hidden_size is not None else max_elem * 8
        self.num_layers = num_layers
        self.output_size = output_size
        self.image_size = tuple(int(v) for v in image_size)
        self.reference_canvas_size = tuple(int(v) for v in reference_canvas_size)
        self.backbone_feature_size = backbone_feature_size
        self.model_num_classes = model_num_classes
        self.model_subfolder = model_subfolder
        self.processor_subfolder = processor_subfolder
        self.condition_types = list(condition_types or ["content_image"])

    @property
    def public_num_labels(self) -> int:
        """Return the number of public semantic labels."""
        return len(cast(dict[int, str], self.id2label))

public_num_labels property

public_num_labels: int

Return the number of public semantic labels.

__init__

__init__(
    dataset_name: DatasetName
    | str = DatasetName.pku_posterlayout,
    backbone: str = "resnet50",
    max_elem: int = 32,
    in_channels: int = 8,
    out_channels: int = 32,
    hidden_size: int | None = None,
    num_layers: int = 4,
    output_size: int = 8,
    image_size: tuple[int, int] | list[int] = (350, 240),
    reference_canvas_size: tuple[int, int] | list[int] = (
        513,
        750,
    ),
    backbone_feature_size: int = 330,
    model_num_classes: int = 4,
    id2label: Id2LabelMapping | None = None,
    label2id: dict[str, int] | None = None,
    model_subfolder: str = "model",
    processor_subfolder: str = "processor",
    condition_types: list[str]
    | tuple[str, ...]
    | None = None,
    architectures: list[str] | None = None,
    model_type: str | None = None,
    transformers_version: str | None = None,
    torch_dtype: str | None = None,
    dtype: str | None = None,
    name_or_path: str = "",
    _commit_hash: str | None = None,
    **kwargs: str | int | float | bool | None,
) -> None

Initialize DS-GAN configuration.

Source code in models/ds-gan/src/ds_gan/configuration_ds_gan.py
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
def __init__(
    self,
    dataset_name: DatasetName | str = DatasetName.pku_posterlayout,
    backbone: str = "resnet50",
    max_elem: int = 32,
    in_channels: int = 8,
    out_channels: int = 32,
    hidden_size: int | None = None,
    num_layers: int = 4,
    output_size: int = 8,
    image_size: tuple[int, int] | list[int] = (350, 240),
    reference_canvas_size: tuple[int, int] | list[int] = (513, 750),
    backbone_feature_size: int = 330,
    model_num_classes: int = 4,
    id2label: Id2LabelMapping | None = None,
    label2id: dict[str, int] | None = None,
    model_subfolder: str = "model",
    processor_subfolder: str = "processor",
    condition_types: list[str] | tuple[str, ...] | None = None,
    architectures: list[str] | None = None,
    model_type: str | None = None,
    transformers_version: str | None = None,
    torch_dtype: str | None = None,
    dtype: str | None = None,
    name_or_path: str = "",
    _commit_hash: str | None = None,
    **kwargs: str | int | float | bool | None,
) -> None:
    """Initialize DS-GAN configuration."""
    _ = (model_type, transformers_version)
    dataset = normalize_dataset_name(dataset_name)
    if dataset is not DatasetName.pku_posterlayout:
        raise ValueError(f"Unsupported DS-GAN dataset_name: {dataset_name}")

    public_id2label = {
        int(k): v for k, v in (id2label or _semantic_pku_id2label()).items()
    }
    public_label2id = label2id or {v: k for k, v in public_id2label.items()}
    super().__init__(
        id2label=public_id2label,
        label2id=public_label2id,
        architectures=architectures or ["DSGANModel"],
        torch_dtype=torch_dtype,  # ty: ignore[unknown-argument]
        dtype=dtype,
        name_or_path=name_or_path,  # ty: ignore[unknown-argument]
        _commit_hash=_commit_hash,  # ty: ignore[unknown-argument]
        **kwargs,  # ty: ignore[invalid-argument-type]
    )
    self.dataset_name = str(dataset)
    self.backbone = backbone
    self.max_elem = max_elem
    self.in_channels = in_channels
    self.out_channels = out_channels
    self.hidden_size = hidden_size if hidden_size is not None else max_elem * 8
    self.num_layers = num_layers
    self.output_size = output_size
    self.image_size = tuple(int(v) for v in image_size)
    self.reference_canvas_size = tuple(int(v) for v in reference_canvas_size)
    self.backbone_feature_size = backbone_feature_size
    self.model_num_classes = model_num_classes
    self.model_subfolder = model_subfolder
    self.processor_subfolder = processor_subfolder
    self.condition_types = list(condition_types or ["content_image"])

DSGANModel

Bases: PreTrainedModel

Transformers-compatible DS-GAN generator.

Parameters:

Name Type Description Default
config DSGANConfig

DS-GAN model configuration.

required

Examples:

>>> config = DSGANConfig(backbone="resnet18", max_elem=4, hidden_size=32, num_layers=2, image_size=(64, 64), backbone_feature_size=16)
>>> model = DSGANModel(config)
>>> model.config.model_type
'ds_gan'
Source code in models/ds-gan/src/ds_gan/modeling_ds_gan.py
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
class DSGANModel(PreTrainedModel):
    """Transformers-compatible DS-GAN generator.

    Args:
        config: DS-GAN model configuration.

    Examples:
        >>> config = DSGANConfig(backbone="resnet18", max_elem=4, hidden_size=32, num_layers=2, image_size=(64, 64), backbone_feature_size=16)
        >>> model = DSGANModel(config)
        >>> model.config.model_type
        'ds_gan'
    """

    config_class = DSGANConfig
    base_model_prefix = "ds_gan"
    supports_gradient_checkpointing = False

    def _init_weights(self, module: nn.Module) -> None:
        """Leave module-specific initialization unchanged during ``post_init``."""
        del module

    def __init__(
        self, config: DSGANConfig, *, backbone_weights: str | Path | None = None
    ) -> None:
        """Initialize DS-GAN generator layers."""
        super().__init__(config)
        self.resnet_fpn = ResnetBackbone(config, backbone_weights=backbone_weights)
        self.cnnlstm = CNNLSTM(config)
        self.fc1 = nn.Linear(2 * config.hidden_size, config.output_size // 2)
        self.fc2 = nn.Linear(2 * config.hidden_size, config.output_size // 2)
        self.post_init()

    def forward(
        self,
        pixel_values: Float[torch.Tensor, "batch 4 height width"],
        layout: Float[torch.Tensor, "batch elements 2 4"],
        return_dict: bool = True,
    ) -> (
        DSGANModelOutput
        | tuple[
            Float[torch.Tensor, "batch elements 4"],
            Float[torch.Tensor, "batch elements 4"],
        ]
    ):
        """Run a DS-GAN generator forward pass.

        Args:
            pixel_values: RGB plus saliency tensor shaped ``(B, 4, H, W)``.
            layout: Initial internal layout shaped ``(B, max_elem, 2, 4)``.
            return_dict: Whether to return a dataclass output.

        Returns:
            Raw class probabilities and normalized center ``xywh`` boxes.

        Raises:
            ValueError: If tensor shapes do not match the config.
        """
        if pixel_values.ndim != 4 or pixel_values.shape[1] != 4:
            raise ValueError("pixel_values must have shape (batch, 4, height, width)")

        expected_layout = (pixel_values.shape[0], self.config.max_elem, 2, 4)
        if tuple(layout.shape) != expected_layout:
            raise ValueError(f"layout must have shape {expected_layout}")

        pixel_values = pixel_values.to(dtype=self.dtype)
        layout = layout.to(device=pixel_values.device, dtype=self.dtype)
        h0 = self.resnet_fpn(pixel_values)
        lstm_output = self.cnnlstm(layout, h0)
        class_probs = torch.softmax(self.fc1(lstm_output), dim=-1)
        bbox = torch.sigmoid(self.fc2(lstm_output))
        if not return_dict:
            return class_probs, bbox
        return DSGANModelOutput(
            class_probs=class_probs,
            bbox=bbox,
            initial_layout=layout,
        )

__init__

__init__(
    config: DSGANConfig,
    *,
    backbone_weights: str | Path | None = None,
) -> None

Initialize DS-GAN generator layers.

Source code in models/ds-gan/src/ds_gan/modeling_ds_gan.py
171
172
173
174
175
176
177
178
179
180
def __init__(
    self, config: DSGANConfig, *, backbone_weights: str | Path | None = None
) -> None:
    """Initialize DS-GAN generator layers."""
    super().__init__(config)
    self.resnet_fpn = ResnetBackbone(config, backbone_weights=backbone_weights)
    self.cnnlstm = CNNLSTM(config)
    self.fc1 = nn.Linear(2 * config.hidden_size, config.output_size // 2)
    self.fc2 = nn.Linear(2 * config.hidden_size, config.output_size // 2)
    self.post_init()

forward

forward(
    pixel_values: Float[Tensor, "batch 4 height width"],
    layout: Float[Tensor, "batch elements 2 4"],
    return_dict: bool = True,
) -> (
    DSGANModelOutput
    | tuple[
        Float[torch.Tensor, "batch elements 4"],
        Float[torch.Tensor, "batch elements 4"],
    ]
)

Run a DS-GAN generator forward pass.

Parameters:

Name Type Description Default
pixel_values Float[Tensor, 'batch 4 height width']

RGB plus saliency tensor shaped (B, 4, H, W).

required
layout Float[Tensor, 'batch elements 2 4']

Initial internal layout shaped (B, max_elem, 2, 4).

required
return_dict bool

Whether to return a dataclass output.

True

Returns:

Type Description
DSGANModelOutput | tuple[Float[Tensor, 'batch elements 4'], Float[Tensor, 'batch elements 4']]

Raw class probabilities and normalized center xywh boxes.

Raises:

Type Description
ValueError

If tensor shapes do not match the config.

Source code in models/ds-gan/src/ds_gan/modeling_ds_gan.py
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
def forward(
    self,
    pixel_values: Float[torch.Tensor, "batch 4 height width"],
    layout: Float[torch.Tensor, "batch elements 2 4"],
    return_dict: bool = True,
) -> (
    DSGANModelOutput
    | tuple[
        Float[torch.Tensor, "batch elements 4"],
        Float[torch.Tensor, "batch elements 4"],
    ]
):
    """Run a DS-GAN generator forward pass.

    Args:
        pixel_values: RGB plus saliency tensor shaped ``(B, 4, H, W)``.
        layout: Initial internal layout shaped ``(B, max_elem, 2, 4)``.
        return_dict: Whether to return a dataclass output.

    Returns:
        Raw class probabilities and normalized center ``xywh`` boxes.

    Raises:
        ValueError: If tensor shapes do not match the config.
    """
    if pixel_values.ndim != 4 or pixel_values.shape[1] != 4:
        raise ValueError("pixel_values must have shape (batch, 4, height, width)")

    expected_layout = (pixel_values.shape[0], self.config.max_elem, 2, 4)
    if tuple(layout.shape) != expected_layout:
        raise ValueError(f"layout must have shape {expected_layout}")

    pixel_values = pixel_values.to(dtype=self.dtype)
    layout = layout.to(device=pixel_values.device, dtype=self.dtype)
    h0 = self.resnet_fpn(pixel_values)
    lstm_output = self.cnnlstm(layout, h0)
    class_probs = torch.softmax(self.fc1(lstm_output), dim=-1)
    bbox = torch.sigmoid(self.fc2(lstm_output))
    if not return_dict:
        return class_probs, bbox
    return DSGANModelOutput(
        class_probs=class_probs,
        bbox=bbox,
        initial_layout=layout,
    )

DSGANModelOutput dataclass

Bases: ModelOutput

Raw DS-GAN generator output.

Attributes:

Name Type Description
class_probs Float[Tensor, 'batch elements 4']

Internal class probabilities with shape (batch, elements, 4) where id 0 is no object.

bbox Float[Tensor, 'batch elements 4'] | None

Normalized center xywh boxes with shape (batch, elements, 4).

initial_layout Float[Tensor, 'batch elements 2 4'] | None

Initial class/box layout passed to the generator.

Source code in models/ds-gan/src/ds_gan/modeling_ds_gan.py
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
@dataclass
class DSGANModelOutput(ModelOutput):
    """Raw DS-GAN generator output.

    Attributes:
        class_probs: Internal class probabilities with shape
            ``(batch, elements, 4)`` where id 0 is ``no object``.
        bbox: Normalized center ``xywh`` boxes with shape
            ``(batch, elements, 4)``.
        initial_layout: Initial class/box layout passed to the generator.
    """

    class_probs: Float[torch.Tensor, "batch elements 4"]
    bbox: Float[torch.Tensor, "batch elements 4"] | None = None
    initial_layout: Float[torch.Tensor, "batch elements 2 4"] | None = None

DSGANPipeline

Bases: LayoutGenerationPipeline

Transformers-side pipeline for content-aware PosterLayout generation.

Parameters:

Name Type Description Default
model DSGANModel

DS-GAN generator.

required
processor DSGANProcessor | None

Optional processor for images and output decoding.

None
config DSGANConfig | None

Optional root pipeline config.

None
device str | device | None

Optional runtime device.

None

Examples:

>>> config = DSGANConfig(backbone="resnet18", max_elem=4, hidden_size=32, num_layers=2, image_size=(64, 64), backbone_feature_size=16)
>>> pipe = DSGANPipeline(DSGANModel(config))
>>> pipe.config.model_type
'ds_gan'
Source code in models/ds-gan/src/ds_gan/pipeline_ds_gan.py
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
class DSGANPipeline(LayoutGenerationPipeline):
    """Transformers-side pipeline for content-aware PosterLayout generation.

    Args:
        model: DS-GAN generator.
        processor: Optional processor for images and output decoding.
        config: Optional root pipeline config.
        device: Optional runtime device.

    Examples:
        >>> config = DSGANConfig(backbone="resnet18", max_elem=4, hidden_size=32, num_layers=2, image_size=(64, 64), backbone_feature_size=16)
        >>> pipe = DSGANPipeline(DSGANModel(config))
        >>> pipe.config.model_type
        'ds_gan'
    """

    config_class: ClassVar[type[PretrainedConfig]] = DSGANConfig
    component_specs: ClassVar[dict[str, PipelineComponentSpec]] = {
        "model": PipelineComponentSpec(
            attribute_name="model",
            loader=_load_model_component,
            marker_file="config.json",
            config_subfolder_attribute="model_subfolder",
        ),
        "processor": PipelineComponentSpec(
            attribute_name="processor",
            loader=_load_processor_component,
            marker_file="processor_config.json",
            save_with_is_main_process=False,
            config_subfolder_attribute="processor_subfolder",
        ),
    }

    config: DSGANConfig
    model: DSGANModel
    processor: DSGANProcessor

    def __init__(
        self,
        model: DSGANModel,
        processor: DSGANProcessor | None = None,
        config: DSGANConfig | None = None,
        device: str | torch.device | None = None,
    ) -> None:
        """Initialize DS-GAN pipeline."""
        super().__init__(config or model.config)
        self.config = config or model.config
        self.model = model
        self.processor = processor or DSGANProcessor(
            dataset_name=self.config.dataset_name,
            id2label=cast(dict[int | str, str], self.config.id2label),
            image_size=cast(tuple[int, int], self.config.image_size),
        )
        self.model.eval()
        if device is not None:
            self.to(device)

    @classmethod
    def _from_pretrained_components(
        cls,
        *,
        config: PretrainedConfig,
        components: Mapping[str, DSGANPipelineComponent | None],
    ) -> "DSGANPipeline":
        """Build a pipeline from loaded root components."""
        return cls(
            config=cast(DSGANConfig, config),
            model=cast(DSGANModel, components["model"]),
            processor=cast(DSGANProcessor, components["processor"]),
        )

    @torch.no_grad()
    def __call__(  # ty: ignore[invalid-method-override]
        self,
        images: ImageInput
        | list[ImageInput]
        | Float[torch.Tensor, "..."]
        | None = None,
        *,
        batch_size: int = 1,
        seed: int | None = None,
        generator: torch.Generator | None = None,
        condition_type: ConditionType | str = ConditionType.content_image,
        labels: Int[torch.Tensor, "batch elements"]
        | Sequence[ArrayLikeInput]
        | None = None,
        bbox: Float[torch.Tensor, "batch elements 4"]
        | Sequence[ArrayLikeInput]
        | None = None,
        mask: Bool[torch.Tensor, "batch elements"]
        | Sequence[ArrayLikeInput]
        | None = None,
        num_elements: int | list[int] | Int[torch.Tensor, "batch"] | None = None,
        box_format: BoxFormat | str = BoxFormat.xywh,
        normalized: bool = True,
        canvas_size: tuple[int, int] | None = None,
        num_inference_steps: int | None = None,
        output_type: OutputType | str = OutputType.dataclass,
        return_intermediates: bool = False,
        saliency: ImageInput
        | list[ImageInput]
        | Float[torch.Tensor, "..."]
        | None = None,
        saliency_pfpnet: ImageInput
        | list[ImageInput]
        | Float[torch.Tensor, "..."]
        | None = None,
        saliency_basnet: ImageInput
        | list[ImageInput]
        | Float[torch.Tensor, "..."]
        | None = None,
        pixel_values: Float[torch.Tensor, "batch 4 height width"] | None = None,
        initial_layout: Float[torch.Tensor, "batch elements 2 4"] | None = None,
    ) -> (
        LayoutGenerationOutput
        | dict[
            str,
            Shaped[torch.Tensor, "..."]
            | dict[int, str]
            | Mapping[str, Shaped[torch.Tensor, "..."] | ConditionType | str | bool]
            | None,
        ]
    ):
        """Generate layouts from content images and saliency maps.

        Args:
            images: RGB image or batch. Required unless ``pixel_values`` is given.
            batch_size: Batch size used when ``pixel_values`` is supplied.
            seed: Convenience seed. Ignored when ``generator`` is supplied.
            generator: Explicit torch generator.
            condition_type: Must normalize to ``content_image``.
            labels: Optional public labels used only with ``bbox`` to provide a
                fixed initial layout.
            bbox: Optional public boxes used with ``labels`` for a fixed layout.
            mask: Optional valid-element mask for fixed initial layouts.
            num_elements: Reserved compatibility argument.
            box_format: Format of optional ``bbox``.
            normalized: Whether optional ``bbox`` is normalized.
            canvas_size: Pixel canvas size for optional unnormalized ``bbox``.
            num_inference_steps: Reserved compatibility argument.
            output_type: ``dataclass`` or ``dict``.
            return_intermediates: Whether to include raw model tensors.
            saliency: Optional single merged saliency map.
            saliency_pfpnet: Optional PFPNet saliency map.
            saliency_basnet: Optional BASNet saliency map.
            pixel_values: Preprocessed ``(B, 4, H, W)`` tensor.
            initial_layout: Optional internal layout ``(B, max_elem, 2, 4)``.

        Returns:
            Shared layout-generation output.

        Raises:
            ValueError: If the condition, image inputs, or fixed layout are invalid.
        """
        del num_elements, num_inference_steps
        canonical = normalize_condition_type(condition_type)
        resolved_output_type = normalize_output_type(output_type)
        device = self.device or next(self.model.parameters()).device
        if pixel_values is None:
            if images is None:
                raise ValueError("images or pixel_values are required for DS-GAN")

            encoded = self.processor(
                images,
                saliency=saliency,
                saliency_pfpnet=saliency_pfpnet,
                saliency_basnet=saliency_basnet,
            )
            pixel_values = cast(torch.Tensor, encoded["pixel_values"])
        pixel_values = pixel_values.to(device=device, dtype=self.model.dtype)
        batch_size = pixel_values.shape[0] if pixel_values is not None else batch_size
        if initial_layout is None and bbox is not None and labels is not None:
            encoded_layout = self.processor.encode_layout(
                bbox=bbox,
                labels=labels,
                mask=mask,
                box_format=box_format,
                normalized=normalized,
                canvas_size=canvas_size,
                max_elem=self.config.max_elem,
            )
            initial_layout = encoded_layout["layout"]
        if initial_layout is None:
            prepared = self.prepare_generator(generator=generator, seed=seed)
            initial_layout = random_initial_layout(
                batch_size,
                self.config.max_elem,
                generator=prepared,
                device=device,
                dtype=self.model.dtype,
            )
        else:
            initial_layout = initial_layout.to(device=device, dtype=self.model.dtype)
        model_output = self.model(
            pixel_values=pixel_values,
            layout=initial_layout,
            return_dict=True,
        )
        assert isinstance(model_output, DSGANModelOutput)
        intermediates = None
        if return_intermediates:
            intermediates = {
                "condition_type": canonical,
                "initial_layout": initial_layout.detach().cpu(),
                "class_probs": model_output.class_probs.detach().cpu(),
            }
        return self.processor.decode(
            class_probs=model_output.class_probs,
            bbox=model_output.bbox,
            output_type=resolved_output_type.value,
            intermediates=intermediates,
        )

__init__

__init__(
    model: DSGANModel,
    processor: DSGANProcessor | None = None,
    config: DSGANConfig | None = None,
    device: str | device | None = None,
) -> None

Initialize DS-GAN pipeline.

Source code in models/ds-gan/src/ds_gan/pipeline_ds_gan.py
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
def __init__(
    self,
    model: DSGANModel,
    processor: DSGANProcessor | None = None,
    config: DSGANConfig | None = None,
    device: str | torch.device | None = None,
) -> None:
    """Initialize DS-GAN pipeline."""
    super().__init__(config or model.config)
    self.config = config or model.config
    self.model = model
    self.processor = processor or DSGANProcessor(
        dataset_name=self.config.dataset_name,
        id2label=cast(dict[int | str, str], self.config.id2label),
        image_size=cast(tuple[int, int], self.config.image_size),
    )
    self.model.eval()
    if device is not None:
        self.to(device)

__call__

__call__(
    images: ImageInput
    | list[ImageInput]
    | Float[Tensor, "..."]
    | None = None,
    *,
    batch_size: int = 1,
    seed: int | None = None,
    generator: Generator | None = None,
    condition_type: ConditionType
    | str = ConditionType.content_image,
    labels: Int[Tensor, "batch elements"]
    | Sequence[ArrayLikeInput]
    | None = None,
    bbox: Float[Tensor, "batch elements 4"]
    | Sequence[ArrayLikeInput]
    | None = None,
    mask: Bool[Tensor, "batch elements"]
    | Sequence[ArrayLikeInput]
    | None = None,
    num_elements: int
    | list[int]
    | Int[Tensor, "batch"]
    | None = None,
    box_format: BoxFormat | str = BoxFormat.xywh,
    normalized: bool = True,
    canvas_size: tuple[int, int] | None = None,
    num_inference_steps: int | None = None,
    output_type: OutputType | str = OutputType.dataclass,
    return_intermediates: bool = False,
    saliency: ImageInput
    | list[ImageInput]
    | Float[Tensor, "..."]
    | None = None,
    saliency_pfpnet: ImageInput
    | list[ImageInput]
    | Float[Tensor, "..."]
    | None = None,
    saliency_basnet: ImageInput
    | list[ImageInput]
    | Float[Tensor, "..."]
    | None = None,
    pixel_values: Float[Tensor, "batch 4 height width"]
    | None = None,
    initial_layout: Float[Tensor, "batch elements 2 4"]
    | None = None,
) -> (
    LayoutGenerationOutput
    | dict[
        str,
        Shaped[torch.Tensor, "..."]
        | dict[int, str]
        | Mapping[
            str,
            Shaped[torch.Tensor, "..."]
            | ConditionType
            | str
            | bool,
        ]
        | None,
    ]
)

Generate layouts from content images and saliency maps.

Parameters:

Name Type Description Default
images ImageInput | list[ImageInput] | Float[Tensor, '...'] | None

RGB image or batch. Required unless pixel_values is given.

None
batch_size int

Batch size used when pixel_values is supplied.

1
seed int | None

Convenience seed. Ignored when generator is supplied.

None
generator Generator | None

Explicit torch generator.

None
condition_type ConditionType | str

Must normalize to content_image.

content_image
labels Int[Tensor, 'batch elements'] | Sequence[ArrayLikeInput] | None

Optional public labels used only with bbox to provide a fixed initial layout.

None
bbox Float[Tensor, 'batch elements 4'] | Sequence[ArrayLikeInput] | None

Optional public boxes used with labels for a fixed layout.

None
mask Bool[Tensor, 'batch elements'] | Sequence[ArrayLikeInput] | None

Optional valid-element mask for fixed initial layouts.

None
num_elements int | list[int] | Int[Tensor, 'batch'] | None

Reserved compatibility argument.

None
box_format BoxFormat | str

Format of optional bbox.

xywh
normalized bool

Whether optional bbox is normalized.

True
canvas_size tuple[int, int] | None

Pixel canvas size for optional unnormalized bbox.

None
num_inference_steps int | None

Reserved compatibility argument.

None
output_type OutputType | str

dataclass or dict.

dataclass
return_intermediates bool

Whether to include raw model tensors.

False
saliency ImageInput | list[ImageInput] | Float[Tensor, '...'] | None

Optional single merged saliency map.

None
saliency_pfpnet ImageInput | list[ImageInput] | Float[Tensor, '...'] | None

Optional PFPNet saliency map.

None
saliency_basnet ImageInput | list[ImageInput] | Float[Tensor, '...'] | None

Optional BASNet saliency map.

None
pixel_values Float[Tensor, 'batch 4 height width'] | None

Preprocessed (B, 4, H, W) tensor.

None
initial_layout Float[Tensor, 'batch elements 2 4'] | None

Optional internal layout (B, max_elem, 2, 4).

None

Returns:

Type Description
LayoutGenerationOutput | dict[str, Shaped[Tensor, '...'] | dict[int, str] | Mapping[str, Shaped[Tensor, '...'] | ConditionType | str | bool] | None]

Shared layout-generation output.

Raises:

Type Description
ValueError

If the condition, image inputs, or fixed layout are invalid.

Source code in models/ds-gan/src/ds_gan/pipeline_ds_gan.py
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
@torch.no_grad()
def __call__(  # ty: ignore[invalid-method-override]
    self,
    images: ImageInput
    | list[ImageInput]
    | Float[torch.Tensor, "..."]
    | None = None,
    *,
    batch_size: int = 1,
    seed: int | None = None,
    generator: torch.Generator | None = None,
    condition_type: ConditionType | str = ConditionType.content_image,
    labels: Int[torch.Tensor, "batch elements"]
    | Sequence[ArrayLikeInput]
    | None = None,
    bbox: Float[torch.Tensor, "batch elements 4"]
    | Sequence[ArrayLikeInput]
    | None = None,
    mask: Bool[torch.Tensor, "batch elements"]
    | Sequence[ArrayLikeInput]
    | None = None,
    num_elements: int | list[int] | Int[torch.Tensor, "batch"] | None = None,
    box_format: BoxFormat | str = BoxFormat.xywh,
    normalized: bool = True,
    canvas_size: tuple[int, int] | None = None,
    num_inference_steps: int | None = None,
    output_type: OutputType | str = OutputType.dataclass,
    return_intermediates: bool = False,
    saliency: ImageInput
    | list[ImageInput]
    | Float[torch.Tensor, "..."]
    | None = None,
    saliency_pfpnet: ImageInput
    | list[ImageInput]
    | Float[torch.Tensor, "..."]
    | None = None,
    saliency_basnet: ImageInput
    | list[ImageInput]
    | Float[torch.Tensor, "..."]
    | None = None,
    pixel_values: Float[torch.Tensor, "batch 4 height width"] | None = None,
    initial_layout: Float[torch.Tensor, "batch elements 2 4"] | None = None,
) -> (
    LayoutGenerationOutput
    | dict[
        str,
        Shaped[torch.Tensor, "..."]
        | dict[int, str]
        | Mapping[str, Shaped[torch.Tensor, "..."] | ConditionType | str | bool]
        | None,
    ]
):
    """Generate layouts from content images and saliency maps.

    Args:
        images: RGB image or batch. Required unless ``pixel_values`` is given.
        batch_size: Batch size used when ``pixel_values`` is supplied.
        seed: Convenience seed. Ignored when ``generator`` is supplied.
        generator: Explicit torch generator.
        condition_type: Must normalize to ``content_image``.
        labels: Optional public labels used only with ``bbox`` to provide a
            fixed initial layout.
        bbox: Optional public boxes used with ``labels`` for a fixed layout.
        mask: Optional valid-element mask for fixed initial layouts.
        num_elements: Reserved compatibility argument.
        box_format: Format of optional ``bbox``.
        normalized: Whether optional ``bbox`` is normalized.
        canvas_size: Pixel canvas size for optional unnormalized ``bbox``.
        num_inference_steps: Reserved compatibility argument.
        output_type: ``dataclass`` or ``dict``.
        return_intermediates: Whether to include raw model tensors.
        saliency: Optional single merged saliency map.
        saliency_pfpnet: Optional PFPNet saliency map.
        saliency_basnet: Optional BASNet saliency map.
        pixel_values: Preprocessed ``(B, 4, H, W)`` tensor.
        initial_layout: Optional internal layout ``(B, max_elem, 2, 4)``.

    Returns:
        Shared layout-generation output.

    Raises:
        ValueError: If the condition, image inputs, or fixed layout are invalid.
    """
    del num_elements, num_inference_steps
    canonical = normalize_condition_type(condition_type)
    resolved_output_type = normalize_output_type(output_type)
    device = self.device or next(self.model.parameters()).device
    if pixel_values is None:
        if images is None:
            raise ValueError("images or pixel_values are required for DS-GAN")

        encoded = self.processor(
            images,
            saliency=saliency,
            saliency_pfpnet=saliency_pfpnet,
            saliency_basnet=saliency_basnet,
        )
        pixel_values = cast(torch.Tensor, encoded["pixel_values"])
    pixel_values = pixel_values.to(device=device, dtype=self.model.dtype)
    batch_size = pixel_values.shape[0] if pixel_values is not None else batch_size
    if initial_layout is None and bbox is not None and labels is not None:
        encoded_layout = self.processor.encode_layout(
            bbox=bbox,
            labels=labels,
            mask=mask,
            box_format=box_format,
            normalized=normalized,
            canvas_size=canvas_size,
            max_elem=self.config.max_elem,
        )
        initial_layout = encoded_layout["layout"]
    if initial_layout is None:
        prepared = self.prepare_generator(generator=generator, seed=seed)
        initial_layout = random_initial_layout(
            batch_size,
            self.config.max_elem,
            generator=prepared,
            device=device,
            dtype=self.model.dtype,
        )
    else:
        initial_layout = initial_layout.to(device=device, dtype=self.model.dtype)
    model_output = self.model(
        pixel_values=pixel_values,
        layout=initial_layout,
        return_dict=True,
    )
    assert isinstance(model_output, DSGANModelOutput)
    intermediates = None
    if return_intermediates:
        intermediates = {
            "condition_type": canonical,
            "initial_layout": initial_layout.detach().cpu(),
            "class_probs": model_output.class_probs.detach().cpu(),
        }
    return self.processor.decode(
        class_probs=model_output.class_probs,
        bbox=model_output.bbox,
        output_type=resolved_output_type.value,
        intermediates=intermediates,
    )

OutputType

Bases: StrEnum

Supported DS-GAN pipeline output containers.

Source code in models/ds-gan/src/ds_gan/pipeline_ds_gan.py
36
37
38
39
40
class OutputType(StrEnum):
    """Supported DS-GAN pipeline output containers."""

    dataclass = auto()
    dict = auto()

DSGANProcessor

Bases: ProcessorMixin

Prepare PosterLayout RGB/saliency inputs and decode DS-GAN outputs.

Parameters:

Name Type Description Default
dataset_name DatasetName | str

Dataset key. Only PKU PosterLayout is supported.

pku_posterlayout
id2label dict[int | str, str] | None

Public semantic labels excluding model no object.

None
image_size tuple[int, int] | list[int]

Resize target as (height, width).

(350, 240)

Examples:

>>> processor = DSGANProcessor()
>>> processor.id2label[0]
'text'
Source code in models/ds-gan/src/ds_gan/processing_ds_gan.py
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
class DSGANProcessor(ProcessorMixin):
    """Prepare PosterLayout RGB/saliency inputs and decode DS-GAN outputs.

    Args:
        dataset_name: Dataset key. Only PKU PosterLayout is supported.
        id2label: Public semantic labels excluding model ``no object``.
        image_size: Resize target as ``(height, width)``.

    Examples:
        >>> processor = DSGANProcessor()
        >>> processor.id2label[0]
        'text'
    """

    config_name = "processor_config.json"

    def __init__(
        self,
        dataset_name: DatasetName | str = DatasetName.pku_posterlayout,
        id2label: dict[int | str, str] | None = None,
        image_size: tuple[int, int] | list[int] = (350, 240),
    ) -> None:
        """Initialize processor metadata."""
        self.chat_template = None
        dataset = normalize_dataset_name(dataset_name)
        if dataset is not DatasetName.pku_posterlayout:
            raise ValueError(f"Unsupported DS-GAN dataset_name: {dataset_name}")

        self.dataset_name = str(dataset)
        default_id2label = {0: "text", 1: "logo", 2: "underlay"}
        raw_id2label = id2label or default_id2label
        self.id2label = {int(k): v for k, v in raw_id2label.items()}
        self.label2id = {v: k for k, v in self.id2label.items()}
        height, width = image_size
        self.image_size: tuple[int, int] = (int(height), int(width))

    def __call__(
        self,
        images: DSGANImageInput | Sequence[DSGANImageInput],
        *,
        saliency: DSGANImageInput | Sequence[DSGANImageInput] | None = None,
        saliency_pfpnet: DSGANImageInput | Sequence[DSGANImageInput] | None = None,
        saliency_basnet: DSGANImageInput | Sequence[DSGANImageInput] | None = None,
        return_tensors: Literal["pt"] = "pt",
    ) -> BatchEncoding:
        """Encode content images into ``pixel_values``.

        Args:
            images: RGB image or batch of RGB images.
            saliency: Optional saliency image or batch. If omitted and both
                saliency maps are given, the maps are merged by pixelwise max.
            saliency_pfpnet: Optional PFPNet saliency map.
            saliency_basnet: Optional BASNet saliency map.
            return_tensors: Tensor framework. Only ``pt`` is supported.

        Returns:
            Batch encoding containing ``pixel_values`` shaped ``(B, 4, H, W)``.
        """
        if return_tensors != "pt":
            raise ValueError("DSGANProcessor only supports return_tensors='pt'")

        image_rows = _ensure_batch(images)
        saliency_rows = self._resolve_saliency(
            len(image_rows),
            saliency=saliency,
            saliency_pfpnet=saliency_pfpnet,
            saliency_basnet=saliency_basnet,
        )
        tensors = []
        for image, sal in zip(image_rows, saliency_rows, strict=True):
            rgb = _to_rgb_tensor(image, self.image_size)
            sal_tensor = (
                torch.zeros(1, *self.image_size, dtype=torch.float32)
                if sal is None
                else _to_l_tensor(sal, self.image_size)
            )
            tensors.append(torch.cat((rgb, sal_tensor), dim=0))
        return BatchEncoding({"pixel_values": torch.stack(tensors)})

    def decode(
        self,
        *,
        class_probs: Float[torch.Tensor, "batch elements 4"],
        bbox: Float[torch.Tensor, "batch elements 4"],
        output_type: Literal["dataclass", "dict"] = "dataclass",
        scores: Float[torch.Tensor, "batch elements"] | None = None,
        intermediates: Mapping[str, Shaped[torch.Tensor, "..."] | str | bool]
        | None = None,
    ) -> (
        LayoutGenerationOutput
        | dict[
            str,
            Shaped[torch.Tensor, "..."]
            | dict[int, str]
            | Mapping[str, Shaped[torch.Tensor, "..."] | str | bool]
            | None,
        ]
    ):
        """Decode raw DS-GAN class probabilities and boxes.

        Args:
            class_probs: Internal class probabilities shaped ``(B, E, 4)``.
            bbox: Normalized center ``xywh`` boxes shaped ``(B, E, 4)``.
            output_type: Return format.
            scores: Optional per-element class scores.
            intermediates: Optional model-specific intermediate tensors.

        Returns:
            Shared layout output with public labels and mask semantics.
        """
        class_ids = torch.argmax(class_probs, dim=-1)
        mask = class_ids != 0
        public_labels = (class_ids - 1).clamp_min(0).long()
        resolved_scores = scores
        if resolved_scores is None:
            resolved_scores = class_probs.max(dim=-1).values
        output = LayoutGenerationOutput(
            bbox=bbox.detach().cpu().clamp(0.0, 1.0),
            labels=public_labels.detach().cpu(),
            mask=mask.detach().cpu(),
            id2label=dict(self.id2label),
            scores=resolved_scores.detach().cpu(),
            intermediates=intermediates,
        )
        if output_type == "dict":
            return dict(output)
        if output_type == "dataclass":
            return output
        raise ValueError(f"Unsupported output_type: {output_type}")

    def encode_layout(
        self,
        *,
        bbox: Float[torch.Tensor, "batch elements 4"]
        | Float[np.ndarray, "batch elements 4"]
        | Sequence[ArrayLikeInput],
        labels: Int[torch.Tensor, "batch elements"]
        | Int[np.ndarray, "batch elements"]
        | Sequence[ArrayLikeInput],
        mask: Bool[torch.Tensor, "batch elements"]
        | Bool[np.ndarray, "batch elements"]
        | Sequence[ArrayLikeInput]
        | None = None,
        box_format: BoxFormat | str = BoxFormat.xywh,
        normalized: bool = True,
        canvas_size: tuple[int, int] | None = None,
        max_elem: int = 32,
    ) -> dict[str, Shaped[torch.Tensor, "..."]]:
        """Encode public boxes/labels into the internal layout tensor.

        Args:
            bbox: Public boxes.
            labels: Public zero-based semantic labels.
            mask: Optional valid-element mask.
            box_format: Input box format.
            normalized: Whether the boxes are normalized.
            canvas_size: Pixel canvas size used when ``normalized=False``.
            max_elem: Output slot count.

        Returns:
            Dictionary with internal ``layout``, normalized ``bbox``, labels, and mask.
        """
        bbox_t, labels_t, mask_t = prepare_layout_tensors(
            bbox=bbox,
            labels=labels,
            mask=mask,
            box_format=box_format,
            normalized=normalized,
            canvas_size=canvas_size,
            clamp_converted_normalized=True,
        )
        bbox_t, labels_t, mask_t = self.pad(bbox_t, labels_t, mask_t, max_elem=max_elem)
        model_labels = torch.zeros_like(labels_t)
        model_labels[mask_t] = labels_t[mask_t] + 1
        class_one_hot = torch.nn.functional.one_hot(model_labels, num_classes=4).to(
            dtype=bbox_t.dtype
        )
        layout = torch.stack((class_one_hot, bbox_t), dim=2)
        return {"layout": layout, "bbox": bbox_t, "labels": labels_t, "mask": mask_t}

    def pad(
        self,
        bbox: Float[torch.Tensor, "batch elements 4"],
        labels: Int[torch.Tensor, "batch elements"],
        mask: Bool[torch.Tensor, "batch elements"],
        *,
        max_elem: int,
    ) -> tuple[
        Float[torch.Tensor, "batch padded_elements 4"],
        Int[torch.Tensor, "batch padded_elements"],
        Bool[torch.Tensor, "batch padded_elements"],
    ]:
        """Pad layout tensors to DS-GAN ``max_elem`` slots."""
        if bbox.shape[1] > max_elem:
            raise ValueError(f"DS-GAN supports at most {max_elem} elements")

        pad_count = max_elem - bbox.shape[1]
        if pad_count:
            bbox = torch.cat((bbox, torch.zeros(bbox.shape[0], pad_count, 4)), dim=1)
            labels = torch.cat(
                (labels, torch.zeros(labels.shape[0], pad_count, dtype=torch.long)),
                dim=1,
            )
            mask = torch.cat(
                (mask, torch.zeros(mask.shape[0], pad_count, dtype=torch.bool)),
                dim=1,
            )
        labels = labels.clone()
        labels[~mask] = 0
        return bbox, labels, mask

    def _resolve_saliency(
        self,
        batch_size: int,
        *,
        saliency: DSGANImageInput | Sequence[DSGANImageInput] | None,
        saliency_pfpnet: DSGANImageInput | Sequence[DSGANImageInput] | None,
        saliency_basnet: DSGANImageInput | Sequence[DSGANImageInput] | None,
    ) -> list[DSGANImageInput | Float[torch.Tensor, "1 height width"] | None]:
        if saliency is not None:
            rows = _ensure_batch(saliency)
            if len(rows) != batch_size:
                raise ValueError("saliency batch size must match images")

            return cast(
                list[DSGANImageInput | Float[torch.Tensor, "1 height width"] | None],
                rows,
            )
        if saliency_pfpnet is None and saliency_basnet is None:
            return [None] * batch_size
        first = (
            _ensure_batch(saliency_pfpnet)
            if saliency_pfpnet is not None
            else [None] * batch_size
        )
        second = (
            _ensure_batch(saliency_basnet)
            if saliency_basnet is not None
            else [None] * batch_size
        )
        if len(first) != batch_size or len(second) != batch_size:
            raise ValueError("saliency batch size must match images")

        merged: list[Float[torch.Tensor, "1 height width"]] = []
        for left, right in zip(first, second, strict=True):
            if left is None:
                merged.append(_to_l_tensor(right, self.image_size))
            elif right is None:
                merged.append(_to_l_tensor(left, self.image_size))
            else:
                merged.append(_merge_saliency_native(left, right, self.image_size))
        return cast(
            list[DSGANImageInput | Float[torch.Tensor, "1 height width"] | None], merged
        )

__init__

__init__(
    dataset_name: DatasetName
    | str = DatasetName.pku_posterlayout,
    id2label: dict[int | str, str] | None = None,
    image_size: tuple[int, int] | list[int] = (350, 240),
) -> None

Initialize processor metadata.

Source code in models/ds-gan/src/ds_gan/processing_ds_gan.py
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
def __init__(
    self,
    dataset_name: DatasetName | str = DatasetName.pku_posterlayout,
    id2label: dict[int | str, str] | None = None,
    image_size: tuple[int, int] | list[int] = (350, 240),
) -> None:
    """Initialize processor metadata."""
    self.chat_template = None
    dataset = normalize_dataset_name(dataset_name)
    if dataset is not DatasetName.pku_posterlayout:
        raise ValueError(f"Unsupported DS-GAN dataset_name: {dataset_name}")

    self.dataset_name = str(dataset)
    default_id2label = {0: "text", 1: "logo", 2: "underlay"}
    raw_id2label = id2label or default_id2label
    self.id2label = {int(k): v for k, v in raw_id2label.items()}
    self.label2id = {v: k for k, v in self.id2label.items()}
    height, width = image_size
    self.image_size: tuple[int, int] = (int(height), int(width))

__call__

__call__(
    images: DSGANImageInput | Sequence[DSGANImageInput],
    *,
    saliency: DSGANImageInput
    | Sequence[DSGANImageInput]
    | None = None,
    saliency_pfpnet: DSGANImageInput
    | Sequence[DSGANImageInput]
    | None = None,
    saliency_basnet: DSGANImageInput
    | Sequence[DSGANImageInput]
    | None = None,
    return_tensors: Literal["pt"] = "pt",
) -> BatchEncoding

Encode content images into pixel_values.

Parameters:

Name Type Description Default
images DSGANImageInput | Sequence[DSGANImageInput]

RGB image or batch of RGB images.

required
saliency DSGANImageInput | Sequence[DSGANImageInput] | None

Optional saliency image or batch. If omitted and both saliency maps are given, the maps are merged by pixelwise max.

None
saliency_pfpnet DSGANImageInput | Sequence[DSGANImageInput] | None

Optional PFPNet saliency map.

None
saliency_basnet DSGANImageInput | Sequence[DSGANImageInput] | None

Optional BASNet saliency map.

None
return_tensors Literal['pt']

Tensor framework. Only pt is supported.

'pt'

Returns:

Type Description
BatchEncoding

Batch encoding containing pixel_values shaped (B, 4, H, W).

Source code in models/ds-gan/src/ds_gan/processing_ds_gan.py
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
def __call__(
    self,
    images: DSGANImageInput | Sequence[DSGANImageInput],
    *,
    saliency: DSGANImageInput | Sequence[DSGANImageInput] | None = None,
    saliency_pfpnet: DSGANImageInput | Sequence[DSGANImageInput] | None = None,
    saliency_basnet: DSGANImageInput | Sequence[DSGANImageInput] | None = None,
    return_tensors: Literal["pt"] = "pt",
) -> BatchEncoding:
    """Encode content images into ``pixel_values``.

    Args:
        images: RGB image or batch of RGB images.
        saliency: Optional saliency image or batch. If omitted and both
            saliency maps are given, the maps are merged by pixelwise max.
        saliency_pfpnet: Optional PFPNet saliency map.
        saliency_basnet: Optional BASNet saliency map.
        return_tensors: Tensor framework. Only ``pt`` is supported.

    Returns:
        Batch encoding containing ``pixel_values`` shaped ``(B, 4, H, W)``.
    """
    if return_tensors != "pt":
        raise ValueError("DSGANProcessor only supports return_tensors='pt'")

    image_rows = _ensure_batch(images)
    saliency_rows = self._resolve_saliency(
        len(image_rows),
        saliency=saliency,
        saliency_pfpnet=saliency_pfpnet,
        saliency_basnet=saliency_basnet,
    )
    tensors = []
    for image, sal in zip(image_rows, saliency_rows, strict=True):
        rgb = _to_rgb_tensor(image, self.image_size)
        sal_tensor = (
            torch.zeros(1, *self.image_size, dtype=torch.float32)
            if sal is None
            else _to_l_tensor(sal, self.image_size)
        )
        tensors.append(torch.cat((rgb, sal_tensor), dim=0))
    return BatchEncoding({"pixel_values": torch.stack(tensors)})

decode

decode(
    *,
    class_probs: Float[Tensor, "batch elements 4"],
    bbox: Float[Tensor, "batch elements 4"],
    output_type: Literal["dataclass", "dict"] = "dataclass",
    scores: Float[Tensor, "batch elements"] | None = None,
    intermediates: Mapping[
        str, Shaped[Tensor, "..."] | str | bool
    ]
    | None = None,
) -> (
    LayoutGenerationOutput
    | dict[
        str,
        Shaped[torch.Tensor, "..."]
        | dict[int, str]
        | Mapping[
            str, Shaped[torch.Tensor, "..."] | str | bool
        ]
        | None,
    ]
)

Decode raw DS-GAN class probabilities and boxes.

Parameters:

Name Type Description Default
class_probs Float[Tensor, 'batch elements 4']

Internal class probabilities shaped (B, E, 4).

required
bbox Float[Tensor, 'batch elements 4']

Normalized center xywh boxes shaped (B, E, 4).

required
output_type Literal['dataclass', 'dict']

Return format.

'dataclass'
scores Float[Tensor, 'batch elements'] | None

Optional per-element class scores.

None
intermediates Mapping[str, Shaped[Tensor, '...'] | str | bool] | None

Optional model-specific intermediate tensors.

None

Returns:

Type Description
LayoutGenerationOutput | dict[str, Shaped[Tensor, '...'] | dict[int, str] | Mapping[str, Shaped[Tensor, '...'] | str | bool] | None]

Shared layout output with public labels and mask semantics.

Source code in models/ds-gan/src/ds_gan/processing_ds_gan.py
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
def decode(
    self,
    *,
    class_probs: Float[torch.Tensor, "batch elements 4"],
    bbox: Float[torch.Tensor, "batch elements 4"],
    output_type: Literal["dataclass", "dict"] = "dataclass",
    scores: Float[torch.Tensor, "batch elements"] | None = None,
    intermediates: Mapping[str, Shaped[torch.Tensor, "..."] | str | bool]
    | None = None,
) -> (
    LayoutGenerationOutput
    | dict[
        str,
        Shaped[torch.Tensor, "..."]
        | dict[int, str]
        | Mapping[str, Shaped[torch.Tensor, "..."] | str | bool]
        | None,
    ]
):
    """Decode raw DS-GAN class probabilities and boxes.

    Args:
        class_probs: Internal class probabilities shaped ``(B, E, 4)``.
        bbox: Normalized center ``xywh`` boxes shaped ``(B, E, 4)``.
        output_type: Return format.
        scores: Optional per-element class scores.
        intermediates: Optional model-specific intermediate tensors.

    Returns:
        Shared layout output with public labels and mask semantics.
    """
    class_ids = torch.argmax(class_probs, dim=-1)
    mask = class_ids != 0
    public_labels = (class_ids - 1).clamp_min(0).long()
    resolved_scores = scores
    if resolved_scores is None:
        resolved_scores = class_probs.max(dim=-1).values
    output = LayoutGenerationOutput(
        bbox=bbox.detach().cpu().clamp(0.0, 1.0),
        labels=public_labels.detach().cpu(),
        mask=mask.detach().cpu(),
        id2label=dict(self.id2label),
        scores=resolved_scores.detach().cpu(),
        intermediates=intermediates,
    )
    if output_type == "dict":
        return dict(output)
    if output_type == "dataclass":
        return output
    raise ValueError(f"Unsupported output_type: {output_type}")

encode_layout

encode_layout(
    *,
    bbox: Float[Tensor, "batch elements 4"]
    | Float[ndarray, "batch elements 4"]
    | Sequence[ArrayLikeInput],
    labels: Int[Tensor, "batch elements"]
    | Int[ndarray, "batch elements"]
    | Sequence[ArrayLikeInput],
    mask: Bool[Tensor, "batch elements"]
    | Bool[ndarray, "batch elements"]
    | Sequence[ArrayLikeInput]
    | None = None,
    box_format: BoxFormat | str = BoxFormat.xywh,
    normalized: bool = True,
    canvas_size: tuple[int, int] | None = None,
    max_elem: int = 32,
) -> dict[str, Shaped[torch.Tensor, "..."]]

Encode public boxes/labels into the internal layout tensor.

Parameters:

Name Type Description Default
bbox Float[Tensor, 'batch elements 4'] | Float[ndarray, 'batch elements 4'] | Sequence[ArrayLikeInput]

Public boxes.

required
labels Int[Tensor, 'batch elements'] | Int[ndarray, 'batch elements'] | Sequence[ArrayLikeInput]

Public zero-based semantic labels.

required
mask Bool[Tensor, 'batch elements'] | Bool[ndarray, 'batch elements'] | Sequence[ArrayLikeInput] | None

Optional valid-element mask.

None
box_format BoxFormat | str

Input box format.

xywh
normalized bool

Whether the boxes are normalized.

True
canvas_size tuple[int, int] | None

Pixel canvas size used when normalized=False.

None
max_elem int

Output slot count.

32

Returns:

Type Description
dict[str, Shaped[Tensor, '...']]

Dictionary with internal layout, normalized bbox, labels, and mask.

Source code in models/ds-gan/src/ds_gan/processing_ds_gan.py
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
def encode_layout(
    self,
    *,
    bbox: Float[torch.Tensor, "batch elements 4"]
    | Float[np.ndarray, "batch elements 4"]
    | Sequence[ArrayLikeInput],
    labels: Int[torch.Tensor, "batch elements"]
    | Int[np.ndarray, "batch elements"]
    | Sequence[ArrayLikeInput],
    mask: Bool[torch.Tensor, "batch elements"]
    | Bool[np.ndarray, "batch elements"]
    | Sequence[ArrayLikeInput]
    | None = None,
    box_format: BoxFormat | str = BoxFormat.xywh,
    normalized: bool = True,
    canvas_size: tuple[int, int] | None = None,
    max_elem: int = 32,
) -> dict[str, Shaped[torch.Tensor, "..."]]:
    """Encode public boxes/labels into the internal layout tensor.

    Args:
        bbox: Public boxes.
        labels: Public zero-based semantic labels.
        mask: Optional valid-element mask.
        box_format: Input box format.
        normalized: Whether the boxes are normalized.
        canvas_size: Pixel canvas size used when ``normalized=False``.
        max_elem: Output slot count.

    Returns:
        Dictionary with internal ``layout``, normalized ``bbox``, labels, and mask.
    """
    bbox_t, labels_t, mask_t = prepare_layout_tensors(
        bbox=bbox,
        labels=labels,
        mask=mask,
        box_format=box_format,
        normalized=normalized,
        canvas_size=canvas_size,
        clamp_converted_normalized=True,
    )
    bbox_t, labels_t, mask_t = self.pad(bbox_t, labels_t, mask_t, max_elem=max_elem)
    model_labels = torch.zeros_like(labels_t)
    model_labels[mask_t] = labels_t[mask_t] + 1
    class_one_hot = torch.nn.functional.one_hot(model_labels, num_classes=4).to(
        dtype=bbox_t.dtype
    )
    layout = torch.stack((class_one_hot, bbox_t), dim=2)
    return {"layout": layout, "bbox": bbox_t, "labels": labels_t, "mask": mask_t}

pad

pad(
    bbox: Float[Tensor, "batch elements 4"],
    labels: Int[Tensor, "batch elements"],
    mask: Bool[Tensor, "batch elements"],
    *,
    max_elem: int,
) -> tuple[
    Float[torch.Tensor, "batch padded_elements 4"],
    Int[torch.Tensor, "batch padded_elements"],
    Bool[torch.Tensor, "batch padded_elements"],
]

Pad layout tensors to DS-GAN max_elem slots.

Source code in models/ds-gan/src/ds_gan/processing_ds_gan.py
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
def pad(
    self,
    bbox: Float[torch.Tensor, "batch elements 4"],
    labels: Int[torch.Tensor, "batch elements"],
    mask: Bool[torch.Tensor, "batch elements"],
    *,
    max_elem: int,
) -> tuple[
    Float[torch.Tensor, "batch padded_elements 4"],
    Int[torch.Tensor, "batch padded_elements"],
    Bool[torch.Tensor, "batch padded_elements"],
]:
    """Pad layout tensors to DS-GAN ``max_elem`` slots."""
    if bbox.shape[1] > max_elem:
        raise ValueError(f"DS-GAN supports at most {max_elem} elements")

    pad_count = max_elem - bbox.shape[1]
    if pad_count:
        bbox = torch.cat((bbox, torch.zeros(bbox.shape[0], pad_count, 4)), dim=1)
        labels = torch.cat(
            (labels, torch.zeros(labels.shape[0], pad_count, dtype=torch.long)),
            dim=1,
        )
        mask = torch.cat(
            (mask, torch.zeros(mask.shape[0], pad_count, dtype=torch.bool)),
            dim=1,
        )
    labels = labels.clone()
    labels[~mask] = 0
    return bbox, labels, mask

default_ds_gan_config

default_ds_gan_config() -> DSGANConfig

Return the reference-compatible DS-GAN default configuration.

Examples:

>>> default_ds_gan_config().backbone
'resnet50'
Source code in models/ds-gan/src/ds_gan/configuration_ds_gan.py
125
126
127
128
129
130
131
132
def default_ds_gan_config() -> DSGANConfig:
    """Return the reference-compatible DS-GAN default configuration.

    Examples:
        >>> default_ds_gan_config().backbone
        'resnet50'
    """
    return DSGANConfig()

convert_vendor_state_dict

convert_vendor_state_dict(
    state_dict: Mapping[str, Shaped[Tensor, "..."]],
) -> dict[str, Shaped[torch.Tensor, "..."]]

Convert vendor DS-GAN generator keys to DSGANModel keys.

The released checkpoint was commonly saved from torch.nn.DataParallel; this helper strips the leading module. prefix and keeps all generator module names otherwise unchanged.

Parameters:

Name Type Description Default
state_dict Mapping[str, Shaped[Tensor, '...']]

Original checkpoint mapping.

required

Returns:

Type Description
dict[str, Shaped[Tensor, '...']]

Converted state dictionary.

Examples:

>>> convert_vendor_state_dict({"module.fc1.weight": torch.zeros(1)})["fc1.weight"].shape
torch.Size([1])
Source code in models/ds-gan/src/ds_gan/conversion.py
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
def convert_vendor_state_dict(
    state_dict: Mapping[str, Shaped[torch.Tensor, "..."]],
) -> dict[str, Shaped[torch.Tensor, "..."]]:
    """Convert vendor DS-GAN generator keys to ``DSGANModel`` keys.

    The released checkpoint was commonly saved from ``torch.nn.DataParallel``;
    this helper strips the leading ``module.`` prefix and keeps all generator
    module names otherwise unchanged.

    Args:
        state_dict: Original checkpoint mapping.

    Returns:
        Converted state dictionary.

    Examples:
        >>> convert_vendor_state_dict({"module.fc1.weight": torch.zeros(1)})["fc1.weight"].shape
        torch.Size([1])
    """
    converted: dict[str, Shaped[torch.Tensor, "..."]] = {}
    for key, value in state_dict.items():
        name = key.removeprefix("module.")
        name = name.removeprefix("generator.")
        converted[name] = value
    return converted

random_initial_layout

random_initial_layout(
    batch_size: int,
    max_elem: int,
    *,
    generator: Generator | None = None,
    seed: int | None = None,
    device: device | str | None = None,
    dtype: dtype = torch.float32,
    weighted_classes: bool = True,
    use_numpy_classes: bool = False,
) -> Float[torch.Tensor, "batch elements 2 4"]

Sample the DS-GAN initial layout tensor.

Parameters:

Name Type Description Default
batch_size int

Batch size.

required
max_elem int

Number of layout slots.

required
generator Generator | None

Optional torch generator. Takes precedence over seed.

None
seed int | None

Convenience seed used only when generator is absent.

None
device device | str | None

Target torch device.

None
dtype dtype

Target floating dtype.

float32
weighted_classes bool

Whether to use the released inference class prior.

True
use_numpy_classes bool

Use NumPy's legacy RandomState class sampler to mirror the reference script's weighted class prior when seed is supplied. Torch box sampling still follows generator or seed.

False

Returns:

Type Description
Float[Tensor, 'batch elements 2 4']

Tensor shaped (batch, max_elem, 2, 4).

Source code in models/ds-gan/src/ds_gan/modeling_ds_gan.py
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
def random_initial_layout(
    batch_size: int,
    max_elem: int,
    *,
    generator: torch.Generator | None = None,
    seed: int | None = None,
    device: torch.device | str | None = None,
    dtype: torch.dtype = torch.float32,
    weighted_classes: bool = True,
    use_numpy_classes: bool = False,
) -> Float[torch.Tensor, "batch elements 2 4"]:
    """Sample the DS-GAN initial layout tensor.

    Args:
        batch_size: Batch size.
        max_elem: Number of layout slots.
        generator: Optional torch generator. Takes precedence over ``seed``.
        seed: Convenience seed used only when ``generator`` is absent.
        device: Target torch device.
        dtype: Target floating dtype.
        weighted_classes: Whether to use the released inference class prior.
        use_numpy_classes: Use NumPy's legacy ``RandomState`` class sampler to
            mirror the reference script's weighted class prior when ``seed`` is
            supplied. Torch box sampling still follows ``generator`` or ``seed``.

    Returns:
        Tensor shaped ``(batch, max_elem, 2, 4)``.
    """
    resolved_device = (
        torch.device(device) if device is not None else torch.device("cpu")
    )
    generator = resolve_torch_generator(generator=generator, seed=seed)
    if weighted_classes:
        probs = torch.tensor((0.1, 0.8, 1.0, 1.0), device=resolved_device)
        probs = probs / probs.sum()
    else:
        probs = torch.full((4,), 0.25, device=resolved_device)
    if use_numpy_classes:
        rng = np.random.RandomState(seed)
        np_probs = probs.detach().cpu().numpy()
        class_ids = torch.as_tensor(
            rng.choice(4, size=(batch_size, max_elem, 1), p=np_probs),
            dtype=torch.long,
            device=resolved_device,
        )
    else:
        class_ids = multinomial(
            probs,
            num_samples=batch_size * max_elem,
            replacement=True,
            generator=generator,
            device=resolved_device,
        ).reshape(batch_size, max_elem, 1)
    class_one_hot = torch.zeros(
        batch_size,
        max_elem,
        4,
        dtype=dtype,
        device=resolved_device,
    )
    class_one_hot.scatter_(-1, class_ids, 1)
    box_xyxy = normal(
        0.5,
        0.15,
        size=(batch_size, max_elem, 1, 4),
        generator=generator,
        device=resolved_device,
        dtype=dtype,
    )
    bbox = xyxy_to_xywh(box_xyxy)
    return torch.concat([class_one_hot.unsqueeze(2), bbox], dim=2)

annotations_from_pku_example

annotations_from_pku_example(
    example: Mapping[str, DSGANExampleValue],
    *,
    max_elem: int = 32,
    include_unconnected_underlays: bool = True,
    canvas_size: tuple[int, int] | None = None,
    vendor_box_fix: bool = False,
) -> dict[
    str, Shaped[torch.Tensor, "..."] | tuple[int, int]
]

Convert a PKU PosterLayout dataset row into public layout tensors.

The adapter filters INVALID annotations, converts pixel ltrb boxes to normalized center xywh, derives canvas size from the image columns, and applies the reference designSeq.reorder ordering policy. Set vendor_box_fix to reproduce the reference loader's overlapping-slice assignment for inverted boxes.

Source code in models/ds-gan/src/ds_gan/processing_ds_gan.py
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
def annotations_from_pku_example(
    example: Mapping[str, DSGANExampleValue],
    *,
    max_elem: int = 32,
    include_unconnected_underlays: bool = True,
    canvas_size: tuple[int, int] | None = None,
    vendor_box_fix: bool = False,
) -> dict[str, Shaped[torch.Tensor, "..."] | tuple[int, int]]:
    """Convert a PKU PosterLayout dataset row into public layout tensors.

    The adapter filters ``INVALID`` annotations, converts pixel ``ltrb`` boxes
    to normalized center ``xywh``, derives canvas size from the image columns,
    and applies the reference ``designSeq.reorder`` ordering policy. Set
    ``vendor_box_fix`` to reproduce the reference loader's overlapping-slice
    assignment for inverted boxes.
    """
    annotations = cast(
        Mapping[
            str, Sequence[str | int | float] | Sequence[Sequence[str | int | float]]
        ],
        example.get("annotations", example),
    )
    raw_labels = cast(Sequence[str | int | float], annotations["cls_elem"])
    raw_boxes = cast(
        Sequence[str | Sequence[int | float | str]], annotations["box_elem"]
    )
    resolved_canvas_size = canvas_size or _canvas_size_from_example(example)
    model_labels: list[int] = []
    public_labels: list[int] = []
    boxes: list[list[float]] = []

    for raw_label, raw_box in zip(raw_labels, raw_boxes, strict=True):
        if isinstance(raw_label, (int, float)):
            numeric_label = int(raw_label)
            if numeric_label != raw_label or numeric_label not in PKU_SOURCE_LABELS:
                raise ValueError(f"Unsupported PKU source label: {raw_label!r}")

            label = PKU_SOURCE_LABELS[numeric_label]
        else:
            label = raw_label

        if label == "INVALID":
            continue
        public_id = PKU_DATASET_LABEL2ID[label]
        model_labels.append(PKU_MODEL_LABEL2ID[label])
        public_labels.append(public_id)
        boxes.append(_parse_box(raw_box, sort=not vendor_box_fix))

    if boxes:
        box_t = torch.tensor(boxes, dtype=torch.float32)
        ordering_boxes = torch.tensor(
            boxes,
            dtype=(
                torch.int64
                if all(
                    not isinstance(value, float) or value.is_integer()
                    for box in boxes
                    for value in box
                )
                else torch.float32
            ),
        )
        order = _designseq_reorder(
            model_labels,
            ordering_boxes,
            max_elem=max_elem,
            include_unconnected_underlays=include_unconnected_underlays,
        )
        box_t = box_t[torch.tensor(order, dtype=torch.long)]
        labels_t = torch.tensor([public_labels[i] for i in order], dtype=torch.long)
        if vendor_box_fix:
            inverted = (box_t[:, 0] > box_t[:, 2]) | (box_t[:, 1] > box_t[:, 3])
            if inverted.any():
                box_t[inverted, :2] = box_t[inverted, 2:]
                box_t[inverted, 2:] = box_t[inverted, :2]
        else:
            left = torch.minimum(box_t[:, 0], box_t[:, 2])
            top = torch.minimum(box_t[:, 1], box_t[:, 3])
            right = torch.maximum(box_t[:, 0], box_t[:, 2])
            bottom = torch.maximum(box_t[:, 1], box_t[:, 3])
            box_t = torch.stack((left, top, right, bottom), dim=-1)
        bbox_t = _normalize_vendor_ltrb(box_t, resolved_canvas_size)
    else:
        bbox_t = torch.zeros(0, 4, dtype=torch.float32)
        labels_t = torch.zeros(0, dtype=torch.long)
    mask_t = torch.ones(labels_t.shape, dtype=torch.bool)
    return {
        "bbox": bbox_t.unsqueeze(0),
        "labels": labels_t.unsqueeze(0),
        "mask": mask_t.unsqueeze(0),
        "canvas_size": resolved_canvas_size,
    }

processor_for_dataset

processor_for_dataset(
    dataset_name: DatasetName | str,
) -> DSGANProcessor

Create a DS-GAN processor for a supported dataset.

Source code in models/ds-gan/src/ds_gan/processing_ds_gan.py
308
309
310
def processor_for_dataset(dataset_name: DatasetName | str) -> DSGANProcessor:
    """Create a DS-GAN processor for a supported dataset."""
    return DSGANProcessor(dataset_name=dataset_name)

configuration_ds_gan

Configuration for converted DS-GAN checkpoints.

DSGANConfig

Bases: PretrainedConfig

Store DS-GAN architecture and dataset metadata.

Parameters:

Name Type Description Default
dataset_name DatasetName | str

Dataset key. DS-GAN currently supports PKU PosterLayout.

pku_posterlayout
backbone str

Timm ResNet backbone name used by the reference checkpoint.

'resnet50'
max_elem int

Maximum number of generated elements.

32
in_channels int

CNN-LSTM input channels after flattening class and box planes.

8
out_channels int

CNN-LSTM convolution output channels.

32
hidden_size int | None

Bidirectional LSTM hidden size.

None
num_layers int

Number of LSTM layers.

4
output_size int

Combined class and box output width in the internal model.

8
image_size tuple[int, int] | list[int]

Processor/model input size as (height, width).

(350, 240)
reference_canvas_size tuple[int, int] | list[int]

Reference normalization canvas as (width, height).

(513, 750)
backbone_feature_size int

Flattened ResNet-FPN spatial size. The reference default is 22 * 15 = 330 for image_size=(350, 240).

330
model_num_classes int

Internal class channels including 0 = no object.

4
id2label Id2LabelMapping | None

Public zero-based semantic label mapping.

None
model_subfolder str

Pipeline subfolder for the model component.

'model'
processor_subfolder str

Pipeline subfolder for the processor component.

'processor'

Examples:

>>> DSGANConfig().max_elem
32
Source code in models/ds-gan/src/ds_gan/configuration_ds_gan.py
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
class DSGANConfig(PretrainedConfig):
    """Store DS-GAN architecture and dataset metadata.

    Args:
        dataset_name: Dataset key. DS-GAN currently supports PKU PosterLayout.
        backbone: Timm ResNet backbone name used by the reference checkpoint.
        max_elem: Maximum number of generated elements.
        in_channels: CNN-LSTM input channels after flattening class and box planes.
        out_channels: CNN-LSTM convolution output channels.
        hidden_size: Bidirectional LSTM hidden size.
        num_layers: Number of LSTM layers.
        output_size: Combined class and box output width in the internal model.
        image_size: Processor/model input size as ``(height, width)``.
        reference_canvas_size: Reference normalization canvas as ``(width, height)``.
        backbone_feature_size: Flattened ResNet-FPN spatial size. The reference
            default is ``22 * 15 = 330`` for ``image_size=(350, 240)``.
        model_num_classes: Internal class channels including ``0 = no object``.
        id2label: Public zero-based semantic label mapping.
        model_subfolder: Pipeline subfolder for the model component.
        processor_subfolder: Pipeline subfolder for the processor component.

    Examples:
        >>> DSGANConfig().max_elem
        32
    """

    model_type = "ds_gan"

    def __init__(
        self,
        dataset_name: DatasetName | str = DatasetName.pku_posterlayout,
        backbone: str = "resnet50",
        max_elem: int = 32,
        in_channels: int = 8,
        out_channels: int = 32,
        hidden_size: int | None = None,
        num_layers: int = 4,
        output_size: int = 8,
        image_size: tuple[int, int] | list[int] = (350, 240),
        reference_canvas_size: tuple[int, int] | list[int] = (513, 750),
        backbone_feature_size: int = 330,
        model_num_classes: int = 4,
        id2label: Id2LabelMapping | None = None,
        label2id: dict[str, int] | None = None,
        model_subfolder: str = "model",
        processor_subfolder: str = "processor",
        condition_types: list[str] | tuple[str, ...] | None = None,
        architectures: list[str] | None = None,
        model_type: str | None = None,
        transformers_version: str | None = None,
        torch_dtype: str | None = None,
        dtype: str | None = None,
        name_or_path: str = "",
        _commit_hash: str | None = None,
        **kwargs: str | int | float | bool | None,
    ) -> None:
        """Initialize DS-GAN configuration."""
        _ = (model_type, transformers_version)
        dataset = normalize_dataset_name(dataset_name)
        if dataset is not DatasetName.pku_posterlayout:
            raise ValueError(f"Unsupported DS-GAN dataset_name: {dataset_name}")

        public_id2label = {
            int(k): v for k, v in (id2label or _semantic_pku_id2label()).items()
        }
        public_label2id = label2id or {v: k for k, v in public_id2label.items()}
        super().__init__(
            id2label=public_id2label,
            label2id=public_label2id,
            architectures=architectures or ["DSGANModel"],
            torch_dtype=torch_dtype,  # ty: ignore[unknown-argument]
            dtype=dtype,
            name_or_path=name_or_path,  # ty: ignore[unknown-argument]
            _commit_hash=_commit_hash,  # ty: ignore[unknown-argument]
            **kwargs,  # ty: ignore[invalid-argument-type]
        )
        self.dataset_name = str(dataset)
        self.backbone = backbone
        self.max_elem = max_elem
        self.in_channels = in_channels
        self.out_channels = out_channels
        self.hidden_size = hidden_size if hidden_size is not None else max_elem * 8
        self.num_layers = num_layers
        self.output_size = output_size
        self.image_size = tuple(int(v) for v in image_size)
        self.reference_canvas_size = tuple(int(v) for v in reference_canvas_size)
        self.backbone_feature_size = backbone_feature_size
        self.model_num_classes = model_num_classes
        self.model_subfolder = model_subfolder
        self.processor_subfolder = processor_subfolder
        self.condition_types = list(condition_types or ["content_image"])

    @property
    def public_num_labels(self) -> int:
        """Return the number of public semantic labels."""
        return len(cast(dict[int, str], self.id2label))

public_num_labels property

public_num_labels: int

Return the number of public semantic labels.

__init__

__init__(
    dataset_name: DatasetName
    | str = DatasetName.pku_posterlayout,
    backbone: str = "resnet50",
    max_elem: int = 32,
    in_channels: int = 8,
    out_channels: int = 32,
    hidden_size: int | None = None,
    num_layers: int = 4,
    output_size: int = 8,
    image_size: tuple[int, int] | list[int] = (350, 240),
    reference_canvas_size: tuple[int, int] | list[int] = (
        513,
        750,
    ),
    backbone_feature_size: int = 330,
    model_num_classes: int = 4,
    id2label: Id2LabelMapping | None = None,
    label2id: dict[str, int] | None = None,
    model_subfolder: str = "model",
    processor_subfolder: str = "processor",
    condition_types: list[str]
    | tuple[str, ...]
    | None = None,
    architectures: list[str] | None = None,
    model_type: str | None = None,
    transformers_version: str | None = None,
    torch_dtype: str | None = None,
    dtype: str | None = None,
    name_or_path: str = "",
    _commit_hash: str | None = None,
    **kwargs: str | int | float | bool | None,
) -> None

Initialize DS-GAN configuration.

Source code in models/ds-gan/src/ds_gan/configuration_ds_gan.py
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
def __init__(
    self,
    dataset_name: DatasetName | str = DatasetName.pku_posterlayout,
    backbone: str = "resnet50",
    max_elem: int = 32,
    in_channels: int = 8,
    out_channels: int = 32,
    hidden_size: int | None = None,
    num_layers: int = 4,
    output_size: int = 8,
    image_size: tuple[int, int] | list[int] = (350, 240),
    reference_canvas_size: tuple[int, int] | list[int] = (513, 750),
    backbone_feature_size: int = 330,
    model_num_classes: int = 4,
    id2label: Id2LabelMapping | None = None,
    label2id: dict[str, int] | None = None,
    model_subfolder: str = "model",
    processor_subfolder: str = "processor",
    condition_types: list[str] | tuple[str, ...] | None = None,
    architectures: list[str] | None = None,
    model_type: str | None = None,
    transformers_version: str | None = None,
    torch_dtype: str | None = None,
    dtype: str | None = None,
    name_or_path: str = "",
    _commit_hash: str | None = None,
    **kwargs: str | int | float | bool | None,
) -> None:
    """Initialize DS-GAN configuration."""
    _ = (model_type, transformers_version)
    dataset = normalize_dataset_name(dataset_name)
    if dataset is not DatasetName.pku_posterlayout:
        raise ValueError(f"Unsupported DS-GAN dataset_name: {dataset_name}")

    public_id2label = {
        int(k): v for k, v in (id2label or _semantic_pku_id2label()).items()
    }
    public_label2id = label2id or {v: k for k, v in public_id2label.items()}
    super().__init__(
        id2label=public_id2label,
        label2id=public_label2id,
        architectures=architectures or ["DSGANModel"],
        torch_dtype=torch_dtype,  # ty: ignore[unknown-argument]
        dtype=dtype,
        name_or_path=name_or_path,  # ty: ignore[unknown-argument]
        _commit_hash=_commit_hash,  # ty: ignore[unknown-argument]
        **kwargs,  # ty: ignore[invalid-argument-type]
    )
    self.dataset_name = str(dataset)
    self.backbone = backbone
    self.max_elem = max_elem
    self.in_channels = in_channels
    self.out_channels = out_channels
    self.hidden_size = hidden_size if hidden_size is not None else max_elem * 8
    self.num_layers = num_layers
    self.output_size = output_size
    self.image_size = tuple(int(v) for v in image_size)
    self.reference_canvas_size = tuple(int(v) for v in reference_canvas_size)
    self.backbone_feature_size = backbone_feature_size
    self.model_num_classes = model_num_classes
    self.model_subfolder = model_subfolder
    self.processor_subfolder = processor_subfolder
    self.condition_types = list(condition_types or ["content_image"])

default_ds_gan_config

default_ds_gan_config() -> DSGANConfig

Return the reference-compatible DS-GAN default configuration.

Examples:

>>> default_ds_gan_config().backbone
'resnet50'
Source code in models/ds-gan/src/ds_gan/configuration_ds_gan.py
125
126
127
128
129
130
131
132
def default_ds_gan_config() -> DSGANConfig:
    """Return the reference-compatible DS-GAN default configuration.

    Examples:
        >>> default_ds_gan_config().backbone
        'resnet50'
    """
    return DSGANConfig()

pku_model_label2id

pku_model_label2id() -> dict[str, int]

Return DS-GAN model labels including the no-object class.

Examples:

>>> pku_model_label2id()["no_object"]
0
Source code in models/ds-gan/src/ds_gan/configuration_ds_gan.py
135
136
137
138
139
140
141
142
def pku_model_label2id() -> dict[str, int]:
    """Return DS-GAN model labels including the no-object class.

    Examples:
        >>> pku_model_label2id()["no_object"]
        0
    """
    return {"no_object": 0, "text": 1, "logo": 2, "underlay": 3}

pku_dataset_label2id

pku_dataset_label2id() -> dict[str, int]

Return PKU dataset annotation labels including INVALID.

Source code in models/ds-gan/src/ds_gan/configuration_ds_gan.py
145
146
147
def pku_dataset_label2id() -> dict[str, int]:
    """Return PKU dataset annotation labels including ``INVALID``."""
    return {"text": 0, "logo": 1, "underlay": 2, "INVALID": 3}

conversion

Conversion helpers for original PosterLayout DS-GAN checkpoints.

convert_vendor_state_dict

convert_vendor_state_dict(
    state_dict: Mapping[str, Shaped[Tensor, "..."]],
) -> dict[str, Shaped[torch.Tensor, "..."]]

Convert vendor DS-GAN generator keys to DSGANModel keys.

The released checkpoint was commonly saved from torch.nn.DataParallel; this helper strips the leading module. prefix and keeps all generator module names otherwise unchanged.

Parameters:

Name Type Description Default
state_dict Mapping[str, Shaped[Tensor, '...']]

Original checkpoint mapping.

required

Returns:

Type Description
dict[str, Shaped[Tensor, '...']]

Converted state dictionary.

Examples:

>>> convert_vendor_state_dict({"module.fc1.weight": torch.zeros(1)})["fc1.weight"].shape
torch.Size([1])
Source code in models/ds-gan/src/ds_gan/conversion.py
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
def convert_vendor_state_dict(
    state_dict: Mapping[str, Shaped[torch.Tensor, "..."]],
) -> dict[str, Shaped[torch.Tensor, "..."]]:
    """Convert vendor DS-GAN generator keys to ``DSGANModel`` keys.

    The released checkpoint was commonly saved from ``torch.nn.DataParallel``;
    this helper strips the leading ``module.`` prefix and keeps all generator
    module names otherwise unchanged.

    Args:
        state_dict: Original checkpoint mapping.

    Returns:
        Converted state dictionary.

    Examples:
        >>> convert_vendor_state_dict({"module.fc1.weight": torch.zeros(1)})["fc1.weight"].shape
        torch.Size([1])
    """
    converted: dict[str, Shaped[torch.Tensor, "..."]] = {}
    for key, value in state_dict.items():
        name = key.removeprefix("module.")
        name = name.removeprefix("generator.")
        converted[name] = value
    return converted

config_from_vendor_args

config_from_vendor_args(
    args: Namespace
    | SimpleNamespace
    | Mapping[str, DSGANArgValue]
    | None = None,
) -> DSGANConfig

Build a DS-GAN config from vendor args or defaults.

Source code in models/ds-gan/src/ds_gan/conversion.py
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
def config_from_vendor_args(
    args: argparse.Namespace
    | SimpleNamespace
    | Mapping[str, DSGANArgValue]
    | None = None,
) -> DSGANConfig:
    """Build a DS-GAN config from vendor args or defaults."""
    if args is None:
        return DSGANConfig()
    values = _values(args)
    max_elem = _int_value(values, "max_elem", 32)
    return DSGANConfig(
        backbone=str(values.get("backbone", "resnet50")),
        max_elem=max_elem,
        in_channels=_int_value(values, "in_channels", 8),
        out_channels=_int_value(values, "out_channels", 32),
        hidden_size=_int_value(values, "hidden_size", max_elem * 8),
        num_layers=_int_value(values, "num_layers", 4),
        output_size=_int_value(values, "output_size", 8),
    )

model_card

Hub model-card helper for converted DS-GAN checkpoints.

dsgan_model_card

dsgan_model_card() -> ModelCard

Build the DS-GAN PKU PosterLayout Hub model card.

Source code in models/ds-gan/src/ds_gan/model_card.py
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
def dsgan_model_card() -> ModelCard:
    """Build the DS-GAN PKU PosterLayout Hub model card."""
    return build_layout_model_card(
        model_id="creative-graphic-design/ds-gan-pku-posterlayout",
        model_name="DS-GAN PosterLayout",
        dataset_ids=["creative-graphic-design/PKU-PosterLayout"],
        license="other",
        library_name="transformers",
        pipeline_tag="image-to-image",
        tags=["ds-gan", "posterlayout", "layout-generation", "poster-generation"],
        model_details=(
            "DS-GAN predicts poster element boxes and semantic labels from an RGB "
            "poster/background image plus saliency. The converted package returns "
            "normalized center xywh boxes, zero-based labels, and mask-based padding."
        ),
        intended_uses=(
            "Research use for content-aware poster layout generation and checkpoint "
            "parity studies against the original CVPR 2023 implementation."
        ),
        limitations=(
            "The upstream repository does not include a redistribution license. "
            "Converted weights should not be published until license permission is "
            "resolved."
        ),
        how_to_use=(
            "from ds_gan import DSGANPipeline\n\n"
            "pipe = DSGANPipeline.from_pretrained(\n"
            '    "creative-graphic-design/ds-gan-pku-posterlayout"\n'
            ")\n"
            "out = pipe(images=image, saliency=saliency, seed=0)"
        ),
        training_data=(
            "PKU PosterLayout via creative-graphic-design/PKU-PosterLayout; "
            "annotations contain pixel ltrb boxes and an INVALID class that is "
            "excluded from public semantic labels."
        ),
        parity_metrics=[
            {
                "dataset": "PKU PosterLayout",
                "tokenizer_exact": "not applicable",
                "deterministic_exact": "905/905 exact",
                "logits_max_abs": 0.0,
                "logits_max_rel": 0.0,
            }
        ],
        citation_bibtex=(
            "@inproceedings{Hsu-2023-posterlayout,\n"
            "  title={PosterLayout: A New Benchmark and Approach for "
            "Content-Aware Visual-Textual Presentation Layout},\n"
            "  author={HsiaoYuan Hsu and Xiangteng He and Yuxin Peng and "
            "Hao Kong and Qing Zhang},\n"
            "  booktitle={CVPR},\n"
            "  year={2023}\n"
            "}"
        ),
        original_implementation_url=(
            "https://github.com/PKU-ICST-MIPL/PosterLayout-CVPR2023"
        ),
    )

write_dsgan_model_card

write_dsgan_model_card(output_dir: str | Path) -> Path

Write README.md for a converted DS-GAN checkpoint directory.

Source code in models/ds-gan/src/ds_gan/model_card.py
73
74
75
76
77
def write_dsgan_model_card(output_dir: str | Path) -> Path:
    """Write ``README.md`` for a converted DS-GAN checkpoint directory."""
    path = Path(output_dir) / "README.md"
    path.write_text(str(dsgan_model_card()), encoding="utf-8")
    return path

modeling_ds_gan

PyTorch/Transformers implementation of the PosterLayout DS-GAN generator.

DSGANModelOutput dataclass

Bases: ModelOutput

Raw DS-GAN generator output.

Attributes:

Name Type Description
class_probs Float[Tensor, 'batch elements 4']

Internal class probabilities with shape (batch, elements, 4) where id 0 is no object.

bbox Float[Tensor, 'batch elements 4'] | None

Normalized center xywh boxes with shape (batch, elements, 4).

initial_layout Float[Tensor, 'batch elements 2 4'] | None

Initial class/box layout passed to the generator.

Source code in models/ds-gan/src/ds_gan/modeling_ds_gan.py
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
@dataclass
class DSGANModelOutput(ModelOutput):
    """Raw DS-GAN generator output.

    Attributes:
        class_probs: Internal class probabilities with shape
            ``(batch, elements, 4)`` where id 0 is ``no object``.
        bbox: Normalized center ``xywh`` boxes with shape
            ``(batch, elements, 4)``.
        initial_layout: Initial class/box layout passed to the generator.
    """

    class_probs: Float[torch.Tensor, "batch elements 4"]
    bbox: Float[torch.Tensor, "batch elements 4"] | None = None
    initial_layout: Float[torch.Tensor, "batch elements 2 4"] | None = None

ResnetBackbone

Bases: Module

DS-GAN ResNet-FPN encoder used to initialize the DS-GAN LSTM state.

Source code in models/ds-gan/src/ds_gan/modeling_ds_gan.py
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
class ResnetBackbone(nn.Module):
    """DS-GAN ResNet-FPN encoder used to initialize the DS-GAN LSTM state."""

    def __init__(
        self, config: DSGANConfig, *, backbone_weights: str | Path | None = None
    ) -> None:
        """Initialize the ResNet-FPN encoder."""
        super().__init__()
        try:
            import timm
        except ImportError as exc:
            raise ImportError(
                "DSGANModel requires the optional timm dependency"
            ) from exc

        if config.backbone == "resnet50":
            channels = (1024, 2048)
        elif config.backbone == "resnet18":
            channels = (256, 512)
        else:
            raise ValueError(f"Unsupported DS-GAN backbone: {config.backbone}")

        resnet = timm.create_model(config.backbone, pretrained=False)
        if backbone_weights is not None:
            weight_path = Path(backbone_weights)
            if not weight_path.is_file():
                raise FileNotFoundError(f"backbone weights not found: {weight_path}")

            weights = torch.load(weight_path, map_location="cpu", weights_only=False)
            if not isinstance(weights, Mapping):
                raise TypeError("backbone weights must contain a state-dict mapping")

            resnet.load_state_dict(weights, strict=True)

        resnet.conv1 = nn.Conv2d(
            4,
            64,
            kernel_size=(7, 7),
            stride=(2, 2),
            padding=(3, 3),
            bias=False,
        )
        children = list(resnet.children())
        self.resnet_tilconv4 = nn.Sequential(*children[:7])
        self.resnet_conv5 = children[7]
        self.fpn_conv11_4 = nn.Conv2d(channels[0], 256, 1, 1, 0)
        self.fpn_conv11_5 = nn.Conv2d(channels[1], 256, 1, 1, 0)
        self.fpn_conv33 = nn.Conv2d(256, 256, 3, 1, 1)
        self.proj = nn.Conv2d(512, 8 * config.max_elem, 1, 1, 0)
        self.fc_h0 = nn.Linear(config.backbone_feature_size, config.num_layers * 2)

    def forward(
        self, pixel_values: Float[torch.Tensor, "batch 4 height width"]
    ) -> Float[torch.Tensor, "layers2 batch hidden"]:
        """Encode image/saliency tensors into an LSTM initial hidden state."""
        resnet_f4 = self.resnet_tilconv4(pixel_values)
        resnet_f5 = self.resnet_conv5(resnet_f4)
        resnet_f4p = self.fpn_conv11_4(resnet_f4)
        resnet_f5p = self.fpn_conv11_5(resnet_f5)
        resnet_f5up = F.interpolate(
            resnet_f5p, size=resnet_f4p.shape[2:], mode="nearest"
        )
        fused = torch.concat(
            [resnet_f5up, self.fpn_conv33(resnet_f5up + resnet_f4p)], dim=1
        )
        projected = self.proj(fused)
        flattened = projected.flatten(start_dim=-2)
        return self.fc_h0(flattened).permute(2, 0, 1)

__init__

__init__(
    config: DSGANConfig,
    *,
    backbone_weights: str | Path | None = None,
) -> None

Initialize the ResNet-FPN encoder.

Source code in models/ds-gan/src/ds_gan/modeling_ds_gan.py
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def __init__(
    self, config: DSGANConfig, *, backbone_weights: str | Path | None = None
) -> None:
    """Initialize the ResNet-FPN encoder."""
    super().__init__()
    try:
        import timm
    except ImportError as exc:
        raise ImportError(
            "DSGANModel requires the optional timm dependency"
        ) from exc

    if config.backbone == "resnet50":
        channels = (1024, 2048)
    elif config.backbone == "resnet18":
        channels = (256, 512)
    else:
        raise ValueError(f"Unsupported DS-GAN backbone: {config.backbone}")

    resnet = timm.create_model(config.backbone, pretrained=False)
    if backbone_weights is not None:
        weight_path = Path(backbone_weights)
        if not weight_path.is_file():
            raise FileNotFoundError(f"backbone weights not found: {weight_path}")

        weights = torch.load(weight_path, map_location="cpu", weights_only=False)
        if not isinstance(weights, Mapping):
            raise TypeError("backbone weights must contain a state-dict mapping")

        resnet.load_state_dict(weights, strict=True)

    resnet.conv1 = nn.Conv2d(
        4,
        64,
        kernel_size=(7, 7),
        stride=(2, 2),
        padding=(3, 3),
        bias=False,
    )
    children = list(resnet.children())
    self.resnet_tilconv4 = nn.Sequential(*children[:7])
    self.resnet_conv5 = children[7]
    self.fpn_conv11_4 = nn.Conv2d(channels[0], 256, 1, 1, 0)
    self.fpn_conv11_5 = nn.Conv2d(channels[1], 256, 1, 1, 0)
    self.fpn_conv33 = nn.Conv2d(256, 256, 3, 1, 1)
    self.proj = nn.Conv2d(512, 8 * config.max_elem, 1, 1, 0)
    self.fc_h0 = nn.Linear(config.backbone_feature_size, config.num_layers * 2)

forward

forward(
    pixel_values: Float[Tensor, "batch 4 height width"],
) -> Float[torch.Tensor, "layers2 batch hidden"]

Encode image/saliency tensors into an LSTM initial hidden state.

Source code in models/ds-gan/src/ds_gan/modeling_ds_gan.py
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
def forward(
    self, pixel_values: Float[torch.Tensor, "batch 4 height width"]
) -> Float[torch.Tensor, "layers2 batch hidden"]:
    """Encode image/saliency tensors into an LSTM initial hidden state."""
    resnet_f4 = self.resnet_tilconv4(pixel_values)
    resnet_f5 = self.resnet_conv5(resnet_f4)
    resnet_f4p = self.fpn_conv11_4(resnet_f4)
    resnet_f5p = self.fpn_conv11_5(resnet_f5)
    resnet_f5up = F.interpolate(
        resnet_f5p, size=resnet_f4p.shape[2:], mode="nearest"
    )
    fused = torch.concat(
        [resnet_f5up, self.fpn_conv33(resnet_f5up + resnet_f4p)], dim=1
    )
    projected = self.proj(fused)
    flattened = projected.flatten(start_dim=-2)
    return self.fc_h0(flattened).permute(2, 0, 1)

CNNLSTM

Bases: Module

DS-GAN CNN-LSTM sequence model.

Source code in models/ds-gan/src/ds_gan/modeling_ds_gan.py
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
class CNNLSTM(nn.Module):
    """DS-GAN CNN-LSTM sequence model."""

    def __init__(self, config: DSGANConfig) -> None:
        """Initialize the CNN-LSTM block."""
        super().__init__()
        self.conv = nn.Sequential(
            nn.Conv1d(
                in_channels=config.in_channels,
                out_channels=config.out_channels,
                kernel_size=3,
                padding="same",
            ),
            nn.ReLU(),
            nn.MaxPool1d(3, stride=1, padding=1),
        )
        self.lstm = nn.LSTM(
            input_size=config.out_channels,
            hidden_size=config.hidden_size,
            num_layers=config.num_layers,
            batch_first=True,
            bidirectional=True,
        )

    def forward(
        self,
        layout: Float[torch.Tensor, "batch elements 2 4"],
        h0: Float[torch.Tensor, "layers2 batch hidden"],
    ) -> Float[torch.Tensor, "batch elements hidden2"]:
        """Run the DS-GAN CNN-LSTM over initial layout tensors."""
        self.lstm.flatten_parameters()
        x = layout.flatten(start_dim=2).permute(0, 2, 1).contiguous()
        x = self.conv(x).permute(0, 2, 1).contiguous()
        output, _ = self.lstm(x, (torch.zeros_like(h0).contiguous(), h0.contiguous()))
        return output

__init__

__init__(config: DSGANConfig) -> None

Initialize the CNN-LSTM block.

Source code in models/ds-gan/src/ds_gan/modeling_ds_gan.py
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
def __init__(self, config: DSGANConfig) -> None:
    """Initialize the CNN-LSTM block."""
    super().__init__()
    self.conv = nn.Sequential(
        nn.Conv1d(
            in_channels=config.in_channels,
            out_channels=config.out_channels,
            kernel_size=3,
            padding="same",
        ),
        nn.ReLU(),
        nn.MaxPool1d(3, stride=1, padding=1),
    )
    self.lstm = nn.LSTM(
        input_size=config.out_channels,
        hidden_size=config.hidden_size,
        num_layers=config.num_layers,
        batch_first=True,
        bidirectional=True,
    )

forward

forward(
    layout: Float[Tensor, "batch elements 2 4"],
    h0: Float[Tensor, "layers2 batch hidden"],
) -> Float[torch.Tensor, "batch elements hidden2"]

Run the DS-GAN CNN-LSTM over initial layout tensors.

Source code in models/ds-gan/src/ds_gan/modeling_ds_gan.py
137
138
139
140
141
142
143
144
145
146
147
def forward(
    self,
    layout: Float[torch.Tensor, "batch elements 2 4"],
    h0: Float[torch.Tensor, "layers2 batch hidden"],
) -> Float[torch.Tensor, "batch elements hidden2"]:
    """Run the DS-GAN CNN-LSTM over initial layout tensors."""
    self.lstm.flatten_parameters()
    x = layout.flatten(start_dim=2).permute(0, 2, 1).contiguous()
    x = self.conv(x).permute(0, 2, 1).contiguous()
    output, _ = self.lstm(x, (torch.zeros_like(h0).contiguous(), h0.contiguous()))
    return output

DSGANModel

Bases: PreTrainedModel

Transformers-compatible DS-GAN generator.

Parameters:

Name Type Description Default
config DSGANConfig

DS-GAN model configuration.

required

Examples:

>>> config = DSGANConfig(backbone="resnet18", max_elem=4, hidden_size=32, num_layers=2, image_size=(64, 64), backbone_feature_size=16)
>>> model = DSGANModel(config)
>>> model.config.model_type
'ds_gan'
Source code in models/ds-gan/src/ds_gan/modeling_ds_gan.py
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
class DSGANModel(PreTrainedModel):
    """Transformers-compatible DS-GAN generator.

    Args:
        config: DS-GAN model configuration.

    Examples:
        >>> config = DSGANConfig(backbone="resnet18", max_elem=4, hidden_size=32, num_layers=2, image_size=(64, 64), backbone_feature_size=16)
        >>> model = DSGANModel(config)
        >>> model.config.model_type
        'ds_gan'
    """

    config_class = DSGANConfig
    base_model_prefix = "ds_gan"
    supports_gradient_checkpointing = False

    def _init_weights(self, module: nn.Module) -> None:
        """Leave module-specific initialization unchanged during ``post_init``."""
        del module

    def __init__(
        self, config: DSGANConfig, *, backbone_weights: str | Path | None = None
    ) -> None:
        """Initialize DS-GAN generator layers."""
        super().__init__(config)
        self.resnet_fpn = ResnetBackbone(config, backbone_weights=backbone_weights)
        self.cnnlstm = CNNLSTM(config)
        self.fc1 = nn.Linear(2 * config.hidden_size, config.output_size // 2)
        self.fc2 = nn.Linear(2 * config.hidden_size, config.output_size // 2)
        self.post_init()

    def forward(
        self,
        pixel_values: Float[torch.Tensor, "batch 4 height width"],
        layout: Float[torch.Tensor, "batch elements 2 4"],
        return_dict: bool = True,
    ) -> (
        DSGANModelOutput
        | tuple[
            Float[torch.Tensor, "batch elements 4"],
            Float[torch.Tensor, "batch elements 4"],
        ]
    ):
        """Run a DS-GAN generator forward pass.

        Args:
            pixel_values: RGB plus saliency tensor shaped ``(B, 4, H, W)``.
            layout: Initial internal layout shaped ``(B, max_elem, 2, 4)``.
            return_dict: Whether to return a dataclass output.

        Returns:
            Raw class probabilities and normalized center ``xywh`` boxes.

        Raises:
            ValueError: If tensor shapes do not match the config.
        """
        if pixel_values.ndim != 4 or pixel_values.shape[1] != 4:
            raise ValueError("pixel_values must have shape (batch, 4, height, width)")

        expected_layout = (pixel_values.shape[0], self.config.max_elem, 2, 4)
        if tuple(layout.shape) != expected_layout:
            raise ValueError(f"layout must have shape {expected_layout}")

        pixel_values = pixel_values.to(dtype=self.dtype)
        layout = layout.to(device=pixel_values.device, dtype=self.dtype)
        h0 = self.resnet_fpn(pixel_values)
        lstm_output = self.cnnlstm(layout, h0)
        class_probs = torch.softmax(self.fc1(lstm_output), dim=-1)
        bbox = torch.sigmoid(self.fc2(lstm_output))
        if not return_dict:
            return class_probs, bbox
        return DSGANModelOutput(
            class_probs=class_probs,
            bbox=bbox,
            initial_layout=layout,
        )

__init__

__init__(
    config: DSGANConfig,
    *,
    backbone_weights: str | Path | None = None,
) -> None

Initialize DS-GAN generator layers.

Source code in models/ds-gan/src/ds_gan/modeling_ds_gan.py
171
172
173
174
175
176
177
178
179
180
def __init__(
    self, config: DSGANConfig, *, backbone_weights: str | Path | None = None
) -> None:
    """Initialize DS-GAN generator layers."""
    super().__init__(config)
    self.resnet_fpn = ResnetBackbone(config, backbone_weights=backbone_weights)
    self.cnnlstm = CNNLSTM(config)
    self.fc1 = nn.Linear(2 * config.hidden_size, config.output_size // 2)
    self.fc2 = nn.Linear(2 * config.hidden_size, config.output_size // 2)
    self.post_init()

forward

forward(
    pixel_values: Float[Tensor, "batch 4 height width"],
    layout: Float[Tensor, "batch elements 2 4"],
    return_dict: bool = True,
) -> (
    DSGANModelOutput
    | tuple[
        Float[torch.Tensor, "batch elements 4"],
        Float[torch.Tensor, "batch elements 4"],
    ]
)

Run a DS-GAN generator forward pass.

Parameters:

Name Type Description Default
pixel_values Float[Tensor, 'batch 4 height width']

RGB plus saliency tensor shaped (B, 4, H, W).

required
layout Float[Tensor, 'batch elements 2 4']

Initial internal layout shaped (B, max_elem, 2, 4).

required
return_dict bool

Whether to return a dataclass output.

True

Returns:

Type Description
DSGANModelOutput | tuple[Float[Tensor, 'batch elements 4'], Float[Tensor, 'batch elements 4']]

Raw class probabilities and normalized center xywh boxes.

Raises:

Type Description
ValueError

If tensor shapes do not match the config.

Source code in models/ds-gan/src/ds_gan/modeling_ds_gan.py
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
def forward(
    self,
    pixel_values: Float[torch.Tensor, "batch 4 height width"],
    layout: Float[torch.Tensor, "batch elements 2 4"],
    return_dict: bool = True,
) -> (
    DSGANModelOutput
    | tuple[
        Float[torch.Tensor, "batch elements 4"],
        Float[torch.Tensor, "batch elements 4"],
    ]
):
    """Run a DS-GAN generator forward pass.

    Args:
        pixel_values: RGB plus saliency tensor shaped ``(B, 4, H, W)``.
        layout: Initial internal layout shaped ``(B, max_elem, 2, 4)``.
        return_dict: Whether to return a dataclass output.

    Returns:
        Raw class probabilities and normalized center ``xywh`` boxes.

    Raises:
        ValueError: If tensor shapes do not match the config.
    """
    if pixel_values.ndim != 4 or pixel_values.shape[1] != 4:
        raise ValueError("pixel_values must have shape (batch, 4, height, width)")

    expected_layout = (pixel_values.shape[0], self.config.max_elem, 2, 4)
    if tuple(layout.shape) != expected_layout:
        raise ValueError(f"layout must have shape {expected_layout}")

    pixel_values = pixel_values.to(dtype=self.dtype)
    layout = layout.to(device=pixel_values.device, dtype=self.dtype)
    h0 = self.resnet_fpn(pixel_values)
    lstm_output = self.cnnlstm(layout, h0)
    class_probs = torch.softmax(self.fc1(lstm_output), dim=-1)
    bbox = torch.sigmoid(self.fc2(lstm_output))
    if not return_dict:
        return class_probs, bbox
    return DSGANModelOutput(
        class_probs=class_probs,
        bbox=bbox,
        initial_layout=layout,
    )

xyxy_to_xywh

xyxy_to_xywh(
    bbox: Float[Tensor, "... 4"],
) -> Float[torch.Tensor, "... 4"]

Convert left/top/right/bottom boxes to center xywh boxes.

Source code in models/ds-gan/src/ds_gan/modeling_ds_gan.py
229
230
231
232
233
234
235
236
237
def xyxy_to_xywh(
    bbox: Float[torch.Tensor, "... 4"],
) -> Float[torch.Tensor, "... 4"]:
    """Convert left/top/right/bottom boxes to center ``xywh`` boxes."""
    left, top, right, bottom = bbox.unbind(-1)
    return torch.stack(
        ((left + right) / 2, (top + bottom) / 2, right - left, bottom - top),
        dim=-1,
    )

random_initial_layout

random_initial_layout(
    batch_size: int,
    max_elem: int,
    *,
    generator: Generator | None = None,
    seed: int | None = None,
    device: device | str | None = None,
    dtype: dtype = torch.float32,
    weighted_classes: bool = True,
    use_numpy_classes: bool = False,
) -> Float[torch.Tensor, "batch elements 2 4"]

Sample the DS-GAN initial layout tensor.

Parameters:

Name Type Description Default
batch_size int

Batch size.

required
max_elem int

Number of layout slots.

required
generator Generator | None

Optional torch generator. Takes precedence over seed.

None
seed int | None

Convenience seed used only when generator is absent.

None
device device | str | None

Target torch device.

None
dtype dtype

Target floating dtype.

float32
weighted_classes bool

Whether to use the released inference class prior.

True
use_numpy_classes bool

Use NumPy's legacy RandomState class sampler to mirror the reference script's weighted class prior when seed is supplied. Torch box sampling still follows generator or seed.

False

Returns:

Type Description
Float[Tensor, 'batch elements 2 4']

Tensor shaped (batch, max_elem, 2, 4).

Source code in models/ds-gan/src/ds_gan/modeling_ds_gan.py
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
def random_initial_layout(
    batch_size: int,
    max_elem: int,
    *,
    generator: torch.Generator | None = None,
    seed: int | None = None,
    device: torch.device | str | None = None,
    dtype: torch.dtype = torch.float32,
    weighted_classes: bool = True,
    use_numpy_classes: bool = False,
) -> Float[torch.Tensor, "batch elements 2 4"]:
    """Sample the DS-GAN initial layout tensor.

    Args:
        batch_size: Batch size.
        max_elem: Number of layout slots.
        generator: Optional torch generator. Takes precedence over ``seed``.
        seed: Convenience seed used only when ``generator`` is absent.
        device: Target torch device.
        dtype: Target floating dtype.
        weighted_classes: Whether to use the released inference class prior.
        use_numpy_classes: Use NumPy's legacy ``RandomState`` class sampler to
            mirror the reference script's weighted class prior when ``seed`` is
            supplied. Torch box sampling still follows ``generator`` or ``seed``.

    Returns:
        Tensor shaped ``(batch, max_elem, 2, 4)``.
    """
    resolved_device = (
        torch.device(device) if device is not None else torch.device("cpu")
    )
    generator = resolve_torch_generator(generator=generator, seed=seed)
    if weighted_classes:
        probs = torch.tensor((0.1, 0.8, 1.0, 1.0), device=resolved_device)
        probs = probs / probs.sum()
    else:
        probs = torch.full((4,), 0.25, device=resolved_device)
    if use_numpy_classes:
        rng = np.random.RandomState(seed)
        np_probs = probs.detach().cpu().numpy()
        class_ids = torch.as_tensor(
            rng.choice(4, size=(batch_size, max_elem, 1), p=np_probs),
            dtype=torch.long,
            device=resolved_device,
        )
    else:
        class_ids = multinomial(
            probs,
            num_samples=batch_size * max_elem,
            replacement=True,
            generator=generator,
            device=resolved_device,
        ).reshape(batch_size, max_elem, 1)
    class_one_hot = torch.zeros(
        batch_size,
        max_elem,
        4,
        dtype=dtype,
        device=resolved_device,
    )
    class_one_hot.scatter_(-1, class_ids, 1)
    box_xyxy = normal(
        0.5,
        0.15,
        size=(batch_size, max_elem, 1, 4),
        generator=generator,
        device=resolved_device,
        dtype=dtype,
    )
    bbox = xyxy_to_xywh(box_xyxy)
    return torch.concat([class_one_hot.unsqueeze(2), bbox], dim=2)

pipeline_ds_gan

Pipeline interface for PosterLayout DS-GAN generation.

DSGANPipelineComponent

Bases: Protocol

Runtime-checkable loaded pipeline component marker.

Source code in models/ds-gan/src/ds_gan/pipeline_ds_gan.py
31
32
33
@runtime_checkable
class DSGANPipelineComponent(Protocol):
    """Runtime-checkable loaded pipeline component marker."""

OutputType

Bases: StrEnum

Supported DS-GAN pipeline output containers.

Source code in models/ds-gan/src/ds_gan/pipeline_ds_gan.py
36
37
38
39
40
class OutputType(StrEnum):
    """Supported DS-GAN pipeline output containers."""

    dataclass = auto()
    dict = auto()

DSGANPipeline

Bases: LayoutGenerationPipeline

Transformers-side pipeline for content-aware PosterLayout generation.

Parameters:

Name Type Description Default
model DSGANModel

DS-GAN generator.

required
processor DSGANProcessor | None

Optional processor for images and output decoding.

None
config DSGANConfig | None

Optional root pipeline config.

None
device str | device | None

Optional runtime device.

None

Examples:

>>> config = DSGANConfig(backbone="resnet18", max_elem=4, hidden_size=32, num_layers=2, image_size=(64, 64), backbone_feature_size=16)
>>> pipe = DSGANPipeline(DSGANModel(config))
>>> pipe.config.model_type
'ds_gan'
Source code in models/ds-gan/src/ds_gan/pipeline_ds_gan.py
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
class DSGANPipeline(LayoutGenerationPipeline):
    """Transformers-side pipeline for content-aware PosterLayout generation.

    Args:
        model: DS-GAN generator.
        processor: Optional processor for images and output decoding.
        config: Optional root pipeline config.
        device: Optional runtime device.

    Examples:
        >>> config = DSGANConfig(backbone="resnet18", max_elem=4, hidden_size=32, num_layers=2, image_size=(64, 64), backbone_feature_size=16)
        >>> pipe = DSGANPipeline(DSGANModel(config))
        >>> pipe.config.model_type
        'ds_gan'
    """

    config_class: ClassVar[type[PretrainedConfig]] = DSGANConfig
    component_specs: ClassVar[dict[str, PipelineComponentSpec]] = {
        "model": PipelineComponentSpec(
            attribute_name="model",
            loader=_load_model_component,
            marker_file="config.json",
            config_subfolder_attribute="model_subfolder",
        ),
        "processor": PipelineComponentSpec(
            attribute_name="processor",
            loader=_load_processor_component,
            marker_file="processor_config.json",
            save_with_is_main_process=False,
            config_subfolder_attribute="processor_subfolder",
        ),
    }

    config: DSGANConfig
    model: DSGANModel
    processor: DSGANProcessor

    def __init__(
        self,
        model: DSGANModel,
        processor: DSGANProcessor | None = None,
        config: DSGANConfig | None = None,
        device: str | torch.device | None = None,
    ) -> None:
        """Initialize DS-GAN pipeline."""
        super().__init__(config or model.config)
        self.config = config or model.config
        self.model = model
        self.processor = processor or DSGANProcessor(
            dataset_name=self.config.dataset_name,
            id2label=cast(dict[int | str, str], self.config.id2label),
            image_size=cast(tuple[int, int], self.config.image_size),
        )
        self.model.eval()
        if device is not None:
            self.to(device)

    @classmethod
    def _from_pretrained_components(
        cls,
        *,
        config: PretrainedConfig,
        components: Mapping[str, DSGANPipelineComponent | None],
    ) -> "DSGANPipeline":
        """Build a pipeline from loaded root components."""
        return cls(
            config=cast(DSGANConfig, config),
            model=cast(DSGANModel, components["model"]),
            processor=cast(DSGANProcessor, components["processor"]),
        )

    @torch.no_grad()
    def __call__(  # ty: ignore[invalid-method-override]
        self,
        images: ImageInput
        | list[ImageInput]
        | Float[torch.Tensor, "..."]
        | None = None,
        *,
        batch_size: int = 1,
        seed: int | None = None,
        generator: torch.Generator | None = None,
        condition_type: ConditionType | str = ConditionType.content_image,
        labels: Int[torch.Tensor, "batch elements"]
        | Sequence[ArrayLikeInput]
        | None = None,
        bbox: Float[torch.Tensor, "batch elements 4"]
        | Sequence[ArrayLikeInput]
        | None = None,
        mask: Bool[torch.Tensor, "batch elements"]
        | Sequence[ArrayLikeInput]
        | None = None,
        num_elements: int | list[int] | Int[torch.Tensor, "batch"] | None = None,
        box_format: BoxFormat | str = BoxFormat.xywh,
        normalized: bool = True,
        canvas_size: tuple[int, int] | None = None,
        num_inference_steps: int | None = None,
        output_type: OutputType | str = OutputType.dataclass,
        return_intermediates: bool = False,
        saliency: ImageInput
        | list[ImageInput]
        | Float[torch.Tensor, "..."]
        | None = None,
        saliency_pfpnet: ImageInput
        | list[ImageInput]
        | Float[torch.Tensor, "..."]
        | None = None,
        saliency_basnet: ImageInput
        | list[ImageInput]
        | Float[torch.Tensor, "..."]
        | None = None,
        pixel_values: Float[torch.Tensor, "batch 4 height width"] | None = None,
        initial_layout: Float[torch.Tensor, "batch elements 2 4"] | None = None,
    ) -> (
        LayoutGenerationOutput
        | dict[
            str,
            Shaped[torch.Tensor, "..."]
            | dict[int, str]
            | Mapping[str, Shaped[torch.Tensor, "..."] | ConditionType | str | bool]
            | None,
        ]
    ):
        """Generate layouts from content images and saliency maps.

        Args:
            images: RGB image or batch. Required unless ``pixel_values`` is given.
            batch_size: Batch size used when ``pixel_values`` is supplied.
            seed: Convenience seed. Ignored when ``generator`` is supplied.
            generator: Explicit torch generator.
            condition_type: Must normalize to ``content_image``.
            labels: Optional public labels used only with ``bbox`` to provide a
                fixed initial layout.
            bbox: Optional public boxes used with ``labels`` for a fixed layout.
            mask: Optional valid-element mask for fixed initial layouts.
            num_elements: Reserved compatibility argument.
            box_format: Format of optional ``bbox``.
            normalized: Whether optional ``bbox`` is normalized.
            canvas_size: Pixel canvas size for optional unnormalized ``bbox``.
            num_inference_steps: Reserved compatibility argument.
            output_type: ``dataclass`` or ``dict``.
            return_intermediates: Whether to include raw model tensors.
            saliency: Optional single merged saliency map.
            saliency_pfpnet: Optional PFPNet saliency map.
            saliency_basnet: Optional BASNet saliency map.
            pixel_values: Preprocessed ``(B, 4, H, W)`` tensor.
            initial_layout: Optional internal layout ``(B, max_elem, 2, 4)``.

        Returns:
            Shared layout-generation output.

        Raises:
            ValueError: If the condition, image inputs, or fixed layout are invalid.
        """
        del num_elements, num_inference_steps
        canonical = normalize_condition_type(condition_type)
        resolved_output_type = normalize_output_type(output_type)
        device = self.device or next(self.model.parameters()).device
        if pixel_values is None:
            if images is None:
                raise ValueError("images or pixel_values are required for DS-GAN")

            encoded = self.processor(
                images,
                saliency=saliency,
                saliency_pfpnet=saliency_pfpnet,
                saliency_basnet=saliency_basnet,
            )
            pixel_values = cast(torch.Tensor, encoded["pixel_values"])
        pixel_values = pixel_values.to(device=device, dtype=self.model.dtype)
        batch_size = pixel_values.shape[0] if pixel_values is not None else batch_size
        if initial_layout is None and bbox is not None and labels is not None:
            encoded_layout = self.processor.encode_layout(
                bbox=bbox,
                labels=labels,
                mask=mask,
                box_format=box_format,
                normalized=normalized,
                canvas_size=canvas_size,
                max_elem=self.config.max_elem,
            )
            initial_layout = encoded_layout["layout"]
        if initial_layout is None:
            prepared = self.prepare_generator(generator=generator, seed=seed)
            initial_layout = random_initial_layout(
                batch_size,
                self.config.max_elem,
                generator=prepared,
                device=device,
                dtype=self.model.dtype,
            )
        else:
            initial_layout = initial_layout.to(device=device, dtype=self.model.dtype)
        model_output = self.model(
            pixel_values=pixel_values,
            layout=initial_layout,
            return_dict=True,
        )
        assert isinstance(model_output, DSGANModelOutput)
        intermediates = None
        if return_intermediates:
            intermediates = {
                "condition_type": canonical,
                "initial_layout": initial_layout.detach().cpu(),
                "class_probs": model_output.class_probs.detach().cpu(),
            }
        return self.processor.decode(
            class_probs=model_output.class_probs,
            bbox=model_output.bbox,
            output_type=resolved_output_type.value,
            intermediates=intermediates,
        )

__init__

__init__(
    model: DSGANModel,
    processor: DSGANProcessor | None = None,
    config: DSGANConfig | None = None,
    device: str | device | None = None,
) -> None

Initialize DS-GAN pipeline.

Source code in models/ds-gan/src/ds_gan/pipeline_ds_gan.py
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
def __init__(
    self,
    model: DSGANModel,
    processor: DSGANProcessor | None = None,
    config: DSGANConfig | None = None,
    device: str | torch.device | None = None,
) -> None:
    """Initialize DS-GAN pipeline."""
    super().__init__(config or model.config)
    self.config = config or model.config
    self.model = model
    self.processor = processor or DSGANProcessor(
        dataset_name=self.config.dataset_name,
        id2label=cast(dict[int | str, str], self.config.id2label),
        image_size=cast(tuple[int, int], self.config.image_size),
    )
    self.model.eval()
    if device is not None:
        self.to(device)

__call__

__call__(
    images: ImageInput
    | list[ImageInput]
    | Float[Tensor, "..."]
    | None = None,
    *,
    batch_size: int = 1,
    seed: int | None = None,
    generator: Generator | None = None,
    condition_type: ConditionType
    | str = ConditionType.content_image,
    labels: Int[Tensor, "batch elements"]
    | Sequence[ArrayLikeInput]
    | None = None,
    bbox: Float[Tensor, "batch elements 4"]
    | Sequence[ArrayLikeInput]
    | None = None,
    mask: Bool[Tensor, "batch elements"]
    | Sequence[ArrayLikeInput]
    | None = None,
    num_elements: int
    | list[int]
    | Int[Tensor, "batch"]
    | None = None,
    box_format: BoxFormat | str = BoxFormat.xywh,
    normalized: bool = True,
    canvas_size: tuple[int, int] | None = None,
    num_inference_steps: int | None = None,
    output_type: OutputType | str = OutputType.dataclass,
    return_intermediates: bool = False,
    saliency: ImageInput
    | list[ImageInput]
    | Float[Tensor, "..."]
    | None = None,
    saliency_pfpnet: ImageInput
    | list[ImageInput]
    | Float[Tensor, "..."]
    | None = None,
    saliency_basnet: ImageInput
    | list[ImageInput]
    | Float[Tensor, "..."]
    | None = None,
    pixel_values: Float[Tensor, "batch 4 height width"]
    | None = None,
    initial_layout: Float[Tensor, "batch elements 2 4"]
    | None = None,
) -> (
    LayoutGenerationOutput
    | dict[
        str,
        Shaped[torch.Tensor, "..."]
        | dict[int, str]
        | Mapping[
            str,
            Shaped[torch.Tensor, "..."]
            | ConditionType
            | str
            | bool,
        ]
        | None,
    ]
)

Generate layouts from content images and saliency maps.

Parameters:

Name Type Description Default
images ImageInput | list[ImageInput] | Float[Tensor, '...'] | None

RGB image or batch. Required unless pixel_values is given.

None
batch_size int

Batch size used when pixel_values is supplied.

1
seed int | None

Convenience seed. Ignored when generator is supplied.

None
generator Generator | None

Explicit torch generator.

None
condition_type ConditionType | str

Must normalize to content_image.

content_image
labels Int[Tensor, 'batch elements'] | Sequence[ArrayLikeInput] | None

Optional public labels used only with bbox to provide a fixed initial layout.

None
bbox Float[Tensor, 'batch elements 4'] | Sequence[ArrayLikeInput] | None

Optional public boxes used with labels for a fixed layout.

None
mask Bool[Tensor, 'batch elements'] | Sequence[ArrayLikeInput] | None

Optional valid-element mask for fixed initial layouts.

None
num_elements int | list[int] | Int[Tensor, 'batch'] | None

Reserved compatibility argument.

None
box_format BoxFormat | str

Format of optional bbox.

xywh
normalized bool

Whether optional bbox is normalized.

True
canvas_size tuple[int, int] | None

Pixel canvas size for optional unnormalized bbox.

None
num_inference_steps int | None

Reserved compatibility argument.

None
output_type OutputType | str

dataclass or dict.

dataclass
return_intermediates bool

Whether to include raw model tensors.

False
saliency ImageInput | list[ImageInput] | Float[Tensor, '...'] | None

Optional single merged saliency map.

None
saliency_pfpnet ImageInput | list[ImageInput] | Float[Tensor, '...'] | None

Optional PFPNet saliency map.

None
saliency_basnet ImageInput | list[ImageInput] | Float[Tensor, '...'] | None

Optional BASNet saliency map.

None
pixel_values Float[Tensor, 'batch 4 height width'] | None

Preprocessed (B, 4, H, W) tensor.

None
initial_layout Float[Tensor, 'batch elements 2 4'] | None

Optional internal layout (B, max_elem, 2, 4).

None

Returns:

Type Description
LayoutGenerationOutput | dict[str, Shaped[Tensor, '...'] | dict[int, str] | Mapping[str, Shaped[Tensor, '...'] | ConditionType | str | bool] | None]

Shared layout-generation output.

Raises:

Type Description
ValueError

If the condition, image inputs, or fixed layout are invalid.

Source code in models/ds-gan/src/ds_gan/pipeline_ds_gan.py
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
@torch.no_grad()
def __call__(  # ty: ignore[invalid-method-override]
    self,
    images: ImageInput
    | list[ImageInput]
    | Float[torch.Tensor, "..."]
    | None = None,
    *,
    batch_size: int = 1,
    seed: int | None = None,
    generator: torch.Generator | None = None,
    condition_type: ConditionType | str = ConditionType.content_image,
    labels: Int[torch.Tensor, "batch elements"]
    | Sequence[ArrayLikeInput]
    | None = None,
    bbox: Float[torch.Tensor, "batch elements 4"]
    | Sequence[ArrayLikeInput]
    | None = None,
    mask: Bool[torch.Tensor, "batch elements"]
    | Sequence[ArrayLikeInput]
    | None = None,
    num_elements: int | list[int] | Int[torch.Tensor, "batch"] | None = None,
    box_format: BoxFormat | str = BoxFormat.xywh,
    normalized: bool = True,
    canvas_size: tuple[int, int] | None = None,
    num_inference_steps: int | None = None,
    output_type: OutputType | str = OutputType.dataclass,
    return_intermediates: bool = False,
    saliency: ImageInput
    | list[ImageInput]
    | Float[torch.Tensor, "..."]
    | None = None,
    saliency_pfpnet: ImageInput
    | list[ImageInput]
    | Float[torch.Tensor, "..."]
    | None = None,
    saliency_basnet: ImageInput
    | list[ImageInput]
    | Float[torch.Tensor, "..."]
    | None = None,
    pixel_values: Float[torch.Tensor, "batch 4 height width"] | None = None,
    initial_layout: Float[torch.Tensor, "batch elements 2 4"] | None = None,
) -> (
    LayoutGenerationOutput
    | dict[
        str,
        Shaped[torch.Tensor, "..."]
        | dict[int, str]
        | Mapping[str, Shaped[torch.Tensor, "..."] | ConditionType | str | bool]
        | None,
    ]
):
    """Generate layouts from content images and saliency maps.

    Args:
        images: RGB image or batch. Required unless ``pixel_values`` is given.
        batch_size: Batch size used when ``pixel_values`` is supplied.
        seed: Convenience seed. Ignored when ``generator`` is supplied.
        generator: Explicit torch generator.
        condition_type: Must normalize to ``content_image``.
        labels: Optional public labels used only with ``bbox`` to provide a
            fixed initial layout.
        bbox: Optional public boxes used with ``labels`` for a fixed layout.
        mask: Optional valid-element mask for fixed initial layouts.
        num_elements: Reserved compatibility argument.
        box_format: Format of optional ``bbox``.
        normalized: Whether optional ``bbox`` is normalized.
        canvas_size: Pixel canvas size for optional unnormalized ``bbox``.
        num_inference_steps: Reserved compatibility argument.
        output_type: ``dataclass`` or ``dict``.
        return_intermediates: Whether to include raw model tensors.
        saliency: Optional single merged saliency map.
        saliency_pfpnet: Optional PFPNet saliency map.
        saliency_basnet: Optional BASNet saliency map.
        pixel_values: Preprocessed ``(B, 4, H, W)`` tensor.
        initial_layout: Optional internal layout ``(B, max_elem, 2, 4)``.

    Returns:
        Shared layout-generation output.

    Raises:
        ValueError: If the condition, image inputs, or fixed layout are invalid.
    """
    del num_elements, num_inference_steps
    canonical = normalize_condition_type(condition_type)
    resolved_output_type = normalize_output_type(output_type)
    device = self.device or next(self.model.parameters()).device
    if pixel_values is None:
        if images is None:
            raise ValueError("images or pixel_values are required for DS-GAN")

        encoded = self.processor(
            images,
            saliency=saliency,
            saliency_pfpnet=saliency_pfpnet,
            saliency_basnet=saliency_basnet,
        )
        pixel_values = cast(torch.Tensor, encoded["pixel_values"])
    pixel_values = pixel_values.to(device=device, dtype=self.model.dtype)
    batch_size = pixel_values.shape[0] if pixel_values is not None else batch_size
    if initial_layout is None and bbox is not None and labels is not None:
        encoded_layout = self.processor.encode_layout(
            bbox=bbox,
            labels=labels,
            mask=mask,
            box_format=box_format,
            normalized=normalized,
            canvas_size=canvas_size,
            max_elem=self.config.max_elem,
        )
        initial_layout = encoded_layout["layout"]
    if initial_layout is None:
        prepared = self.prepare_generator(generator=generator, seed=seed)
        initial_layout = random_initial_layout(
            batch_size,
            self.config.max_elem,
            generator=prepared,
            device=device,
            dtype=self.model.dtype,
        )
    else:
        initial_layout = initial_layout.to(device=device, dtype=self.model.dtype)
    model_output = self.model(
        pixel_values=pixel_values,
        layout=initial_layout,
        return_dict=True,
    )
    assert isinstance(model_output, DSGANModelOutput)
    intermediates = None
    if return_intermediates:
        intermediates = {
            "condition_type": canonical,
            "initial_layout": initial_layout.detach().cpu(),
            "class_probs": model_output.class_probs.detach().cpu(),
        }
    return self.processor.decode(
        class_probs=model_output.class_probs,
        bbox=model_output.bbox,
        output_type=resolved_output_type.value,
        intermediates=intermediates,
    )

normalize_condition_type

normalize_condition_type(
    condition_type: ConditionType | str | None,
) -> ConditionType

Normalize DS-GAN condition aliases.

Parameters:

Name Type Description Default
condition_type ConditionType | str | None

Canonical condition enum, alias, or None.

required

Returns:

Type Description
ConditionType

Canonical content_image condition.

Raises:

Type Description
ValueError

If DS-GAN does not support the requested mode.

Examples:

>>> str(normalize_condition_type("content"))
'content_image'
Source code in models/ds-gan/src/ds_gan/pipeline_ds_gan.py
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
def normalize_condition_type(
    condition_type: ConditionType | str | None,
) -> ConditionType:
    """Normalize DS-GAN condition aliases.

    Args:
        condition_type: Canonical condition enum, alias, or ``None``.

    Returns:
        Canonical ``content_image`` condition.

    Raises:
        ValueError: If DS-GAN does not support the requested mode.

    Examples:
        >>> str(normalize_condition_type("content"))
        'content_image'
    """
    if condition_type is None:
        canonical = ConditionType.content_image
    elif isinstance(condition_type, ConditionType):
        canonical = condition_type
    else:
        key = condition_type.lower().replace("-", "_")
        canonical = _DSGAN_CONDITION_ALIASES.get(key)
        if canonical is None:
            canonical = normalize_shared_condition_type(condition_type)
    if canonical not in _SUPPORTED_CONDITION_TYPES:
        raise ValueError(f"Unsupported DS-GAN condition_type: {condition_type}")

    return canonical

normalize_output_type

normalize_output_type(
    output_type: OutputType | str,
) -> OutputType

Normalize public output type aliases.

Source code in models/ds-gan/src/ds_gan/pipeline_ds_gan.py
85
86
87
88
89
90
91
92
def normalize_output_type(output_type: OutputType | str) -> OutputType:
    """Normalize public output type aliases."""
    if isinstance(output_type, OutputType):
        return output_type
    try:
        return OutputType(output_type)
    except ValueError as exc:
        raise ValueError(f"Unsupported output_type: {output_type}") from exc

processing_ds_gan

Processor for DS-GAN content-image inputs and layout decoding.

DSGANProcessor

Bases: ProcessorMixin

Prepare PosterLayout RGB/saliency inputs and decode DS-GAN outputs.

Parameters:

Name Type Description Default
dataset_name DatasetName | str

Dataset key. Only PKU PosterLayout is supported.

pku_posterlayout
id2label dict[int | str, str] | None

Public semantic labels excluding model no object.

None
image_size tuple[int, int] | list[int]

Resize target as (height, width).

(350, 240)

Examples:

>>> processor = DSGANProcessor()
>>> processor.id2label[0]
'text'
Source code in models/ds-gan/src/ds_gan/processing_ds_gan.py
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
class DSGANProcessor(ProcessorMixin):
    """Prepare PosterLayout RGB/saliency inputs and decode DS-GAN outputs.

    Args:
        dataset_name: Dataset key. Only PKU PosterLayout is supported.
        id2label: Public semantic labels excluding model ``no object``.
        image_size: Resize target as ``(height, width)``.

    Examples:
        >>> processor = DSGANProcessor()
        >>> processor.id2label[0]
        'text'
    """

    config_name = "processor_config.json"

    def __init__(
        self,
        dataset_name: DatasetName | str = DatasetName.pku_posterlayout,
        id2label: dict[int | str, str] | None = None,
        image_size: tuple[int, int] | list[int] = (350, 240),
    ) -> None:
        """Initialize processor metadata."""
        self.chat_template = None
        dataset = normalize_dataset_name(dataset_name)
        if dataset is not DatasetName.pku_posterlayout:
            raise ValueError(f"Unsupported DS-GAN dataset_name: {dataset_name}")

        self.dataset_name = str(dataset)
        default_id2label = {0: "text", 1: "logo", 2: "underlay"}
        raw_id2label = id2label or default_id2label
        self.id2label = {int(k): v for k, v in raw_id2label.items()}
        self.label2id = {v: k for k, v in self.id2label.items()}
        height, width = image_size
        self.image_size: tuple[int, int] = (int(height), int(width))

    def __call__(
        self,
        images: DSGANImageInput | Sequence[DSGANImageInput],
        *,
        saliency: DSGANImageInput | Sequence[DSGANImageInput] | None = None,
        saliency_pfpnet: DSGANImageInput | Sequence[DSGANImageInput] | None = None,
        saliency_basnet: DSGANImageInput | Sequence[DSGANImageInput] | None = None,
        return_tensors: Literal["pt"] = "pt",
    ) -> BatchEncoding:
        """Encode content images into ``pixel_values``.

        Args:
            images: RGB image or batch of RGB images.
            saliency: Optional saliency image or batch. If omitted and both
                saliency maps are given, the maps are merged by pixelwise max.
            saliency_pfpnet: Optional PFPNet saliency map.
            saliency_basnet: Optional BASNet saliency map.
            return_tensors: Tensor framework. Only ``pt`` is supported.

        Returns:
            Batch encoding containing ``pixel_values`` shaped ``(B, 4, H, W)``.
        """
        if return_tensors != "pt":
            raise ValueError("DSGANProcessor only supports return_tensors='pt'")

        image_rows = _ensure_batch(images)
        saliency_rows = self._resolve_saliency(
            len(image_rows),
            saliency=saliency,
            saliency_pfpnet=saliency_pfpnet,
            saliency_basnet=saliency_basnet,
        )
        tensors = []
        for image, sal in zip(image_rows, saliency_rows, strict=True):
            rgb = _to_rgb_tensor(image, self.image_size)
            sal_tensor = (
                torch.zeros(1, *self.image_size, dtype=torch.float32)
                if sal is None
                else _to_l_tensor(sal, self.image_size)
            )
            tensors.append(torch.cat((rgb, sal_tensor), dim=0))
        return BatchEncoding({"pixel_values": torch.stack(tensors)})

    def decode(
        self,
        *,
        class_probs: Float[torch.Tensor, "batch elements 4"],
        bbox: Float[torch.Tensor, "batch elements 4"],
        output_type: Literal["dataclass", "dict"] = "dataclass",
        scores: Float[torch.Tensor, "batch elements"] | None = None,
        intermediates: Mapping[str, Shaped[torch.Tensor, "..."] | str | bool]
        | None = None,
    ) -> (
        LayoutGenerationOutput
        | dict[
            str,
            Shaped[torch.Tensor, "..."]
            | dict[int, str]
            | Mapping[str, Shaped[torch.Tensor, "..."] | str | bool]
            | None,
        ]
    ):
        """Decode raw DS-GAN class probabilities and boxes.

        Args:
            class_probs: Internal class probabilities shaped ``(B, E, 4)``.
            bbox: Normalized center ``xywh`` boxes shaped ``(B, E, 4)``.
            output_type: Return format.
            scores: Optional per-element class scores.
            intermediates: Optional model-specific intermediate tensors.

        Returns:
            Shared layout output with public labels and mask semantics.
        """
        class_ids = torch.argmax(class_probs, dim=-1)
        mask = class_ids != 0
        public_labels = (class_ids - 1).clamp_min(0).long()
        resolved_scores = scores
        if resolved_scores is None:
            resolved_scores = class_probs.max(dim=-1).values
        output = LayoutGenerationOutput(
            bbox=bbox.detach().cpu().clamp(0.0, 1.0),
            labels=public_labels.detach().cpu(),
            mask=mask.detach().cpu(),
            id2label=dict(self.id2label),
            scores=resolved_scores.detach().cpu(),
            intermediates=intermediates,
        )
        if output_type == "dict":
            return dict(output)
        if output_type == "dataclass":
            return output
        raise ValueError(f"Unsupported output_type: {output_type}")

    def encode_layout(
        self,
        *,
        bbox: Float[torch.Tensor, "batch elements 4"]
        | Float[np.ndarray, "batch elements 4"]
        | Sequence[ArrayLikeInput],
        labels: Int[torch.Tensor, "batch elements"]
        | Int[np.ndarray, "batch elements"]
        | Sequence[ArrayLikeInput],
        mask: Bool[torch.Tensor, "batch elements"]
        | Bool[np.ndarray, "batch elements"]
        | Sequence[ArrayLikeInput]
        | None = None,
        box_format: BoxFormat | str = BoxFormat.xywh,
        normalized: bool = True,
        canvas_size: tuple[int, int] | None = None,
        max_elem: int = 32,
    ) -> dict[str, Shaped[torch.Tensor, "..."]]:
        """Encode public boxes/labels into the internal layout tensor.

        Args:
            bbox: Public boxes.
            labels: Public zero-based semantic labels.
            mask: Optional valid-element mask.
            box_format: Input box format.
            normalized: Whether the boxes are normalized.
            canvas_size: Pixel canvas size used when ``normalized=False``.
            max_elem: Output slot count.

        Returns:
            Dictionary with internal ``layout``, normalized ``bbox``, labels, and mask.
        """
        bbox_t, labels_t, mask_t = prepare_layout_tensors(
            bbox=bbox,
            labels=labels,
            mask=mask,
            box_format=box_format,
            normalized=normalized,
            canvas_size=canvas_size,
            clamp_converted_normalized=True,
        )
        bbox_t, labels_t, mask_t = self.pad(bbox_t, labels_t, mask_t, max_elem=max_elem)
        model_labels = torch.zeros_like(labels_t)
        model_labels[mask_t] = labels_t[mask_t] + 1
        class_one_hot = torch.nn.functional.one_hot(model_labels, num_classes=4).to(
            dtype=bbox_t.dtype
        )
        layout = torch.stack((class_one_hot, bbox_t), dim=2)
        return {"layout": layout, "bbox": bbox_t, "labels": labels_t, "mask": mask_t}

    def pad(
        self,
        bbox: Float[torch.Tensor, "batch elements 4"],
        labels: Int[torch.Tensor, "batch elements"],
        mask: Bool[torch.Tensor, "batch elements"],
        *,
        max_elem: int,
    ) -> tuple[
        Float[torch.Tensor, "batch padded_elements 4"],
        Int[torch.Tensor, "batch padded_elements"],
        Bool[torch.Tensor, "batch padded_elements"],
    ]:
        """Pad layout tensors to DS-GAN ``max_elem`` slots."""
        if bbox.shape[1] > max_elem:
            raise ValueError(f"DS-GAN supports at most {max_elem} elements")

        pad_count = max_elem - bbox.shape[1]
        if pad_count:
            bbox = torch.cat((bbox, torch.zeros(bbox.shape[0], pad_count, 4)), dim=1)
            labels = torch.cat(
                (labels, torch.zeros(labels.shape[0], pad_count, dtype=torch.long)),
                dim=1,
            )
            mask = torch.cat(
                (mask, torch.zeros(mask.shape[0], pad_count, dtype=torch.bool)),
                dim=1,
            )
        labels = labels.clone()
        labels[~mask] = 0
        return bbox, labels, mask

    def _resolve_saliency(
        self,
        batch_size: int,
        *,
        saliency: DSGANImageInput | Sequence[DSGANImageInput] | None,
        saliency_pfpnet: DSGANImageInput | Sequence[DSGANImageInput] | None,
        saliency_basnet: DSGANImageInput | Sequence[DSGANImageInput] | None,
    ) -> list[DSGANImageInput | Float[torch.Tensor, "1 height width"] | None]:
        if saliency is not None:
            rows = _ensure_batch(saliency)
            if len(rows) != batch_size:
                raise ValueError("saliency batch size must match images")

            return cast(
                list[DSGANImageInput | Float[torch.Tensor, "1 height width"] | None],
                rows,
            )
        if saliency_pfpnet is None and saliency_basnet is None:
            return [None] * batch_size
        first = (
            _ensure_batch(saliency_pfpnet)
            if saliency_pfpnet is not None
            else [None] * batch_size
        )
        second = (
            _ensure_batch(saliency_basnet)
            if saliency_basnet is not None
            else [None] * batch_size
        )
        if len(first) != batch_size or len(second) != batch_size:
            raise ValueError("saliency batch size must match images")

        merged: list[Float[torch.Tensor, "1 height width"]] = []
        for left, right in zip(first, second, strict=True):
            if left is None:
                merged.append(_to_l_tensor(right, self.image_size))
            elif right is None:
                merged.append(_to_l_tensor(left, self.image_size))
            else:
                merged.append(_merge_saliency_native(left, right, self.image_size))
        return cast(
            list[DSGANImageInput | Float[torch.Tensor, "1 height width"] | None], merged
        )

__init__

__init__(
    dataset_name: DatasetName
    | str = DatasetName.pku_posterlayout,
    id2label: dict[int | str, str] | None = None,
    image_size: tuple[int, int] | list[int] = (350, 240),
) -> None

Initialize processor metadata.

Source code in models/ds-gan/src/ds_gan/processing_ds_gan.py
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
def __init__(
    self,
    dataset_name: DatasetName | str = DatasetName.pku_posterlayout,
    id2label: dict[int | str, str] | None = None,
    image_size: tuple[int, int] | list[int] = (350, 240),
) -> None:
    """Initialize processor metadata."""
    self.chat_template = None
    dataset = normalize_dataset_name(dataset_name)
    if dataset is not DatasetName.pku_posterlayout:
        raise ValueError(f"Unsupported DS-GAN dataset_name: {dataset_name}")

    self.dataset_name = str(dataset)
    default_id2label = {0: "text", 1: "logo", 2: "underlay"}
    raw_id2label = id2label or default_id2label
    self.id2label = {int(k): v for k, v in raw_id2label.items()}
    self.label2id = {v: k for k, v in self.id2label.items()}
    height, width = image_size
    self.image_size: tuple[int, int] = (int(height), int(width))

__call__

__call__(
    images: DSGANImageInput | Sequence[DSGANImageInput],
    *,
    saliency: DSGANImageInput
    | Sequence[DSGANImageInput]
    | None = None,
    saliency_pfpnet: DSGANImageInput
    | Sequence[DSGANImageInput]
    | None = None,
    saliency_basnet: DSGANImageInput
    | Sequence[DSGANImageInput]
    | None = None,
    return_tensors: Literal["pt"] = "pt",
) -> BatchEncoding

Encode content images into pixel_values.

Parameters:

Name Type Description Default
images DSGANImageInput | Sequence[DSGANImageInput]

RGB image or batch of RGB images.

required
saliency DSGANImageInput | Sequence[DSGANImageInput] | None

Optional saliency image or batch. If omitted and both saliency maps are given, the maps are merged by pixelwise max.

None
saliency_pfpnet DSGANImageInput | Sequence[DSGANImageInput] | None

Optional PFPNet saliency map.

None
saliency_basnet DSGANImageInput | Sequence[DSGANImageInput] | None

Optional BASNet saliency map.

None
return_tensors Literal['pt']

Tensor framework. Only pt is supported.

'pt'

Returns:

Type Description
BatchEncoding

Batch encoding containing pixel_values shaped (B, 4, H, W).

Source code in models/ds-gan/src/ds_gan/processing_ds_gan.py
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
def __call__(
    self,
    images: DSGANImageInput | Sequence[DSGANImageInput],
    *,
    saliency: DSGANImageInput | Sequence[DSGANImageInput] | None = None,
    saliency_pfpnet: DSGANImageInput | Sequence[DSGANImageInput] | None = None,
    saliency_basnet: DSGANImageInput | Sequence[DSGANImageInput] | None = None,
    return_tensors: Literal["pt"] = "pt",
) -> BatchEncoding:
    """Encode content images into ``pixel_values``.

    Args:
        images: RGB image or batch of RGB images.
        saliency: Optional saliency image or batch. If omitted and both
            saliency maps are given, the maps are merged by pixelwise max.
        saliency_pfpnet: Optional PFPNet saliency map.
        saliency_basnet: Optional BASNet saliency map.
        return_tensors: Tensor framework. Only ``pt`` is supported.

    Returns:
        Batch encoding containing ``pixel_values`` shaped ``(B, 4, H, W)``.
    """
    if return_tensors != "pt":
        raise ValueError("DSGANProcessor only supports return_tensors='pt'")

    image_rows = _ensure_batch(images)
    saliency_rows = self._resolve_saliency(
        len(image_rows),
        saliency=saliency,
        saliency_pfpnet=saliency_pfpnet,
        saliency_basnet=saliency_basnet,
    )
    tensors = []
    for image, sal in zip(image_rows, saliency_rows, strict=True):
        rgb = _to_rgb_tensor(image, self.image_size)
        sal_tensor = (
            torch.zeros(1, *self.image_size, dtype=torch.float32)
            if sal is None
            else _to_l_tensor(sal, self.image_size)
        )
        tensors.append(torch.cat((rgb, sal_tensor), dim=0))
    return BatchEncoding({"pixel_values": torch.stack(tensors)})

decode

decode(
    *,
    class_probs: Float[Tensor, "batch elements 4"],
    bbox: Float[Tensor, "batch elements 4"],
    output_type: Literal["dataclass", "dict"] = "dataclass",
    scores: Float[Tensor, "batch elements"] | None = None,
    intermediates: Mapping[
        str, Shaped[Tensor, "..."] | str | bool
    ]
    | None = None,
) -> (
    LayoutGenerationOutput
    | dict[
        str,
        Shaped[torch.Tensor, "..."]
        | dict[int, str]
        | Mapping[
            str, Shaped[torch.Tensor, "..."] | str | bool
        ]
        | None,
    ]
)

Decode raw DS-GAN class probabilities and boxes.

Parameters:

Name Type Description Default
class_probs Float[Tensor, 'batch elements 4']

Internal class probabilities shaped (B, E, 4).

required
bbox Float[Tensor, 'batch elements 4']

Normalized center xywh boxes shaped (B, E, 4).

required
output_type Literal['dataclass', 'dict']

Return format.

'dataclass'
scores Float[Tensor, 'batch elements'] | None

Optional per-element class scores.

None
intermediates Mapping[str, Shaped[Tensor, '...'] | str | bool] | None

Optional model-specific intermediate tensors.

None

Returns:

Type Description
LayoutGenerationOutput | dict[str, Shaped[Tensor, '...'] | dict[int, str] | Mapping[str, Shaped[Tensor, '...'] | str | bool] | None]

Shared layout output with public labels and mask semantics.

Source code in models/ds-gan/src/ds_gan/processing_ds_gan.py
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
def decode(
    self,
    *,
    class_probs: Float[torch.Tensor, "batch elements 4"],
    bbox: Float[torch.Tensor, "batch elements 4"],
    output_type: Literal["dataclass", "dict"] = "dataclass",
    scores: Float[torch.Tensor, "batch elements"] | None = None,
    intermediates: Mapping[str, Shaped[torch.Tensor, "..."] | str | bool]
    | None = None,
) -> (
    LayoutGenerationOutput
    | dict[
        str,
        Shaped[torch.Tensor, "..."]
        | dict[int, str]
        | Mapping[str, Shaped[torch.Tensor, "..."] | str | bool]
        | None,
    ]
):
    """Decode raw DS-GAN class probabilities and boxes.

    Args:
        class_probs: Internal class probabilities shaped ``(B, E, 4)``.
        bbox: Normalized center ``xywh`` boxes shaped ``(B, E, 4)``.
        output_type: Return format.
        scores: Optional per-element class scores.
        intermediates: Optional model-specific intermediate tensors.

    Returns:
        Shared layout output with public labels and mask semantics.
    """
    class_ids = torch.argmax(class_probs, dim=-1)
    mask = class_ids != 0
    public_labels = (class_ids - 1).clamp_min(0).long()
    resolved_scores = scores
    if resolved_scores is None:
        resolved_scores = class_probs.max(dim=-1).values
    output = LayoutGenerationOutput(
        bbox=bbox.detach().cpu().clamp(0.0, 1.0),
        labels=public_labels.detach().cpu(),
        mask=mask.detach().cpu(),
        id2label=dict(self.id2label),
        scores=resolved_scores.detach().cpu(),
        intermediates=intermediates,
    )
    if output_type == "dict":
        return dict(output)
    if output_type == "dataclass":
        return output
    raise ValueError(f"Unsupported output_type: {output_type}")

encode_layout

encode_layout(
    *,
    bbox: Float[Tensor, "batch elements 4"]
    | Float[ndarray, "batch elements 4"]
    | Sequence[ArrayLikeInput],
    labels: Int[Tensor, "batch elements"]
    | Int[ndarray, "batch elements"]
    | Sequence[ArrayLikeInput],
    mask: Bool[Tensor, "batch elements"]
    | Bool[ndarray, "batch elements"]
    | Sequence[ArrayLikeInput]
    | None = None,
    box_format: BoxFormat | str = BoxFormat.xywh,
    normalized: bool = True,
    canvas_size: tuple[int, int] | None = None,
    max_elem: int = 32,
) -> dict[str, Shaped[torch.Tensor, "..."]]

Encode public boxes/labels into the internal layout tensor.

Parameters:

Name Type Description Default
bbox Float[Tensor, 'batch elements 4'] | Float[ndarray, 'batch elements 4'] | Sequence[ArrayLikeInput]

Public boxes.

required
labels Int[Tensor, 'batch elements'] | Int[ndarray, 'batch elements'] | Sequence[ArrayLikeInput]

Public zero-based semantic labels.

required
mask Bool[Tensor, 'batch elements'] | Bool[ndarray, 'batch elements'] | Sequence[ArrayLikeInput] | None

Optional valid-element mask.

None
box_format BoxFormat | str

Input box format.

xywh
normalized bool

Whether the boxes are normalized.

True
canvas_size tuple[int, int] | None

Pixel canvas size used when normalized=False.

None
max_elem int

Output slot count.

32

Returns:

Type Description
dict[str, Shaped[Tensor, '...']]

Dictionary with internal layout, normalized bbox, labels, and mask.

Source code in models/ds-gan/src/ds_gan/processing_ds_gan.py
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
def encode_layout(
    self,
    *,
    bbox: Float[torch.Tensor, "batch elements 4"]
    | Float[np.ndarray, "batch elements 4"]
    | Sequence[ArrayLikeInput],
    labels: Int[torch.Tensor, "batch elements"]
    | Int[np.ndarray, "batch elements"]
    | Sequence[ArrayLikeInput],
    mask: Bool[torch.Tensor, "batch elements"]
    | Bool[np.ndarray, "batch elements"]
    | Sequence[ArrayLikeInput]
    | None = None,
    box_format: BoxFormat | str = BoxFormat.xywh,
    normalized: bool = True,
    canvas_size: tuple[int, int] | None = None,
    max_elem: int = 32,
) -> dict[str, Shaped[torch.Tensor, "..."]]:
    """Encode public boxes/labels into the internal layout tensor.

    Args:
        bbox: Public boxes.
        labels: Public zero-based semantic labels.
        mask: Optional valid-element mask.
        box_format: Input box format.
        normalized: Whether the boxes are normalized.
        canvas_size: Pixel canvas size used when ``normalized=False``.
        max_elem: Output slot count.

    Returns:
        Dictionary with internal ``layout``, normalized ``bbox``, labels, and mask.
    """
    bbox_t, labels_t, mask_t = prepare_layout_tensors(
        bbox=bbox,
        labels=labels,
        mask=mask,
        box_format=box_format,
        normalized=normalized,
        canvas_size=canvas_size,
        clamp_converted_normalized=True,
    )
    bbox_t, labels_t, mask_t = self.pad(bbox_t, labels_t, mask_t, max_elem=max_elem)
    model_labels = torch.zeros_like(labels_t)
    model_labels[mask_t] = labels_t[mask_t] + 1
    class_one_hot = torch.nn.functional.one_hot(model_labels, num_classes=4).to(
        dtype=bbox_t.dtype
    )
    layout = torch.stack((class_one_hot, bbox_t), dim=2)
    return {"layout": layout, "bbox": bbox_t, "labels": labels_t, "mask": mask_t}

pad

pad(
    bbox: Float[Tensor, "batch elements 4"],
    labels: Int[Tensor, "batch elements"],
    mask: Bool[Tensor, "batch elements"],
    *,
    max_elem: int,
) -> tuple[
    Float[torch.Tensor, "batch padded_elements 4"],
    Int[torch.Tensor, "batch padded_elements"],
    Bool[torch.Tensor, "batch padded_elements"],
]

Pad layout tensors to DS-GAN max_elem slots.

Source code in models/ds-gan/src/ds_gan/processing_ds_gan.py
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
def pad(
    self,
    bbox: Float[torch.Tensor, "batch elements 4"],
    labels: Int[torch.Tensor, "batch elements"],
    mask: Bool[torch.Tensor, "batch elements"],
    *,
    max_elem: int,
) -> tuple[
    Float[torch.Tensor, "batch padded_elements 4"],
    Int[torch.Tensor, "batch padded_elements"],
    Bool[torch.Tensor, "batch padded_elements"],
]:
    """Pad layout tensors to DS-GAN ``max_elem`` slots."""
    if bbox.shape[1] > max_elem:
        raise ValueError(f"DS-GAN supports at most {max_elem} elements")

    pad_count = max_elem - bbox.shape[1]
    if pad_count:
        bbox = torch.cat((bbox, torch.zeros(bbox.shape[0], pad_count, 4)), dim=1)
        labels = torch.cat(
            (labels, torch.zeros(labels.shape[0], pad_count, dtype=torch.long)),
            dim=1,
        )
        mask = torch.cat(
            (mask, torch.zeros(mask.shape[0], pad_count, dtype=torch.bool)),
            dim=1,
        )
    labels = labels.clone()
    labels[~mask] = 0
    return bbox, labels, mask

processor_for_dataset

processor_for_dataset(
    dataset_name: DatasetName | str,
) -> DSGANProcessor

Create a DS-GAN processor for a supported dataset.

Source code in models/ds-gan/src/ds_gan/processing_ds_gan.py
308
309
310
def processor_for_dataset(dataset_name: DatasetName | str) -> DSGANProcessor:
    """Create a DS-GAN processor for a supported dataset."""
    return DSGANProcessor(dataset_name=dataset_name)

annotations_from_pku_example

annotations_from_pku_example(
    example: Mapping[str, DSGANExampleValue],
    *,
    max_elem: int = 32,
    include_unconnected_underlays: bool = True,
    canvas_size: tuple[int, int] | None = None,
    vendor_box_fix: bool = False,
) -> dict[
    str, Shaped[torch.Tensor, "..."] | tuple[int, int]
]

Convert a PKU PosterLayout dataset row into public layout tensors.

The adapter filters INVALID annotations, converts pixel ltrb boxes to normalized center xywh, derives canvas size from the image columns, and applies the reference designSeq.reorder ordering policy. Set vendor_box_fix to reproduce the reference loader's overlapping-slice assignment for inverted boxes.

Source code in models/ds-gan/src/ds_gan/processing_ds_gan.py
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
def annotations_from_pku_example(
    example: Mapping[str, DSGANExampleValue],
    *,
    max_elem: int = 32,
    include_unconnected_underlays: bool = True,
    canvas_size: tuple[int, int] | None = None,
    vendor_box_fix: bool = False,
) -> dict[str, Shaped[torch.Tensor, "..."] | tuple[int, int]]:
    """Convert a PKU PosterLayout dataset row into public layout tensors.

    The adapter filters ``INVALID`` annotations, converts pixel ``ltrb`` boxes
    to normalized center ``xywh``, derives canvas size from the image columns,
    and applies the reference ``designSeq.reorder`` ordering policy. Set
    ``vendor_box_fix`` to reproduce the reference loader's overlapping-slice
    assignment for inverted boxes.
    """
    annotations = cast(
        Mapping[
            str, Sequence[str | int | float] | Sequence[Sequence[str | int | float]]
        ],
        example.get("annotations", example),
    )
    raw_labels = cast(Sequence[str | int | float], annotations["cls_elem"])
    raw_boxes = cast(
        Sequence[str | Sequence[int | float | str]], annotations["box_elem"]
    )
    resolved_canvas_size = canvas_size or _canvas_size_from_example(example)
    model_labels: list[int] = []
    public_labels: list[int] = []
    boxes: list[list[float]] = []

    for raw_label, raw_box in zip(raw_labels, raw_boxes, strict=True):
        if isinstance(raw_label, (int, float)):
            numeric_label = int(raw_label)
            if numeric_label != raw_label or numeric_label not in PKU_SOURCE_LABELS:
                raise ValueError(f"Unsupported PKU source label: {raw_label!r}")

            label = PKU_SOURCE_LABELS[numeric_label]
        else:
            label = raw_label

        if label == "INVALID":
            continue
        public_id = PKU_DATASET_LABEL2ID[label]
        model_labels.append(PKU_MODEL_LABEL2ID[label])
        public_labels.append(public_id)
        boxes.append(_parse_box(raw_box, sort=not vendor_box_fix))

    if boxes:
        box_t = torch.tensor(boxes, dtype=torch.float32)
        ordering_boxes = torch.tensor(
            boxes,
            dtype=(
                torch.int64
                if all(
                    not isinstance(value, float) or value.is_integer()
                    for box in boxes
                    for value in box
                )
                else torch.float32
            ),
        )
        order = _designseq_reorder(
            model_labels,
            ordering_boxes,
            max_elem=max_elem,
            include_unconnected_underlays=include_unconnected_underlays,
        )
        box_t = box_t[torch.tensor(order, dtype=torch.long)]
        labels_t = torch.tensor([public_labels[i] for i in order], dtype=torch.long)
        if vendor_box_fix:
            inverted = (box_t[:, 0] > box_t[:, 2]) | (box_t[:, 1] > box_t[:, 3])
            if inverted.any():
                box_t[inverted, :2] = box_t[inverted, 2:]
                box_t[inverted, 2:] = box_t[inverted, :2]
        else:
            left = torch.minimum(box_t[:, 0], box_t[:, 2])
            top = torch.minimum(box_t[:, 1], box_t[:, 3])
            right = torch.maximum(box_t[:, 0], box_t[:, 2])
            bottom = torch.maximum(box_t[:, 1], box_t[:, 3])
            box_t = torch.stack((left, top, right, bottom), dim=-1)
        bbox_t = _normalize_vendor_ltrb(box_t, resolved_canvas_size)
    else:
        bbox_t = torch.zeros(0, 4, dtype=torch.float32)
        labels_t = torch.zeros(0, dtype=torch.long)
    mask_t = torch.ones(labels_t.shape, dtype=torch.bool)
    return {
        "bbox": bbox_t.unsqueeze(0),
        "labels": labels_t.unsqueeze(0),
        "mask": mask_t.unsqueeze(0),
        "canvas_size": resolved_canvas_size,
    }

training

Package-local training components for DS-GAN.

DSGANDataModule

Bases: LightningDataModule

Prepare training and test streams with the reference transforms.

Source code in models/ds-gan/src/ds_gan/training/datamodule.py
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
class DSGANDataModule(LightningDataModule):
    """Prepare training and test streams with the reference transforms."""

    def __init__(
        self,
        *,
        dataset_name: str = "pku_posterlayout",
        cache_dir: str | None = None,
        batch_size: int = 128,
        test_batch_size: int = 4,
        max_elem: int = 32,
        num_workers: int = 0,
        seed: int = 0,
    ) -> None:
        """Initialize the approved-source loader settings."""
        super().__init__()
        if dataset_name not in {"pku_posterlayout", "PKU-PosterLayout"}:
            raise ValueError(f"unsupported DS-GAN dataset: {dataset_name}")

        self.cache_dir = cache_dir
        self.batch_size = batch_size
        self.test_batch_size = test_batch_size
        self.max_elem = max_elem
        self.num_workers = num_workers
        self.seed = seed
        self.processor = DSGANProcessor()
        self.dataset: DatasetDict | None = None
        self._train_generator: torch.Generator | None = None

    def setup(self, stage: str | None = None) -> None:
        """Open cached source rows for fitting or evaluation."""
        if stage in {None, "fit", "test"}:
            self.dataset = load_cached_dataset(self.cache_dir)

    def train_dataloader(self) -> DataLoader[dict[str, Shaped[torch.Tensor, "..."]]]:
        """Return the reference-compatible shuffled training stream."""
        if self.dataset is None:
            self.setup("fit")

        assert self.dataset is not None
        dataset = DSGANDataset(
            self.dataset["train"],
            split="train",
            processor=self.processor,
            max_elem=self.max_elem,
        )
        generator = self.training_generator
        return DataLoader(
            dataset,
            batch_size=self.batch_size,
            sampler=_VendorRandomSampler(dataset, generator),
            num_workers=self.num_workers,
            generator=generator,
        )

    def test_dataloader(self) -> DataLoader[dict[str, Shaped[torch.Tensor, "..."]]]:
        """Return the complete unshuffled test stream."""
        if self.dataset is None:
            self.setup("test")

        assert self.dataset is not None
        return DataLoader(
            DSGANDataset(
                self.dataset["test"],
                split="test",
                processor=self.processor,
                max_elem=self.max_elem,
            ),
            batch_size=self.test_batch_size,
            shuffle=False,
            num_workers=self.num_workers,
        )

    @property
    def training_generator(self) -> torch.Generator:
        """Return the shared sampler and layout-initialization RNG stream."""
        if self._train_generator is None:
            self._train_generator = torch.Generator(device="cpu")
            self._train_generator.manual_seed(self.seed)

        return self._train_generator

training_generator property

training_generator: Generator

Return the shared sampler and layout-initialization RNG stream.

__init__

__init__(
    *,
    dataset_name: str = "pku_posterlayout",
    cache_dir: str | None = None,
    batch_size: int = 128,
    test_batch_size: int = 4,
    max_elem: int = 32,
    num_workers: int = 0,
    seed: int = 0,
) -> None

Initialize the approved-source loader settings.

Source code in models/ds-gan/src/ds_gan/training/datamodule.py
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
def __init__(
    self,
    *,
    dataset_name: str = "pku_posterlayout",
    cache_dir: str | None = None,
    batch_size: int = 128,
    test_batch_size: int = 4,
    max_elem: int = 32,
    num_workers: int = 0,
    seed: int = 0,
) -> None:
    """Initialize the approved-source loader settings."""
    super().__init__()
    if dataset_name not in {"pku_posterlayout", "PKU-PosterLayout"}:
        raise ValueError(f"unsupported DS-GAN dataset: {dataset_name}")

    self.cache_dir = cache_dir
    self.batch_size = batch_size
    self.test_batch_size = test_batch_size
    self.max_elem = max_elem
    self.num_workers = num_workers
    self.seed = seed
    self.processor = DSGANProcessor()
    self.dataset: DatasetDict | None = None
    self._train_generator: torch.Generator | None = None

setup

setup(stage: str | None = None) -> None

Open cached source rows for fitting or evaluation.

Source code in models/ds-gan/src/ds_gan/training/datamodule.py
66
67
68
69
def setup(self, stage: str | None = None) -> None:
    """Open cached source rows for fitting or evaluation."""
    if stage in {None, "fit", "test"}:
        self.dataset = load_cached_dataset(self.cache_dir)

train_dataloader

train_dataloader() -> DataLoader[
    dict[str, Shaped[torch.Tensor, "..."]]
]

Return the reference-compatible shuffled training stream.

Source code in models/ds-gan/src/ds_gan/training/datamodule.py
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
def train_dataloader(self) -> DataLoader[dict[str, Shaped[torch.Tensor, "..."]]]:
    """Return the reference-compatible shuffled training stream."""
    if self.dataset is None:
        self.setup("fit")

    assert self.dataset is not None
    dataset = DSGANDataset(
        self.dataset["train"],
        split="train",
        processor=self.processor,
        max_elem=self.max_elem,
    )
    generator = self.training_generator
    return DataLoader(
        dataset,
        batch_size=self.batch_size,
        sampler=_VendorRandomSampler(dataset, generator),
        num_workers=self.num_workers,
        generator=generator,
    )

test_dataloader

test_dataloader() -> DataLoader[
    dict[str, Shaped[torch.Tensor, "..."]]
]

Return the complete unshuffled test stream.

Source code in models/ds-gan/src/ds_gan/training/datamodule.py
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
def test_dataloader(self) -> DataLoader[dict[str, Shaped[torch.Tensor, "..."]]]:
    """Return the complete unshuffled test stream."""
    if self.dataset is None:
        self.setup("test")

    assert self.dataset is not None
    return DataLoader(
        DSGANDataset(
            self.dataset["test"],
            split="test",
            processor=self.processor,
            max_elem=self.max_elem,
        ),
        batch_size=self.test_batch_size,
        shuffle=False,
        num_workers=self.num_workers,
    )

DSGANDataset

Bases: Dataset[dict[str, Shaped[Tensor, '...']]]

Lazy DS-GAN dataset using the approved source and reference preprocessing.

Source code in models/ds-gan/src/ds_gan/training/dataset.py
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
class DSGANDataset(TorchDataset[dict[str, Shaped[torch.Tensor, "..."]]]):
    """Lazy DS-GAN dataset using the approved source and reference preprocessing."""

    def __init__(
        self,
        rows: Dataset,
        *,
        split: str,
        processor: DSGANProcessor | None = None,
        max_elem: int = 32,
    ) -> None:
        """Initialize the source bridge."""
        self.rows = rows
        self.split = split
        self.processor = processor or DSGANProcessor()
        self.max_elem = max_elem

    def __len__(self) -> int:
        """Return the number of source rows."""
        return len(self.rows)

    def __getitem__(self, index: int) -> dict[str, Shaped[torch.Tensor, "..."]]:
        """Convert one source row into model inputs and padded labels."""
        source = bridge_example(
            cast(Mapping[str, DatasetValue], self.rows[index]), split=self.split
        )
        image = cast(
            DSGANImageInput,
            source["canvas"] if self.split == "test" else source["inpainted_poster"],
        )
        encoded = self.processor(
            image,
            saliency_pfpnet=cast(DSGANImageInput, source["saliency_pfpnet"]),
            saliency_basnet=cast(DSGANImageInput, source["saliency_basnet"]),
        )
        pixel_values = encoded["pixel_values"][0]
        if self.split == "test":
            return {"pixel_values": pixel_values}

        layout_fields = annotations_from_pku_example(
            cast(
                Mapping[str, DSGANExampleValue],
                {
                    "annotations": cast(DSGANExampleValue, source["annotations"]),
                    "canvas": image,
                    "inpainted_poster": image,
                },
            ),
            max_elem=self.max_elem,
            include_unconnected_underlays=False,
            canvas_size=VENDOR_CANVAS_SIZE,
            vendor_box_fix=True,
        )
        bbox = cast(torch.Tensor, layout_fields["bbox"]).squeeze(0)
        labels = cast(torch.Tensor, layout_fields["labels"]).squeeze(0)
        mask = cast(torch.Tensor, layout_fields["mask"]).squeeze(0)
        padded_bbox, padded_labels, padded_mask = self.processor.pad(
            bbox.unsqueeze(0),
            labels.unsqueeze(0),
            mask.unsqueeze(0),
            max_elem=self.max_elem,
        )
        model_labels = torch.zeros_like(padded_labels)
        model_labels[padded_mask] = padded_labels[padded_mask] + 1
        classes = torch.nn.functional.one_hot(model_labels, num_classes=4).to(
            dtype=padded_bbox.dtype
        )
        layout = torch.stack((classes, padded_bbox), dim=2).squeeze(0)
        return {
            "pixel_values": pixel_values,
            "layout": layout,
            "labels": model_labels.squeeze(0),
            "boxes": padded_bbox.squeeze(0),
            "mask": padded_mask.squeeze(0),
        }

__init__

__init__(
    rows: Dataset,
    *,
    split: str,
    processor: DSGANProcessor | None = None,
    max_elem: int = 32,
) -> None

Initialize the source bridge.

Source code in models/ds-gan/src/ds_gan/training/dataset.py
81
82
83
84
85
86
87
88
89
90
91
92
93
def __init__(
    self,
    rows: Dataset,
    *,
    split: str,
    processor: DSGANProcessor | None = None,
    max_elem: int = 32,
) -> None:
    """Initialize the source bridge."""
    self.rows = rows
    self.split = split
    self.processor = processor or DSGANProcessor()
    self.max_elem = max_elem

__len__

__len__() -> int

Return the number of source rows.

Source code in models/ds-gan/src/ds_gan/training/dataset.py
95
96
97
def __len__(self) -> int:
    """Return the number of source rows."""
    return len(self.rows)

__getitem__

__getitem__(
    index: int,
) -> dict[str, Shaped[torch.Tensor, "..."]]

Convert one source row into model inputs and padded labels.

Source code in models/ds-gan/src/ds_gan/training/dataset.py
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
def __getitem__(self, index: int) -> dict[str, Shaped[torch.Tensor, "..."]]:
    """Convert one source row into model inputs and padded labels."""
    source = bridge_example(
        cast(Mapping[str, DatasetValue], self.rows[index]), split=self.split
    )
    image = cast(
        DSGANImageInput,
        source["canvas"] if self.split == "test" else source["inpainted_poster"],
    )
    encoded = self.processor(
        image,
        saliency_pfpnet=cast(DSGANImageInput, source["saliency_pfpnet"]),
        saliency_basnet=cast(DSGANImageInput, source["saliency_basnet"]),
    )
    pixel_values = encoded["pixel_values"][0]
    if self.split == "test":
        return {"pixel_values": pixel_values}

    layout_fields = annotations_from_pku_example(
        cast(
            Mapping[str, DSGANExampleValue],
            {
                "annotations": cast(DSGANExampleValue, source["annotations"]),
                "canvas": image,
                "inpainted_poster": image,
            },
        ),
        max_elem=self.max_elem,
        include_unconnected_underlays=False,
        canvas_size=VENDOR_CANVAS_SIZE,
        vendor_box_fix=True,
    )
    bbox = cast(torch.Tensor, layout_fields["bbox"]).squeeze(0)
    labels = cast(torch.Tensor, layout_fields["labels"]).squeeze(0)
    mask = cast(torch.Tensor, layout_fields["mask"]).squeeze(0)
    padded_bbox, padded_labels, padded_mask = self.processor.pad(
        bbox.unsqueeze(0),
        labels.unsqueeze(0),
        mask.unsqueeze(0),
        max_elem=self.max_elem,
    )
    model_labels = torch.zeros_like(padded_labels)
    model_labels[padded_mask] = padded_labels[padded_mask] + 1
    classes = torch.nn.functional.one_hot(model_labels, num_classes=4).to(
        dtype=padded_bbox.dtype
    )
    layout = torch.stack((classes, padded_bbox), dim=2).squeeze(0)
    return {
        "pixel_values": pixel_values,
        "layout": layout,
        "labels": model_labels.squeeze(0),
        "boxes": padded_bbox.squeeze(0),
        "mask": padded_mask.squeeze(0),
    }

DSGANTrainingModule

Bases: LightningModule

Train generator and discriminator with the reference update ordering.

Source code in models/ds-gan/src/ds_gan/training/lightning_module.py
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
class DSGANTrainingModule(LightningModule):
    """Train generator and discriminator with the reference update ordering."""

    def __init__(
        self,
        *,
        config: TrainingConfigInput,
        discriminator_config: TrainingConfigInput | None = None,
        generator: DSGANModel | None = None,
        discriminator: DSGANDiscriminator | None = None,
        generator_backbone_weights: str | None = None,
        discriminator_backbone_weights: str | None = None,
        generator_learning_rate: float = 1e-4,
        discriminator_learning_rate: float = 1e-3,
        generator_backbone_learning_rate: float = 1e-5,
        discriminator_backbone_learning_rate: float = 1e-4,
        adversarial_weight_ramp_steps: int = 100,
        scheduler_gamma: float = 0.8,
        scheduler_generator_milestones: tuple[int, ...] = (0, 50, 100, 150, 200, 250),
        scheduler_discriminator_milestones: tuple[int, ...] = (
            0,
            25,
            50,
            75,
            100,
            125,
            150,
            175,
            200,
            225,
            250,
            275,
        ),
        seed: int = 0,
    ) -> None:
        """Initialize the two-network DS-GAN training state."""
        super().__init__()
        resolved_config = _resolve_config(config)
        resolved_discriminator_config = (
            _resolve_config(discriminator_config)
            if discriminator_config is not None
            else None
        )
        self.save_hyperparameters(
            ignore=(
                "generator",
                "discriminator",
                "generator_backbone_weights",
                "discriminator_backbone_weights",
            )
        )
        self.ds_gan_config = resolved_config

        if resolved_discriminator_config is None:
            resolved_discriminator_config = DSGANConfig(
                dataset_name=resolved_config.dataset_name,
                backbone="resnet18",
                max_elem=resolved_config.max_elem,
                in_channels=resolved_config.in_channels,
                out_channels=resolved_config.out_channels,
                hidden_size=resolved_config.hidden_size,
                num_layers=2,
                output_size=resolved_config.output_size,
                image_size=cast(tuple[int, int], resolved_config.image_size),
                reference_canvas_size=cast(
                    tuple[int, int], resolved_config.reference_canvas_size
                ),
                backbone_feature_size=resolved_config.backbone_feature_size,
            )

        self.generator = generator or DSGANModel(
            resolved_config, backbone_weights=generator_backbone_weights
        )
        self.discriminator = discriminator or DSGANDiscriminator(
            resolved_discriminator_config,
            backbone_weights=discriminator_backbone_weights,
        )
        self.criterion = DSGANSetCriterion()
        self.generator_learning_rate = generator_learning_rate
        self.discriminator_learning_rate = discriminator_learning_rate
        self.generator_backbone_learning_rate = generator_backbone_learning_rate
        self.discriminator_backbone_learning_rate = discriminator_backbone_learning_rate
        self.adversarial_weight_ramp_steps = adversarial_weight_ramp_steps
        self.scheduler_gamma = scheduler_gamma
        self.scheduler_generator_milestones = scheduler_generator_milestones
        self.scheduler_discriminator_milestones = scheduler_discriminator_milestones
        self.seed = seed
        self.automatic_optimization = False

        self._numpy_rng = np.random.RandomState(seed)
        self._torch_generator = torch.Generator(device="cpu")
        self._torch_generator.manual_seed(seed)
        self.latest_step_trace: dict[str, Shaped[torch.Tensor, "..."]] = {}
        self._step_index = 0

    def forward(
        self,
        pixel_values: Float[torch.Tensor, "batch 4 height width"],
        initial_layout: Float[torch.Tensor, "batch elements 2 4"],
    ) -> DSGANModelOutput:
        """Run the package generator."""
        output = self.generator(pixel_values=pixel_values, layout=initial_layout)
        if not isinstance(output, DSGANModelOutput):
            raise TypeError("DS-GAN generator must return DSGANModelOutput")

        return output

    def on_fit_start(self) -> None:
        """Share the data module's seeded stream with layout initialization."""
        datamodule = getattr(self.trainer, "datamodule", None)
        if datamodule is None:
            return

        generator = getattr(datamodule, "training_generator", None)
        if isinstance(generator, torch.Generator):
            self._torch_generator = generator

    def configure_optimizers(self) -> OptimizerLRScheduler:
        """Return Adam optimizers and reference MultiStepLR schedulers."""
        generator_backbone, generator_head = self._parameter_groups(self.generator)
        discriminator_backbone, discriminator_head = self._parameter_groups(
            self.discriminator
        )
        optimizer_g = torch.optim.Adam(
            [
                {"params": generator_head, "lr": self.generator_learning_rate},
                {
                    "params": generator_backbone,
                    "lr": self.generator_backbone_learning_rate,
                },
            ]
        )
        optimizer_d = torch.optim.Adam(
            [
                {"params": discriminator_head, "lr": self.discriminator_learning_rate},
                {
                    "params": discriminator_backbone,
                    "lr": self.discriminator_backbone_learning_rate,
                },
            ]
        )
        scheduler_g = torch.optim.lr_scheduler.MultiStepLR(
            optimizer_g,
            milestones=self.scheduler_generator_milestones,
            gamma=self.scheduler_gamma,
        )
        scheduler_d = torch.optim.lr_scheduler.MultiStepLR(
            optimizer_d,
            milestones=self.scheduler_discriminator_milestones,
            gamma=self.scheduler_gamma,
        )
        scheduler_configs: list[LRSchedulerConfig] = [
            cast(
                LRSchedulerConfig,
                {"scheduler": scheduler_g, "interval": "epoch", "frequency": 1},
            ),
            cast(
                LRSchedulerConfig,
                {"scheduler": scheduler_d, "interval": "epoch", "frequency": 1},
            ),
        ]
        return [optimizer_g, optimizer_d], scheduler_configs

    def training_step(
        self, batch: Mapping[str, Shaped[torch.Tensor, "..."]], batch_idx: int
    ) -> Float[torch.Tensor, ""]:
        """Run one manually optimized generator-then-discriminator update."""
        del batch_idx
        optimizers = self.optimizers()
        if not isinstance(optimizers, list) or len(optimizers) != 2:
            raise RuntimeError("DS-GAN requires two Lightning optimizers")

        optimizer_g = cast(torch.optim.Optimizer, optimizers[0])
        optimizer_d = cast(torch.optim.Optimizer, optimizers[1])
        return self.step_with_optimizers(
            batch,
            optimizer_g,
            optimizer_d,
            numpy_rng=self._numpy_rng,
            torch_generator=self._torch_generator,
            epoch=self.current_epoch + 1,
        )

    def step_with_optimizers(
        self,
        batch: Mapping[str, Shaped[torch.Tensor, "..."]],
        optimizer_g: torch.optim.Optimizer,
        optimizer_d: torch.optim.Optimizer,
        *,
        initial_layout: Float[torch.Tensor, "batch elements 2 4"] | None = None,
        numpy_rng: np.random.RandomState | None = None,
        torch_generator: torch.Generator | None = None,
        epoch: int = 1,
    ) -> Float[torch.Tensor, ""]:
        """Apply one exact G-then-D update and record its comparison trace."""
        pixel_values = batch["pixel_values"]
        target_layout = batch["layout"]
        batch_size = pixel_values.shape[0]
        if initial_layout is None:
            resolved_numpy_rng = numpy_rng if numpy_rng is not None else np.random
            initial_layout = vendor_random_initial_layout(
                batch_size,
                self.ds_gan_config.max_elem,
                numpy_rng=cast(NumpyChoiceSource, resolved_numpy_rng),
                torch_generator=torch_generator,
                device=pixel_values.device,
            )

        initial_layout = initial_layout.to(pixel_values.device, pixel_values.dtype)
        all_real = torch.ones(batch_size, device=pixel_values.device)
        all_fake = torch.full((batch_size,), -1.0, device=pixel_values.device)
        target_labels = batch["labels"].long()
        target_boxes = batch["boxes"].float()
        targets = [
            {"labels": labels, "boxes": boxes}
            for labels, boxes in zip(target_labels, target_boxes, strict=True)
        ]
        adversarial_weight = min(
            1.0, max(0, epoch - 1) / self.adversarial_weight_ramp_steps
        )

        optimizer_g.zero_grad(set_to_none=True)
        generated = self.forward(pixel_values, initial_layout)
        if generated.bbox is None:
            raise RuntimeError("DS-GAN generator did not return bounding boxes")

        generated_layout = torch.stack((generated.class_probs, generated.bbox), dim=2)
        discriminator_generated = self.discriminator(pixel_values, generated_layout)
        loss_g_adv = nn.functional.hinge_embedding_loss(
            discriminator_generated.reshape(-1), all_real
        )
        reconstruction = self.criterion(generated.class_probs, generated.bbox, targets)
        loss_reconstruction = sum(reconstruction.values())
        loss_g = adversarial_weight * loss_g_adv + loss_reconstruction
        self._backward(loss_g)
        self._configure_gradient_clipping(optimizer_g)
        optimizer_g.step()

        optimizer_d.zero_grad(set_to_none=True)
        discriminator_fake = self.discriminator(pixel_values, generated_layout.detach())
        discriminator_real = self.discriminator(pixel_values, target_layout.clone())
        loss_d_fake = nn.functional.hinge_embedding_loss(
            discriminator_fake.reshape(-1), all_fake
        )
        loss_d_real = nn.functional.hinge_embedding_loss(
            discriminator_real.reshape(-1), all_real
        )
        loss_d = adversarial_weight * (loss_d_real + loss_d_fake)
        self._backward(loss_d)
        self._configure_gradient_clipping(optimizer_d)
        optimizer_d.step()

        self.latest_step_trace = {
            "initial_layout": initial_layout.detach(),
            "class_probs": generated.class_probs.detach(),
            "bbox": generated.bbox.detach(),
            "discriminator_generated": discriminator_generated.detach(),
            "discriminator_fake": discriminator_fake.detach(),
            "discriminator_real": discriminator_real.detach(),
            "loss_g_adv": loss_g_adv.detach(),
            "loss_reconstruction": loss_reconstruction.detach(),
            "loss_g": loss_g.detach(),
            "loss_d_fake": loss_d_fake.detach(),
            "loss_d_real": loss_d_real.detach(),
            "loss_d": loss_d.detach(),
            "adversarial_weight": torch.tensor(
                adversarial_weight, device=pixel_values.device
            ),
        }
        self._step_index += 1
        return loss_g.detach()

    def on_train_epoch_end(self) -> None:
        """Advance both epoch schedulers after all manual updates in an epoch."""
        schedulers = self.lr_schedulers()
        if isinstance(schedulers, list):
            for scheduler in cast(
                list[torch.optim.lr_scheduler.MultiStepLR], schedulers
            ):
                scheduler.step()

    @staticmethod
    def _parameter_groups(
        module: nn.Module,
    ) -> tuple[list[nn.Parameter], list[nn.Parameter]]:
        """Partition ResNet parameters from the trainable head."""
        backbone: list[nn.Parameter] = []
        head: list[nn.Parameter] = []
        for name, parameter in module.named_parameters():
            (backbone if name.startswith("resnet_fpn") else head).append(parameter)

        return backbone, head

    def _backward(self, loss: Float[torch.Tensor, ""]) -> None:
        """Backpropagate through Lightning or a direct parity harness."""
        if self._trainer is None:
            loss.backward()
            return

        self.manual_backward(loss)

    def _configure_gradient_clipping(self, optimizer: torch.optim.Optimizer) -> None:
        if self._trainer is None:
            return

        self.configure_gradient_clipping(
            optimizer,
            gradient_clip_val=self.trainer.gradient_clip_val,
            gradient_clip_algorithm=self.trainer.gradient_clip_algorithm,
        )

    def configure_gradient_clipping(
        self,
        optimizer: torch.optim.Optimizer,
        gradient_clip_val: float | None = None,
        gradient_clip_algorithm: str | None = None,
    ) -> None:
        """Delegate the Trainer's clipping policy for the active optimizer."""
        super().configure_gradient_clipping(
            optimizer,
            gradient_clip_val=gradient_clip_val,
            gradient_clip_algorithm=gradient_clip_algorithm,
        )

__init__

__init__(
    *,
    config: TrainingConfigInput,
    discriminator_config: TrainingConfigInput | None = None,
    generator: DSGANModel | None = None,
    discriminator: DSGANDiscriminator | None = None,
    generator_backbone_weights: str | None = None,
    discriminator_backbone_weights: str | None = None,
    generator_learning_rate: float = 0.0001,
    discriminator_learning_rate: float = 0.001,
    generator_backbone_learning_rate: float = 1e-05,
    discriminator_backbone_learning_rate: float = 0.0001,
    adversarial_weight_ramp_steps: int = 100,
    scheduler_gamma: float = 0.8,
    scheduler_generator_milestones: tuple[int, ...] = (
        0,
        50,
        100,
        150,
        200,
        250,
    ),
    scheduler_discriminator_milestones: tuple[int, ...] = (
        0,
        25,
        50,
        75,
        100,
        125,
        150,
        175,
        200,
        225,
        250,
        275,
    ),
    seed: int = 0,
) -> None

Initialize the two-network DS-GAN training state.

Source code in models/ds-gan/src/ds_gan/training/lightning_module.py
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
def __init__(
    self,
    *,
    config: TrainingConfigInput,
    discriminator_config: TrainingConfigInput | None = None,
    generator: DSGANModel | None = None,
    discriminator: DSGANDiscriminator | None = None,
    generator_backbone_weights: str | None = None,
    discriminator_backbone_weights: str | None = None,
    generator_learning_rate: float = 1e-4,
    discriminator_learning_rate: float = 1e-3,
    generator_backbone_learning_rate: float = 1e-5,
    discriminator_backbone_learning_rate: float = 1e-4,
    adversarial_weight_ramp_steps: int = 100,
    scheduler_gamma: float = 0.8,
    scheduler_generator_milestones: tuple[int, ...] = (0, 50, 100, 150, 200, 250),
    scheduler_discriminator_milestones: tuple[int, ...] = (
        0,
        25,
        50,
        75,
        100,
        125,
        150,
        175,
        200,
        225,
        250,
        275,
    ),
    seed: int = 0,
) -> None:
    """Initialize the two-network DS-GAN training state."""
    super().__init__()
    resolved_config = _resolve_config(config)
    resolved_discriminator_config = (
        _resolve_config(discriminator_config)
        if discriminator_config is not None
        else None
    )
    self.save_hyperparameters(
        ignore=(
            "generator",
            "discriminator",
            "generator_backbone_weights",
            "discriminator_backbone_weights",
        )
    )
    self.ds_gan_config = resolved_config

    if resolved_discriminator_config is None:
        resolved_discriminator_config = DSGANConfig(
            dataset_name=resolved_config.dataset_name,
            backbone="resnet18",
            max_elem=resolved_config.max_elem,
            in_channels=resolved_config.in_channels,
            out_channels=resolved_config.out_channels,
            hidden_size=resolved_config.hidden_size,
            num_layers=2,
            output_size=resolved_config.output_size,
            image_size=cast(tuple[int, int], resolved_config.image_size),
            reference_canvas_size=cast(
                tuple[int, int], resolved_config.reference_canvas_size
            ),
            backbone_feature_size=resolved_config.backbone_feature_size,
        )

    self.generator = generator or DSGANModel(
        resolved_config, backbone_weights=generator_backbone_weights
    )
    self.discriminator = discriminator or DSGANDiscriminator(
        resolved_discriminator_config,
        backbone_weights=discriminator_backbone_weights,
    )
    self.criterion = DSGANSetCriterion()
    self.generator_learning_rate = generator_learning_rate
    self.discriminator_learning_rate = discriminator_learning_rate
    self.generator_backbone_learning_rate = generator_backbone_learning_rate
    self.discriminator_backbone_learning_rate = discriminator_backbone_learning_rate
    self.adversarial_weight_ramp_steps = adversarial_weight_ramp_steps
    self.scheduler_gamma = scheduler_gamma
    self.scheduler_generator_milestones = scheduler_generator_milestones
    self.scheduler_discriminator_milestones = scheduler_discriminator_milestones
    self.seed = seed
    self.automatic_optimization = False

    self._numpy_rng = np.random.RandomState(seed)
    self._torch_generator = torch.Generator(device="cpu")
    self._torch_generator.manual_seed(seed)
    self.latest_step_trace: dict[str, Shaped[torch.Tensor, "..."]] = {}
    self._step_index = 0

forward

forward(
    pixel_values: Float[Tensor, "batch 4 height width"],
    initial_layout: Float[Tensor, "batch elements 2 4"],
) -> DSGANModelOutput

Run the package generator.

Source code in models/ds-gan/src/ds_gan/training/lightning_module.py
175
176
177
178
179
180
181
182
183
184
185
def forward(
    self,
    pixel_values: Float[torch.Tensor, "batch 4 height width"],
    initial_layout: Float[torch.Tensor, "batch elements 2 4"],
) -> DSGANModelOutput:
    """Run the package generator."""
    output = self.generator(pixel_values=pixel_values, layout=initial_layout)
    if not isinstance(output, DSGANModelOutput):
        raise TypeError("DS-GAN generator must return DSGANModelOutput")

    return output

on_fit_start

on_fit_start() -> None

Share the data module's seeded stream with layout initialization.

Source code in models/ds-gan/src/ds_gan/training/lightning_module.py
187
188
189
190
191
192
193
194
195
def on_fit_start(self) -> None:
    """Share the data module's seeded stream with layout initialization."""
    datamodule = getattr(self.trainer, "datamodule", None)
    if datamodule is None:
        return

    generator = getattr(datamodule, "training_generator", None)
    if isinstance(generator, torch.Generator):
        self._torch_generator = generator

configure_optimizers

configure_optimizers() -> OptimizerLRScheduler

Return Adam optimizers and reference MultiStepLR schedulers.

Source code in models/ds-gan/src/ds_gan/training/lightning_module.py
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
def configure_optimizers(self) -> OptimizerLRScheduler:
    """Return Adam optimizers and reference MultiStepLR schedulers."""
    generator_backbone, generator_head = self._parameter_groups(self.generator)
    discriminator_backbone, discriminator_head = self._parameter_groups(
        self.discriminator
    )
    optimizer_g = torch.optim.Adam(
        [
            {"params": generator_head, "lr": self.generator_learning_rate},
            {
                "params": generator_backbone,
                "lr": self.generator_backbone_learning_rate,
            },
        ]
    )
    optimizer_d = torch.optim.Adam(
        [
            {"params": discriminator_head, "lr": self.discriminator_learning_rate},
            {
                "params": discriminator_backbone,
                "lr": self.discriminator_backbone_learning_rate,
            },
        ]
    )
    scheduler_g = torch.optim.lr_scheduler.MultiStepLR(
        optimizer_g,
        milestones=self.scheduler_generator_milestones,
        gamma=self.scheduler_gamma,
    )
    scheduler_d = torch.optim.lr_scheduler.MultiStepLR(
        optimizer_d,
        milestones=self.scheduler_discriminator_milestones,
        gamma=self.scheduler_gamma,
    )
    scheduler_configs: list[LRSchedulerConfig] = [
        cast(
            LRSchedulerConfig,
            {"scheduler": scheduler_g, "interval": "epoch", "frequency": 1},
        ),
        cast(
            LRSchedulerConfig,
            {"scheduler": scheduler_d, "interval": "epoch", "frequency": 1},
        ),
    ]
    return [optimizer_g, optimizer_d], scheduler_configs

training_step

training_step(
    batch: Mapping[str, Shaped[Tensor, "..."]],
    batch_idx: int,
) -> Float[torch.Tensor, ""]

Run one manually optimized generator-then-discriminator update.

Source code in models/ds-gan/src/ds_gan/training/lightning_module.py
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
def training_step(
    self, batch: Mapping[str, Shaped[torch.Tensor, "..."]], batch_idx: int
) -> Float[torch.Tensor, ""]:
    """Run one manually optimized generator-then-discriminator update."""
    del batch_idx
    optimizers = self.optimizers()
    if not isinstance(optimizers, list) or len(optimizers) != 2:
        raise RuntimeError("DS-GAN requires two Lightning optimizers")

    optimizer_g = cast(torch.optim.Optimizer, optimizers[0])
    optimizer_d = cast(torch.optim.Optimizer, optimizers[1])
    return self.step_with_optimizers(
        batch,
        optimizer_g,
        optimizer_d,
        numpy_rng=self._numpy_rng,
        torch_generator=self._torch_generator,
        epoch=self.current_epoch + 1,
    )

step_with_optimizers

step_with_optimizers(
    batch: Mapping[str, Shaped[Tensor, "..."]],
    optimizer_g: Optimizer,
    optimizer_d: Optimizer,
    *,
    initial_layout: Float[Tensor, "batch elements 2 4"]
    | None = None,
    numpy_rng: RandomState | None = None,
    torch_generator: Generator | None = None,
    epoch: int = 1,
) -> Float[torch.Tensor, ""]

Apply one exact G-then-D update and record its comparison trace.

Source code in models/ds-gan/src/ds_gan/training/lightning_module.py
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
def step_with_optimizers(
    self,
    batch: Mapping[str, Shaped[torch.Tensor, "..."]],
    optimizer_g: torch.optim.Optimizer,
    optimizer_d: torch.optim.Optimizer,
    *,
    initial_layout: Float[torch.Tensor, "batch elements 2 4"] | None = None,
    numpy_rng: np.random.RandomState | None = None,
    torch_generator: torch.Generator | None = None,
    epoch: int = 1,
) -> Float[torch.Tensor, ""]:
    """Apply one exact G-then-D update and record its comparison trace."""
    pixel_values = batch["pixel_values"]
    target_layout = batch["layout"]
    batch_size = pixel_values.shape[0]
    if initial_layout is None:
        resolved_numpy_rng = numpy_rng if numpy_rng is not None else np.random
        initial_layout = vendor_random_initial_layout(
            batch_size,
            self.ds_gan_config.max_elem,
            numpy_rng=cast(NumpyChoiceSource, resolved_numpy_rng),
            torch_generator=torch_generator,
            device=pixel_values.device,
        )

    initial_layout = initial_layout.to(pixel_values.device, pixel_values.dtype)
    all_real = torch.ones(batch_size, device=pixel_values.device)
    all_fake = torch.full((batch_size,), -1.0, device=pixel_values.device)
    target_labels = batch["labels"].long()
    target_boxes = batch["boxes"].float()
    targets = [
        {"labels": labels, "boxes": boxes}
        for labels, boxes in zip(target_labels, target_boxes, strict=True)
    ]
    adversarial_weight = min(
        1.0, max(0, epoch - 1) / self.adversarial_weight_ramp_steps
    )

    optimizer_g.zero_grad(set_to_none=True)
    generated = self.forward(pixel_values, initial_layout)
    if generated.bbox is None:
        raise RuntimeError("DS-GAN generator did not return bounding boxes")

    generated_layout = torch.stack((generated.class_probs, generated.bbox), dim=2)
    discriminator_generated = self.discriminator(pixel_values, generated_layout)
    loss_g_adv = nn.functional.hinge_embedding_loss(
        discriminator_generated.reshape(-1), all_real
    )
    reconstruction = self.criterion(generated.class_probs, generated.bbox, targets)
    loss_reconstruction = sum(reconstruction.values())
    loss_g = adversarial_weight * loss_g_adv + loss_reconstruction
    self._backward(loss_g)
    self._configure_gradient_clipping(optimizer_g)
    optimizer_g.step()

    optimizer_d.zero_grad(set_to_none=True)
    discriminator_fake = self.discriminator(pixel_values, generated_layout.detach())
    discriminator_real = self.discriminator(pixel_values, target_layout.clone())
    loss_d_fake = nn.functional.hinge_embedding_loss(
        discriminator_fake.reshape(-1), all_fake
    )
    loss_d_real = nn.functional.hinge_embedding_loss(
        discriminator_real.reshape(-1), all_real
    )
    loss_d = adversarial_weight * (loss_d_real + loss_d_fake)
    self._backward(loss_d)
    self._configure_gradient_clipping(optimizer_d)
    optimizer_d.step()

    self.latest_step_trace = {
        "initial_layout": initial_layout.detach(),
        "class_probs": generated.class_probs.detach(),
        "bbox": generated.bbox.detach(),
        "discriminator_generated": discriminator_generated.detach(),
        "discriminator_fake": discriminator_fake.detach(),
        "discriminator_real": discriminator_real.detach(),
        "loss_g_adv": loss_g_adv.detach(),
        "loss_reconstruction": loss_reconstruction.detach(),
        "loss_g": loss_g.detach(),
        "loss_d_fake": loss_d_fake.detach(),
        "loss_d_real": loss_d_real.detach(),
        "loss_d": loss_d.detach(),
        "adversarial_weight": torch.tensor(
            adversarial_weight, device=pixel_values.device
        ),
    }
    self._step_index += 1
    return loss_g.detach()

on_train_epoch_end

on_train_epoch_end() -> None

Advance both epoch schedulers after all manual updates in an epoch.

Source code in models/ds-gan/src/ds_gan/training/lightning_module.py
352
353
354
355
356
357
358
359
def on_train_epoch_end(self) -> None:
    """Advance both epoch schedulers after all manual updates in an epoch."""
    schedulers = self.lr_schedulers()
    if isinstance(schedulers, list):
        for scheduler in cast(
            list[torch.optim.lr_scheduler.MultiStepLR], schedulers
        ):
            scheduler.step()

configure_gradient_clipping

configure_gradient_clipping(
    optimizer: Optimizer,
    gradient_clip_val: float | None = None,
    gradient_clip_algorithm: str | None = None,
) -> None

Delegate the Trainer's clipping policy for the active optimizer.

Source code in models/ds-gan/src/ds_gan/training/lightning_module.py
391
392
393
394
395
396
397
398
399
400
401
402
def configure_gradient_clipping(
    self,
    optimizer: torch.optim.Optimizer,
    gradient_clip_val: float | None = None,
    gradient_clip_algorithm: str | None = None,
) -> None:
    """Delegate the Trainer's clipping policy for the active optimizer."""
    super().configure_gradient_clipping(
        optimizer,
        gradient_clip_val=gradient_clip_val,
        gradient_clip_algorithm=gradient_clip_algorithm,
    )

DSGANSetCriterion

Bases: Module

Compute the reference class, box, and generalized-IoU losses.

Source code in models/ds-gan/src/ds_gan/training/losses.py
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
class DSGANSetCriterion(nn.Module):
    """Compute the reference class, box, and generalized-IoU losses."""

    def __init__(self, eos_weight: float = 0.1) -> None:
        """Initialize the fixed reference loss weights."""
        super().__init__()
        self.matcher = HungarianMatcher()
        self.register_buffer("empty_weight", torch.tensor((eos_weight, 0.8, 1.0, 1.0)))

    def forward(
        self,
        pred_logits: Float[torch.Tensor, "batch elements 4"],
        pred_boxes: Float[torch.Tensor, "batch elements 4"],
        targets: Sequence[dict[str, Shaped[torch.Tensor, "elements ..."]]],
    ) -> dict[str, Float[torch.Tensor, ""]]:
        """Return the three weighted reconstruction components."""
        outputs = {"pred_logits": pred_logits, "pred_boxes": pred_boxes}
        indices = self.matcher(outputs, targets)
        target_count = (
            torch.as_tensor(
                [sum(len(target["labels"]) for target in targets)],
                dtype=torch.float,
                device=pred_logits.device,
            )
            .clamp_min(1)
            .item()
        )
        batch_indices = torch.cat(
            [
                torch.full_like(source, index)
                for index, (source, _) in enumerate(indices)
            ]
        )
        source_indices = torch.cat([source for source, _ in indices])
        target_labels = torch.cat(
            [
                target["labels"][target_indices]
                for target, (_, target_indices) in zip(targets, indices, strict=True)
            ]
        )
        target_classes = torch.full(
            pred_logits.shape[:2],
            3,
            dtype=torch.long,
            device=pred_logits.device,
        )
        target_classes[batch_indices, source_indices] = target_labels
        loss_ce = torch.nn.functional.cross_entropy(
            pred_logits.transpose(1, 2),
            target_classes,
            cast(torch.Tensor, self.empty_weight),
        )
        source_boxes = pred_boxes[batch_indices, source_indices]
        target_boxes = torch.cat(
            [
                target["boxes"][target_indices]
                for target, (_, target_indices) in zip(targets, indices, strict=True)
            ]
        )
        loss_bbox = (
            torch.nn.functional.l1_loss(
                source_boxes, target_boxes, reduction="none"
            ).sum()
            / target_count
        )
        loss_giou = (
            1
            - torch.diag(
                _generalized_box_iou(
                    _cxcywh_to_xyxy(source_boxes), _cxcywh_to_xyxy(target_boxes)
                )
            )
        ).sum() / target_count
        return {"loss_ce": loss_ce, "loss_bbox": loss_bbox, "loss_giou": loss_giou}

__init__

__init__(eos_weight: float = 0.1) -> None

Initialize the fixed reference loss weights.

Source code in models/ds-gan/src/ds_gan/training/losses.py
113
114
115
116
117
def __init__(self, eos_weight: float = 0.1) -> None:
    """Initialize the fixed reference loss weights."""
    super().__init__()
    self.matcher = HungarianMatcher()
    self.register_buffer("empty_weight", torch.tensor((eos_weight, 0.8, 1.0, 1.0)))

forward

forward(
    pred_logits: Float[Tensor, "batch elements 4"],
    pred_boxes: Float[Tensor, "batch elements 4"],
    targets: Sequence[
        dict[str, Shaped[Tensor, "elements ..."]]
    ],
) -> dict[str, Float[torch.Tensor, ""]]

Return the three weighted reconstruction components.

Source code in models/ds-gan/src/ds_gan/training/losses.py
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
def forward(
    self,
    pred_logits: Float[torch.Tensor, "batch elements 4"],
    pred_boxes: Float[torch.Tensor, "batch elements 4"],
    targets: Sequence[dict[str, Shaped[torch.Tensor, "elements ..."]]],
) -> dict[str, Float[torch.Tensor, ""]]:
    """Return the three weighted reconstruction components."""
    outputs = {"pred_logits": pred_logits, "pred_boxes": pred_boxes}
    indices = self.matcher(outputs, targets)
    target_count = (
        torch.as_tensor(
            [sum(len(target["labels"]) for target in targets)],
            dtype=torch.float,
            device=pred_logits.device,
        )
        .clamp_min(1)
        .item()
    )
    batch_indices = torch.cat(
        [
            torch.full_like(source, index)
            for index, (source, _) in enumerate(indices)
        ]
    )
    source_indices = torch.cat([source for source, _ in indices])
    target_labels = torch.cat(
        [
            target["labels"][target_indices]
            for target, (_, target_indices) in zip(targets, indices, strict=True)
        ]
    )
    target_classes = torch.full(
        pred_logits.shape[:2],
        3,
        dtype=torch.long,
        device=pred_logits.device,
    )
    target_classes[batch_indices, source_indices] = target_labels
    loss_ce = torch.nn.functional.cross_entropy(
        pred_logits.transpose(1, 2),
        target_classes,
        cast(torch.Tensor, self.empty_weight),
    )
    source_boxes = pred_boxes[batch_indices, source_indices]
    target_boxes = torch.cat(
        [
            target["boxes"][target_indices]
            for target, (_, target_indices) in zip(targets, indices, strict=True)
        ]
    )
    loss_bbox = (
        torch.nn.functional.l1_loss(
            source_boxes, target_boxes, reduction="none"
        ).sum()
        / target_count
    )
    loss_giou = (
        1
        - torch.diag(
            _generalized_box_iou(
                _cxcywh_to_xyxy(source_boxes), _cxcywh_to_xyxy(target_boxes)
            )
        )
    ).sum() / target_count
    return {"loss_ce": loss_ce, "loss_bbox": loss_bbox, "loss_giou": loss_giou}

HungarianMatcher

Bases: Module

Match predicted layout slots to target elements.

Source code in models/ds-gan/src/ds_gan/training/losses.py
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
class HungarianMatcher(nn.Module):
    """Match predicted layout slots to target elements."""

    def __init__(
        self, cost_class: float = 2.0, cost_bbox: float = 5.0, cost_giou: float = 2.0
    ) -> None:
        """Initialize weighted matching costs."""
        super().__init__()
        if cost_class == 0 and cost_bbox == 0 and cost_giou == 0:
            raise ValueError("at least one matching cost must be non-zero")

        self.cost_class = cost_class
        self.cost_bbox = cost_bbox
        self.cost_giou = cost_giou

    @torch.no_grad()
    def forward(
        self,
        outputs: dict[str, Float[torch.Tensor, "batch elements channels"]],
        targets: Sequence[dict[str, Shaped[torch.Tensor, "elements ..."]]],
    ) -> list[tuple[Int[torch.Tensor, "matched"], Int[torch.Tensor, "matched"]]]:
        """Return per-example Hungarian assignments."""
        logits = outputs["pred_logits"]
        boxes = outputs["pred_boxes"]
        batch_size, num_queries = logits.shape[:2]
        probabilities = logits.flatten(0, 1).softmax(-1)
        target_labels = torch.cat([target["labels"] for target in targets])
        target_boxes = torch.cat([target["boxes"] for target in targets])
        class_cost = -probabilities[:, target_labels]
        bbox_cost = torch.cdist(boxes.flatten(0, 1), target_boxes, p=1)
        giou_cost = -_generalized_box_iou(
            _cxcywh_to_xyxy(boxes.flatten(0, 1)), _cxcywh_to_xyxy(target_boxes)
        )
        cost = (
            self.cost_bbox * bbox_cost
            + self.cost_class * class_cost
            + self.cost_giou * giou_cost
        )
        cost = cost.reshape(batch_size, num_queries, -1).cpu()
        sizes = [len(target["boxes"]) for target in targets]
        assignments: list[
            tuple[Int[torch.Tensor, "matched"], Int[torch.Tensor, "matched"]]
        ] = []
        offset = 0
        for index, size in enumerate(sizes):
            matrix = cost[index, :, offset : offset + size]
            row, column = linear_sum_assignment(matrix)
            assignments.append(
                (
                    torch.as_tensor(row, dtype=torch.long),
                    torch.as_tensor(column, dtype=torch.long),
                )
            )
            offset += size

        return assignments

__init__

__init__(
    cost_class: float = 2.0,
    cost_bbox: float = 5.0,
    cost_giou: float = 2.0,
) -> None

Initialize weighted matching costs.

Source code in models/ds-gan/src/ds_gan/training/losses.py
55
56
57
58
59
60
61
62
63
64
65
def __init__(
    self, cost_class: float = 2.0, cost_bbox: float = 5.0, cost_giou: float = 2.0
) -> None:
    """Initialize weighted matching costs."""
    super().__init__()
    if cost_class == 0 and cost_bbox == 0 and cost_giou == 0:
        raise ValueError("at least one matching cost must be non-zero")

    self.cost_class = cost_class
    self.cost_bbox = cost_bbox
    self.cost_giou = cost_giou

forward

forward(
    outputs: dict[
        str, Float[Tensor, "batch elements channels"]
    ],
    targets: Sequence[
        dict[str, Shaped[Tensor, "elements ..."]]
    ],
) -> list[
    tuple[
        Int[torch.Tensor, "matched"],
        Int[torch.Tensor, "matched"],
    ]
]

Return per-example Hungarian assignments.

Source code in models/ds-gan/src/ds_gan/training/losses.py
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
@torch.no_grad()
def forward(
    self,
    outputs: dict[str, Float[torch.Tensor, "batch elements channels"]],
    targets: Sequence[dict[str, Shaped[torch.Tensor, "elements ..."]]],
) -> list[tuple[Int[torch.Tensor, "matched"], Int[torch.Tensor, "matched"]]]:
    """Return per-example Hungarian assignments."""
    logits = outputs["pred_logits"]
    boxes = outputs["pred_boxes"]
    batch_size, num_queries = logits.shape[:2]
    probabilities = logits.flatten(0, 1).softmax(-1)
    target_labels = torch.cat([target["labels"] for target in targets])
    target_boxes = torch.cat([target["boxes"] for target in targets])
    class_cost = -probabilities[:, target_labels]
    bbox_cost = torch.cdist(boxes.flatten(0, 1), target_boxes, p=1)
    giou_cost = -_generalized_box_iou(
        _cxcywh_to_xyxy(boxes.flatten(0, 1)), _cxcywh_to_xyxy(target_boxes)
    )
    cost = (
        self.cost_bbox * bbox_cost
        + self.cost_class * class_cost
        + self.cost_giou * giou_cost
    )
    cost = cost.reshape(batch_size, num_queries, -1).cpu()
    sizes = [len(target["boxes"]) for target in targets]
    assignments: list[
        tuple[Int[torch.Tensor, "matched"], Int[torch.Tensor, "matched"]]
    ] = []
    offset = 0
    for index, size in enumerate(sizes):
        matrix = cost[index, :, offset : offset + size]
        row, column = linear_sum_assignment(matrix)
        assignments.append(
            (
                torch.as_tensor(row, dtype=torch.long),
                torch.as_tensor(column, dtype=torch.long),
            )
        )
        offset += size

    return assignments

build_bridge_manifest

build_bridge_manifest(
    *,
    source_revision: str,
    train_count: int,
    test_count: int,
    source_hashes: Mapping[str, str],
    source_fields: Mapping[str, str] | None = None,
    available_fields: Sequence[str] | None = None,
) -> dict[str, ManifestValue]

Build a hash-manifested record of the approved-source bridge.

Source code in models/ds-gan/src/ds_gan/training/dataset.py
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
def build_bridge_manifest(
    *,
    source_revision: str,
    train_count: int,
    test_count: int,
    source_hashes: Mapping[str, str],
    source_fields: Mapping[str, str] | None = None,
    available_fields: Sequence[str] | None = None,
) -> dict[str, ManifestValue]:
    """Build a hash-manifested record of the approved-source bridge."""
    mappings = dict(source_fields or SOURCE_FIELDS)
    expected = set(SOURCE_FIELDS)
    available = set(available_fields or mappings.values())
    missing = sorted(
        field for field in expected if mappings.get(field) not in available
    )
    return {
        "source": {
            "dataset": SOURCE_DATASET,
            "config": SOURCE_CONFIG,
            "revision": source_revision,
            "split_counts": {"train": train_count, "test": test_count},
        },
        "vendor_layout": {
            "train": {
                "inpainted_poster": mappings.get("train.inpainted_poster"),
                "saliency_pfpnet": mappings.get("train.saliency_pfpnet"),
                "saliency_basnet": mappings.get("train.saliency_basnet"),
                "annotations": mappings.get("train.annotations"),
            },
            "test": {
                "canvas": mappings.get("test.canvas"),
                "saliency_pfpnet": mappings.get("test.saliency_pfpnet"),
                "saliency_basnet": mappings.get("test.saliency_basnet"),
            },
        },
        "missing_source_fields": missing,
        "hashes": dict(source_hashes),
        "transform": {
            "resize": [350, 240],
            "saliency_merge": "pixelwise_max",
            "box_format": "ltrb_to_normalized_cxcywh",
            "canvas_size": [513, 750],
            "invalid_class": "filtered",
            "source_label_to_vendor_class": {
                "0": 1,
                "1": 2,
                "2": 3,
                "3": 0,
            },
        },
    }

bridge_example

bridge_example(
    example: Mapping[str, DatasetValue], *, split: str
) -> dict[str, DatasetValue]

Map one approved-source row to the fields consumed by the reference loader.

Source code in models/ds-gan/src/ds_gan/training/dataset.py
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
def bridge_example(
    example: Mapping[str, DatasetValue],
    *,
    split: str,
) -> dict[str, DatasetValue]:
    """Map one approved-source row to the fields consumed by the reference loader."""
    required = (
        ("canvas", "pfpn_saliency_map", "basnet_saliency_map")
        if split == "test"
        else (
            "inpainted_poster",
            "pfpn_saliency_map",
            "basnet_saliency_map",
            "annotations",
        )
    )
    missing = [
        field for field in required if field not in example or example[field] is None
    ]
    if missing:
        raise KeyError(f"approved source lacks required input fields: {missing}")

    return {
        "inpainted_poster": example.get("inpainted_poster"),
        "canvas": example.get("canvas"),
        "saliency_pfpnet": example.get("pfpn_saliency_map"),
        "saliency_basnet": example.get("basnet_saliency_map"),
        "annotations": example.get("annotations"),
    }

explicit_setup_seed

explicit_setup_seed(
    seed: int, *, deterministic: bool = True
) -> None

Set every random source used by the reference training recipe.

Source code in models/ds-gan/src/ds_gan/training/rng.py
29
30
31
32
33
34
35
36
37
38
def explicit_setup_seed(seed: int, *, deterministic: bool = True) -> None:
    """Set every random source used by the reference training recipe."""
    random.seed(seed)
    np.random.seed(seed)
    torch.manual_seed(seed)
    if torch.cuda.is_available():
        torch.cuda.manual_seed_all(seed)

    if deterministic:
        torch.use_deterministic_algorithms(True, warn_only=True)

vendor_random_initial_layout

vendor_random_initial_layout(
    batch_size: int,
    max_elem: int,
    *,
    numpy_rng: NumpyChoiceSource,
    torch_generator: Generator | None,
    device: device | str = "cpu",
) -> Float[torch.Tensor, "batch elements 2 4"]

Reproduce the reference class and box draws with explicit RNG objects.

Source code in models/ds-gan/src/ds_gan/training/rng.py
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
def vendor_random_initial_layout(
    batch_size: int,
    max_elem: int,
    *,
    numpy_rng: NumpyChoiceSource,
    torch_generator: torch.Generator | None,
    device: torch.device | str = "cpu",
) -> Float[torch.Tensor, "batch elements 2 4"]:
    """Reproduce the reference class and box draws with explicit RNG objects."""
    class_ids = numpy_rng.choice(
        4,
        size=(batch_size, max_elem, 1),
        p=np.asarray((0.1, 0.8, 1.0, 1.0), dtype=np.float64) / 2.9,
    )
    class_tensor = torch.zeros(
        batch_size, max_elem, 4, device=device, dtype=torch.float32
    )
    class_tensor.scatter_(
        -1,
        torch.as_tensor(class_ids, device=device, dtype=torch.long),
        1.0,
    )
    box_xyxy = normal(
        0.5,
        0.15,
        size=(batch_size, max_elem, 1, 4),
        generator=torch_generator,
        device=device,
        dtype=torch.float32,
    )
    return torch.cat((class_tensor.unsqueeze(2), xyxy_to_xywh(box_xyxy)), dim=2)

datamodule

Lightning data module for the approved PKU PosterLayout source.

DSGANDataModule

Bases: LightningDataModule

Prepare training and test streams with the reference transforms.

Source code in models/ds-gan/src/ds_gan/training/datamodule.py
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
class DSGANDataModule(LightningDataModule):
    """Prepare training and test streams with the reference transforms."""

    def __init__(
        self,
        *,
        dataset_name: str = "pku_posterlayout",
        cache_dir: str | None = None,
        batch_size: int = 128,
        test_batch_size: int = 4,
        max_elem: int = 32,
        num_workers: int = 0,
        seed: int = 0,
    ) -> None:
        """Initialize the approved-source loader settings."""
        super().__init__()
        if dataset_name not in {"pku_posterlayout", "PKU-PosterLayout"}:
            raise ValueError(f"unsupported DS-GAN dataset: {dataset_name}")

        self.cache_dir = cache_dir
        self.batch_size = batch_size
        self.test_batch_size = test_batch_size
        self.max_elem = max_elem
        self.num_workers = num_workers
        self.seed = seed
        self.processor = DSGANProcessor()
        self.dataset: DatasetDict | None = None
        self._train_generator: torch.Generator | None = None

    def setup(self, stage: str | None = None) -> None:
        """Open cached source rows for fitting or evaluation."""
        if stage in {None, "fit", "test"}:
            self.dataset = load_cached_dataset(self.cache_dir)

    def train_dataloader(self) -> DataLoader[dict[str, Shaped[torch.Tensor, "..."]]]:
        """Return the reference-compatible shuffled training stream."""
        if self.dataset is None:
            self.setup("fit")

        assert self.dataset is not None
        dataset = DSGANDataset(
            self.dataset["train"],
            split="train",
            processor=self.processor,
            max_elem=self.max_elem,
        )
        generator = self.training_generator
        return DataLoader(
            dataset,
            batch_size=self.batch_size,
            sampler=_VendorRandomSampler(dataset, generator),
            num_workers=self.num_workers,
            generator=generator,
        )

    def test_dataloader(self) -> DataLoader[dict[str, Shaped[torch.Tensor, "..."]]]:
        """Return the complete unshuffled test stream."""
        if self.dataset is None:
            self.setup("test")

        assert self.dataset is not None
        return DataLoader(
            DSGANDataset(
                self.dataset["test"],
                split="test",
                processor=self.processor,
                max_elem=self.max_elem,
            ),
            batch_size=self.test_batch_size,
            shuffle=False,
            num_workers=self.num_workers,
        )

    @property
    def training_generator(self) -> torch.Generator:
        """Return the shared sampler and layout-initialization RNG stream."""
        if self._train_generator is None:
            self._train_generator = torch.Generator(device="cpu")
            self._train_generator.manual_seed(self.seed)

        return self._train_generator
training_generator property
training_generator: Generator

Return the shared sampler and layout-initialization RNG stream.

__init__
__init__(
    *,
    dataset_name: str = "pku_posterlayout",
    cache_dir: str | None = None,
    batch_size: int = 128,
    test_batch_size: int = 4,
    max_elem: int = 32,
    num_workers: int = 0,
    seed: int = 0,
) -> None

Initialize the approved-source loader settings.

Source code in models/ds-gan/src/ds_gan/training/datamodule.py
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
def __init__(
    self,
    *,
    dataset_name: str = "pku_posterlayout",
    cache_dir: str | None = None,
    batch_size: int = 128,
    test_batch_size: int = 4,
    max_elem: int = 32,
    num_workers: int = 0,
    seed: int = 0,
) -> None:
    """Initialize the approved-source loader settings."""
    super().__init__()
    if dataset_name not in {"pku_posterlayout", "PKU-PosterLayout"}:
        raise ValueError(f"unsupported DS-GAN dataset: {dataset_name}")

    self.cache_dir = cache_dir
    self.batch_size = batch_size
    self.test_batch_size = test_batch_size
    self.max_elem = max_elem
    self.num_workers = num_workers
    self.seed = seed
    self.processor = DSGANProcessor()
    self.dataset: DatasetDict | None = None
    self._train_generator: torch.Generator | None = None
setup
setup(stage: str | None = None) -> None

Open cached source rows for fitting or evaluation.

Source code in models/ds-gan/src/ds_gan/training/datamodule.py
66
67
68
69
def setup(self, stage: str | None = None) -> None:
    """Open cached source rows for fitting or evaluation."""
    if stage in {None, "fit", "test"}:
        self.dataset = load_cached_dataset(self.cache_dir)
train_dataloader
train_dataloader() -> DataLoader[
    dict[str, Shaped[torch.Tensor, "..."]]
]

Return the reference-compatible shuffled training stream.

Source code in models/ds-gan/src/ds_gan/training/datamodule.py
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
def train_dataloader(self) -> DataLoader[dict[str, Shaped[torch.Tensor, "..."]]]:
    """Return the reference-compatible shuffled training stream."""
    if self.dataset is None:
        self.setup("fit")

    assert self.dataset is not None
    dataset = DSGANDataset(
        self.dataset["train"],
        split="train",
        processor=self.processor,
        max_elem=self.max_elem,
    )
    generator = self.training_generator
    return DataLoader(
        dataset,
        batch_size=self.batch_size,
        sampler=_VendorRandomSampler(dataset, generator),
        num_workers=self.num_workers,
        generator=generator,
    )
test_dataloader
test_dataloader() -> DataLoader[
    dict[str, Shaped[torch.Tensor, "..."]]
]

Return the complete unshuffled test stream.

Source code in models/ds-gan/src/ds_gan/training/datamodule.py
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
def test_dataloader(self) -> DataLoader[dict[str, Shaped[torch.Tensor, "..."]]]:
    """Return the complete unshuffled test stream."""
    if self.dataset is None:
        self.setup("test")

    assert self.dataset is not None
    return DataLoader(
        DSGANDataset(
            self.dataset["test"],
            split="test",
            processor=self.processor,
            max_elem=self.max_elem,
        ),
        batch_size=self.test_batch_size,
        shuffle=False,
        num_workers=self.num_workers,
    )

dataset

Approved-source to reference-layout data bridge for DS-GAN.

DSGANDataset

Bases: Dataset[dict[str, Shaped[Tensor, '...']]]

Lazy DS-GAN dataset using the approved source and reference preprocessing.

Source code in models/ds-gan/src/ds_gan/training/dataset.py
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
class DSGANDataset(TorchDataset[dict[str, Shaped[torch.Tensor, "..."]]]):
    """Lazy DS-GAN dataset using the approved source and reference preprocessing."""

    def __init__(
        self,
        rows: Dataset,
        *,
        split: str,
        processor: DSGANProcessor | None = None,
        max_elem: int = 32,
    ) -> None:
        """Initialize the source bridge."""
        self.rows = rows
        self.split = split
        self.processor = processor or DSGANProcessor()
        self.max_elem = max_elem

    def __len__(self) -> int:
        """Return the number of source rows."""
        return len(self.rows)

    def __getitem__(self, index: int) -> dict[str, Shaped[torch.Tensor, "..."]]:
        """Convert one source row into model inputs and padded labels."""
        source = bridge_example(
            cast(Mapping[str, DatasetValue], self.rows[index]), split=self.split
        )
        image = cast(
            DSGANImageInput,
            source["canvas"] if self.split == "test" else source["inpainted_poster"],
        )
        encoded = self.processor(
            image,
            saliency_pfpnet=cast(DSGANImageInput, source["saliency_pfpnet"]),
            saliency_basnet=cast(DSGANImageInput, source["saliency_basnet"]),
        )
        pixel_values = encoded["pixel_values"][0]
        if self.split == "test":
            return {"pixel_values": pixel_values}

        layout_fields = annotations_from_pku_example(
            cast(
                Mapping[str, DSGANExampleValue],
                {
                    "annotations": cast(DSGANExampleValue, source["annotations"]),
                    "canvas": image,
                    "inpainted_poster": image,
                },
            ),
            max_elem=self.max_elem,
            include_unconnected_underlays=False,
            canvas_size=VENDOR_CANVAS_SIZE,
            vendor_box_fix=True,
        )
        bbox = cast(torch.Tensor, layout_fields["bbox"]).squeeze(0)
        labels = cast(torch.Tensor, layout_fields["labels"]).squeeze(0)
        mask = cast(torch.Tensor, layout_fields["mask"]).squeeze(0)
        padded_bbox, padded_labels, padded_mask = self.processor.pad(
            bbox.unsqueeze(0),
            labels.unsqueeze(0),
            mask.unsqueeze(0),
            max_elem=self.max_elem,
        )
        model_labels = torch.zeros_like(padded_labels)
        model_labels[padded_mask] = padded_labels[padded_mask] + 1
        classes = torch.nn.functional.one_hot(model_labels, num_classes=4).to(
            dtype=padded_bbox.dtype
        )
        layout = torch.stack((classes, padded_bbox), dim=2).squeeze(0)
        return {
            "pixel_values": pixel_values,
            "layout": layout,
            "labels": model_labels.squeeze(0),
            "boxes": padded_bbox.squeeze(0),
            "mask": padded_mask.squeeze(0),
        }
__init__
__init__(
    rows: Dataset,
    *,
    split: str,
    processor: DSGANProcessor | None = None,
    max_elem: int = 32,
) -> None

Initialize the source bridge.

Source code in models/ds-gan/src/ds_gan/training/dataset.py
81
82
83
84
85
86
87
88
89
90
91
92
93
def __init__(
    self,
    rows: Dataset,
    *,
    split: str,
    processor: DSGANProcessor | None = None,
    max_elem: int = 32,
) -> None:
    """Initialize the source bridge."""
    self.rows = rows
    self.split = split
    self.processor = processor or DSGANProcessor()
    self.max_elem = max_elem
__len__
__len__() -> int

Return the number of source rows.

Source code in models/ds-gan/src/ds_gan/training/dataset.py
95
96
97
def __len__(self) -> int:
    """Return the number of source rows."""
    return len(self.rows)
__getitem__
__getitem__(
    index: int,
) -> dict[str, Shaped[torch.Tensor, "..."]]

Convert one source row into model inputs and padded labels.

Source code in models/ds-gan/src/ds_gan/training/dataset.py
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
def __getitem__(self, index: int) -> dict[str, Shaped[torch.Tensor, "..."]]:
    """Convert one source row into model inputs and padded labels."""
    source = bridge_example(
        cast(Mapping[str, DatasetValue], self.rows[index]), split=self.split
    )
    image = cast(
        DSGANImageInput,
        source["canvas"] if self.split == "test" else source["inpainted_poster"],
    )
    encoded = self.processor(
        image,
        saliency_pfpnet=cast(DSGANImageInput, source["saliency_pfpnet"]),
        saliency_basnet=cast(DSGANImageInput, source["saliency_basnet"]),
    )
    pixel_values = encoded["pixel_values"][0]
    if self.split == "test":
        return {"pixel_values": pixel_values}

    layout_fields = annotations_from_pku_example(
        cast(
            Mapping[str, DSGANExampleValue],
            {
                "annotations": cast(DSGANExampleValue, source["annotations"]),
                "canvas": image,
                "inpainted_poster": image,
            },
        ),
        max_elem=self.max_elem,
        include_unconnected_underlays=False,
        canvas_size=VENDOR_CANVAS_SIZE,
        vendor_box_fix=True,
    )
    bbox = cast(torch.Tensor, layout_fields["bbox"]).squeeze(0)
    labels = cast(torch.Tensor, layout_fields["labels"]).squeeze(0)
    mask = cast(torch.Tensor, layout_fields["mask"]).squeeze(0)
    padded_bbox, padded_labels, padded_mask = self.processor.pad(
        bbox.unsqueeze(0),
        labels.unsqueeze(0),
        mask.unsqueeze(0),
        max_elem=self.max_elem,
    )
    model_labels = torch.zeros_like(padded_labels)
    model_labels[padded_mask] = padded_labels[padded_mask] + 1
    classes = torch.nn.functional.one_hot(model_labels, num_classes=4).to(
        dtype=padded_bbox.dtype
    )
    layout = torch.stack((classes, padded_bbox), dim=2).squeeze(0)
    return {
        "pixel_values": pixel_values,
        "layout": layout,
        "labels": model_labels.squeeze(0),
        "boxes": padded_bbox.squeeze(0),
        "mask": padded_mask.squeeze(0),
    }

bridge_example

bridge_example(
    example: Mapping[str, DatasetValue], *, split: str
) -> dict[str, DatasetValue]

Map one approved-source row to the fields consumed by the reference loader.

Source code in models/ds-gan/src/ds_gan/training/dataset.py
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
def bridge_example(
    example: Mapping[str, DatasetValue],
    *,
    split: str,
) -> dict[str, DatasetValue]:
    """Map one approved-source row to the fields consumed by the reference loader."""
    required = (
        ("canvas", "pfpn_saliency_map", "basnet_saliency_map")
        if split == "test"
        else (
            "inpainted_poster",
            "pfpn_saliency_map",
            "basnet_saliency_map",
            "annotations",
        )
    )
    missing = [
        field for field in required if field not in example or example[field] is None
    ]
    if missing:
        raise KeyError(f"approved source lacks required input fields: {missing}")

    return {
        "inpainted_poster": example.get("inpainted_poster"),
        "canvas": example.get("canvas"),
        "saliency_pfpnet": example.get("pfpn_saliency_map"),
        "saliency_basnet": example.get("basnet_saliency_map"),
        "annotations": example.get("annotations"),
    }

build_bridge_manifest

build_bridge_manifest(
    *,
    source_revision: str,
    train_count: int,
    test_count: int,
    source_hashes: Mapping[str, str],
    source_fields: Mapping[str, str] | None = None,
    available_fields: Sequence[str] | None = None,
) -> dict[str, ManifestValue]

Build a hash-manifested record of the approved-source bridge.

Source code in models/ds-gan/src/ds_gan/training/dataset.py
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
def build_bridge_manifest(
    *,
    source_revision: str,
    train_count: int,
    test_count: int,
    source_hashes: Mapping[str, str],
    source_fields: Mapping[str, str] | None = None,
    available_fields: Sequence[str] | None = None,
) -> dict[str, ManifestValue]:
    """Build a hash-manifested record of the approved-source bridge."""
    mappings = dict(source_fields or SOURCE_FIELDS)
    expected = set(SOURCE_FIELDS)
    available = set(available_fields or mappings.values())
    missing = sorted(
        field for field in expected if mappings.get(field) not in available
    )
    return {
        "source": {
            "dataset": SOURCE_DATASET,
            "config": SOURCE_CONFIG,
            "revision": source_revision,
            "split_counts": {"train": train_count, "test": test_count},
        },
        "vendor_layout": {
            "train": {
                "inpainted_poster": mappings.get("train.inpainted_poster"),
                "saliency_pfpnet": mappings.get("train.saliency_pfpnet"),
                "saliency_basnet": mappings.get("train.saliency_basnet"),
                "annotations": mappings.get("train.annotations"),
            },
            "test": {
                "canvas": mappings.get("test.canvas"),
                "saliency_pfpnet": mappings.get("test.saliency_pfpnet"),
                "saliency_basnet": mappings.get("test.saliency_basnet"),
            },
        },
        "missing_source_fields": missing,
        "hashes": dict(source_hashes),
        "transform": {
            "resize": [350, 240],
            "saliency_merge": "pixelwise_max",
            "box_format": "ltrb_to_normalized_cxcywh",
            "canvas_size": [513, 750],
            "invalid_class": "filtered",
            "source_label_to_vendor_class": {
                "0": 1,
                "1": 2,
                "2": 3,
                "3": 0,
            },
        },
    }

hash_file

hash_file(path: Path) -> str

Hash a source metadata or arrow artifact in streaming chunks.

Source code in models/ds-gan/src/ds_gan/training/dataset.py
209
210
211
212
213
214
215
216
def hash_file(path: Path) -> str:
    """Hash a source metadata or arrow artifact in streaming chunks."""
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(chunk)

    return digest.hexdigest()

load_cached_dataset

load_cached_dataset(
    cache_dir: str | Path | None = None,
) -> DatasetDict

Load the approved source from a standard datasets cache without downloading.

Source code in models/ds-gan/src/ds_gan/training/dataset.py
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
def load_cached_dataset(cache_dir: str | Path | None = None) -> DatasetDict:
    """Load the approved source from a standard datasets cache without downloading."""
    root = Path(cache_dir or Path.home() / ".cache" / "huggingface" / "datasets")
    candidates = sorted(
        root.glob(
            "creative-graphic-design___pku-poster_layout/default/0.0.0/*/dataset_info.json"
        )
    )
    if not candidates:
        raise FileNotFoundError(
            "cached PKU PosterLayout dataset metadata is unavailable"
        )

    dataset_dir = candidates[-1].parent
    train_files = sorted(dataset_dir.glob("pku-poster_layout-train-*.arrow"))
    test_file = dataset_dir / "pku-poster_layout-test.arrow"
    if not train_files or not test_file.exists():
        raise FileNotFoundError(
            "cached PKU PosterLayout train/test Arrow files are incomplete"
        )

    train = concatenate_datasets([Dataset.from_file(str(path)) for path in train_files])
    return DatasetDict({"train": train, "test": Dataset.from_file(str(test_file))})

load_cached_test

load_cached_test(
    cache_dir: str | Path | None = None,
) -> Dataset

Load only the approved source TEST Arrow stream from cache.

Source code in models/ds-gan/src/ds_gan/training/dataset.py
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
def load_cached_test(cache_dir: str | Path | None = None) -> Dataset:
    """Load only the approved source TEST Arrow stream from cache."""
    root = Path(cache_dir or Path.home() / ".cache" / "huggingface" / "datasets")
    candidates = sorted(
        root.glob(
            "creative-graphic-design___pku-poster_layout/default/0.0.0/*/dataset_info.json"
        )
    )
    if not candidates:
        raise FileNotFoundError(
            "cached PKU PosterLayout dataset metadata is unavailable"
        )

    test_file = candidates[-1].parent / "pku-poster_layout-test.arrow"
    if not test_file.exists():
        raise FileNotFoundError(
            "cached PKU PosterLayout TEST Arrow file is unavailable"
        )

    return Dataset.from_file(str(test_file))

manifest_from_cached_dataset

manifest_from_cached_dataset(
    cache_dir: str | Path | None = None,
) -> dict[str, ManifestValue]

Create a bridge manifest from cached source metadata and Arrow hashes.

Source code in models/ds-gan/src/ds_gan/training/dataset.py
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
def manifest_from_cached_dataset(
    cache_dir: str | Path | None = None,
) -> dict[str, ManifestValue]:
    """Create a bridge manifest from cached source metadata and Arrow hashes."""
    root = Path(cache_dir or Path.home() / ".cache" / "huggingface" / "datasets")
    metadata = sorted(
        root.glob(
            "creative-graphic-design___pku-poster_layout/default/0.0.0/*/dataset_info.json"
        )
    )[-1]
    dataset_dir = metadata.parent
    train_files = sorted(dataset_dir.glob("pku-poster_layout-train-*.arrow"))
    test_file = dataset_dir / "pku-poster_layout-test.arrow"
    info = json.loads(metadata.read_text())
    return build_bridge_manifest(
        source_revision=info.get("download_checksums", {})
        .popitem()[0]
        .split("@")[1]
        .split("/")[0],
        train_count=int(info["splits"]["train"]["num_examples"]),
        test_count=int(info["splits"]["test"]["num_examples"]),
        source_hashes={
            "dataset_info.json": hash_file(metadata),
            **{path.name: hash_file(path) for path in (*train_files, test_file)},
        },
        available_fields=list(info["features"]),
    )

discriminator

Package-local DS-GAN discriminator and layout ordering.

DSGANDiscriminator

Bases: Module

Package-local discriminator matching the original DS-GAN topology.

Source code in models/ds-gan/src/ds_gan/training/discriminator.py
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
class DSGANDiscriminator(nn.Module):
    """Package-local discriminator matching the original DS-GAN topology."""

    def __init__(
        self, config: DSGANConfig, *, backbone_weights: str | Path | None = None
    ) -> None:
        """Initialize the discriminator from a ResNet-FPN and CNN-LSTM config."""
        super().__init__()
        self.resnet_fpn = ResnetBackbone(config, backbone_weights=backbone_weights)
        self.cnnlstm = CNNLSTM(config)
        self.fc_tf = nn.Linear(2 * config.hidden_size, 1)

    def forward(
        self,
        pixel_values: Float[torch.Tensor, "batch 4 height width"],
        layout: Float[torch.Tensor, "batch elements 2 4"],
    ) -> Float[torch.Tensor, "batch 1"]:
        """Score a layout after hard class selection and ordering."""
        h0 = self.resnet_fpn(pixel_values)
        ordered = _LayoutArgmax.apply(layout)
        output = self.cnnlstm(ordered, h0)[:, -1, :]
        return torch.tanh(self.fc_tf(output))
__init__
__init__(
    config: DSGANConfig,
    *,
    backbone_weights: str | Path | None = None,
) -> None

Initialize the discriminator from a ResNet-FPN and CNN-LSTM config.

Source code in models/ds-gan/src/ds_gan/training/discriminator.py
173
174
175
176
177
178
179
180
def __init__(
    self, config: DSGANConfig, *, backbone_weights: str | Path | None = None
) -> None:
    """Initialize the discriminator from a ResNet-FPN and CNN-LSTM config."""
    super().__init__()
    self.resnet_fpn = ResnetBackbone(config, backbone_weights=backbone_weights)
    self.cnnlstm = CNNLSTM(config)
    self.fc_tf = nn.Linear(2 * config.hidden_size, 1)
forward
forward(
    pixel_values: Float[Tensor, "batch 4 height width"],
    layout: Float[Tensor, "batch elements 2 4"],
) -> Float[torch.Tensor, "batch 1"]

Score a layout after hard class selection and ordering.

Source code in models/ds-gan/src/ds_gan/training/discriminator.py
182
183
184
185
186
187
188
189
190
191
def forward(
    self,
    pixel_values: Float[torch.Tensor, "batch 4 height width"],
    layout: Float[torch.Tensor, "batch elements 2 4"],
) -> Float[torch.Tensor, "batch 1"]:
    """Score a layout after hard class selection and ordering."""
    h0 = self.resnet_fpn(pixel_values)
    ordered = _LayoutArgmax.apply(layout)
    output = self.cnnlstm(ordered, h0)[:, -1, :]
    return torch.tanh(self.fc_tf(output))

lightning_module

Lightning training module matching the released DS-GAN recipe.

DSGANConfigArguments

Bases: TypedDict

JSON-compatible constructor arguments accepted by DSGANConfig.

Source code in models/ds-gan/src/ds_gan/training/lightning_module.py
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
class DSGANConfigArguments(TypedDict, total=False, closed=True):
    """JSON-compatible constructor arguments accepted by ``DSGANConfig``."""

    dataset_name: str
    backbone: str
    max_elem: int
    in_channels: int
    out_channels: int
    hidden_size: int | None
    num_layers: int
    output_size: int
    image_size: tuple[int, int] | list[int]
    reference_canvas_size: tuple[int, int] | list[int]
    backbone_feature_size: int
    model_num_classes: int
    id2label: dict[int | str, str] | None
    label2id: dict[str, int] | None
    model_subfolder: str
    processor_subfolder: str
    condition_types: list[str] | tuple[str, ...] | None
    architectures: list[str] | None
    model_type: str | None
    transformers_version: str | None
    torch_dtype: str | None
    dtype: str | None
    name_or_path: str
    _commit_hash: str | None

DSGANTrainingModule

Bases: LightningModule

Train generator and discriminator with the reference update ordering.

Source code in models/ds-gan/src/ds_gan/training/lightning_module.py
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
class DSGANTrainingModule(LightningModule):
    """Train generator and discriminator with the reference update ordering."""

    def __init__(
        self,
        *,
        config: TrainingConfigInput,
        discriminator_config: TrainingConfigInput | None = None,
        generator: DSGANModel | None = None,
        discriminator: DSGANDiscriminator | None = None,
        generator_backbone_weights: str | None = None,
        discriminator_backbone_weights: str | None = None,
        generator_learning_rate: float = 1e-4,
        discriminator_learning_rate: float = 1e-3,
        generator_backbone_learning_rate: float = 1e-5,
        discriminator_backbone_learning_rate: float = 1e-4,
        adversarial_weight_ramp_steps: int = 100,
        scheduler_gamma: float = 0.8,
        scheduler_generator_milestones: tuple[int, ...] = (0, 50, 100, 150, 200, 250),
        scheduler_discriminator_milestones: tuple[int, ...] = (
            0,
            25,
            50,
            75,
            100,
            125,
            150,
            175,
            200,
            225,
            250,
            275,
        ),
        seed: int = 0,
    ) -> None:
        """Initialize the two-network DS-GAN training state."""
        super().__init__()
        resolved_config = _resolve_config(config)
        resolved_discriminator_config = (
            _resolve_config(discriminator_config)
            if discriminator_config is not None
            else None
        )
        self.save_hyperparameters(
            ignore=(
                "generator",
                "discriminator",
                "generator_backbone_weights",
                "discriminator_backbone_weights",
            )
        )
        self.ds_gan_config = resolved_config

        if resolved_discriminator_config is None:
            resolved_discriminator_config = DSGANConfig(
                dataset_name=resolved_config.dataset_name,
                backbone="resnet18",
                max_elem=resolved_config.max_elem,
                in_channels=resolved_config.in_channels,
                out_channels=resolved_config.out_channels,
                hidden_size=resolved_config.hidden_size,
                num_layers=2,
                output_size=resolved_config.output_size,
                image_size=cast(tuple[int, int], resolved_config.image_size),
                reference_canvas_size=cast(
                    tuple[int, int], resolved_config.reference_canvas_size
                ),
                backbone_feature_size=resolved_config.backbone_feature_size,
            )

        self.generator = generator or DSGANModel(
            resolved_config, backbone_weights=generator_backbone_weights
        )
        self.discriminator = discriminator or DSGANDiscriminator(
            resolved_discriminator_config,
            backbone_weights=discriminator_backbone_weights,
        )
        self.criterion = DSGANSetCriterion()
        self.generator_learning_rate = generator_learning_rate
        self.discriminator_learning_rate = discriminator_learning_rate
        self.generator_backbone_learning_rate = generator_backbone_learning_rate
        self.discriminator_backbone_learning_rate = discriminator_backbone_learning_rate
        self.adversarial_weight_ramp_steps = adversarial_weight_ramp_steps
        self.scheduler_gamma = scheduler_gamma
        self.scheduler_generator_milestones = scheduler_generator_milestones
        self.scheduler_discriminator_milestones = scheduler_discriminator_milestones
        self.seed = seed
        self.automatic_optimization = False

        self._numpy_rng = np.random.RandomState(seed)
        self._torch_generator = torch.Generator(device="cpu")
        self._torch_generator.manual_seed(seed)
        self.latest_step_trace: dict[str, Shaped[torch.Tensor, "..."]] = {}
        self._step_index = 0

    def forward(
        self,
        pixel_values: Float[torch.Tensor, "batch 4 height width"],
        initial_layout: Float[torch.Tensor, "batch elements 2 4"],
    ) -> DSGANModelOutput:
        """Run the package generator."""
        output = self.generator(pixel_values=pixel_values, layout=initial_layout)
        if not isinstance(output, DSGANModelOutput):
            raise TypeError("DS-GAN generator must return DSGANModelOutput")

        return output

    def on_fit_start(self) -> None:
        """Share the data module's seeded stream with layout initialization."""
        datamodule = getattr(self.trainer, "datamodule", None)
        if datamodule is None:
            return

        generator = getattr(datamodule, "training_generator", None)
        if isinstance(generator, torch.Generator):
            self._torch_generator = generator

    def configure_optimizers(self) -> OptimizerLRScheduler:
        """Return Adam optimizers and reference MultiStepLR schedulers."""
        generator_backbone, generator_head = self._parameter_groups(self.generator)
        discriminator_backbone, discriminator_head = self._parameter_groups(
            self.discriminator
        )
        optimizer_g = torch.optim.Adam(
            [
                {"params": generator_head, "lr": self.generator_learning_rate},
                {
                    "params": generator_backbone,
                    "lr": self.generator_backbone_learning_rate,
                },
            ]
        )
        optimizer_d = torch.optim.Adam(
            [
                {"params": discriminator_head, "lr": self.discriminator_learning_rate},
                {
                    "params": discriminator_backbone,
                    "lr": self.discriminator_backbone_learning_rate,
                },
            ]
        )
        scheduler_g = torch.optim.lr_scheduler.MultiStepLR(
            optimizer_g,
            milestones=self.scheduler_generator_milestones,
            gamma=self.scheduler_gamma,
        )
        scheduler_d = torch.optim.lr_scheduler.MultiStepLR(
            optimizer_d,
            milestones=self.scheduler_discriminator_milestones,
            gamma=self.scheduler_gamma,
        )
        scheduler_configs: list[LRSchedulerConfig] = [
            cast(
                LRSchedulerConfig,
                {"scheduler": scheduler_g, "interval": "epoch", "frequency": 1},
            ),
            cast(
                LRSchedulerConfig,
                {"scheduler": scheduler_d, "interval": "epoch", "frequency": 1},
            ),
        ]
        return [optimizer_g, optimizer_d], scheduler_configs

    def training_step(
        self, batch: Mapping[str, Shaped[torch.Tensor, "..."]], batch_idx: int
    ) -> Float[torch.Tensor, ""]:
        """Run one manually optimized generator-then-discriminator update."""
        del batch_idx
        optimizers = self.optimizers()
        if not isinstance(optimizers, list) or len(optimizers) != 2:
            raise RuntimeError("DS-GAN requires two Lightning optimizers")

        optimizer_g = cast(torch.optim.Optimizer, optimizers[0])
        optimizer_d = cast(torch.optim.Optimizer, optimizers[1])
        return self.step_with_optimizers(
            batch,
            optimizer_g,
            optimizer_d,
            numpy_rng=self._numpy_rng,
            torch_generator=self._torch_generator,
            epoch=self.current_epoch + 1,
        )

    def step_with_optimizers(
        self,
        batch: Mapping[str, Shaped[torch.Tensor, "..."]],
        optimizer_g: torch.optim.Optimizer,
        optimizer_d: torch.optim.Optimizer,
        *,
        initial_layout: Float[torch.Tensor, "batch elements 2 4"] | None = None,
        numpy_rng: np.random.RandomState | None = None,
        torch_generator: torch.Generator | None = None,
        epoch: int = 1,
    ) -> Float[torch.Tensor, ""]:
        """Apply one exact G-then-D update and record its comparison trace."""
        pixel_values = batch["pixel_values"]
        target_layout = batch["layout"]
        batch_size = pixel_values.shape[0]
        if initial_layout is None:
            resolved_numpy_rng = numpy_rng if numpy_rng is not None else np.random
            initial_layout = vendor_random_initial_layout(
                batch_size,
                self.ds_gan_config.max_elem,
                numpy_rng=cast(NumpyChoiceSource, resolved_numpy_rng),
                torch_generator=torch_generator,
                device=pixel_values.device,
            )

        initial_layout = initial_layout.to(pixel_values.device, pixel_values.dtype)
        all_real = torch.ones(batch_size, device=pixel_values.device)
        all_fake = torch.full((batch_size,), -1.0, device=pixel_values.device)
        target_labels = batch["labels"].long()
        target_boxes = batch["boxes"].float()
        targets = [
            {"labels": labels, "boxes": boxes}
            for labels, boxes in zip(target_labels, target_boxes, strict=True)
        ]
        adversarial_weight = min(
            1.0, max(0, epoch - 1) / self.adversarial_weight_ramp_steps
        )

        optimizer_g.zero_grad(set_to_none=True)
        generated = self.forward(pixel_values, initial_layout)
        if generated.bbox is None:
            raise RuntimeError("DS-GAN generator did not return bounding boxes")

        generated_layout = torch.stack((generated.class_probs, generated.bbox), dim=2)
        discriminator_generated = self.discriminator(pixel_values, generated_layout)
        loss_g_adv = nn.functional.hinge_embedding_loss(
            discriminator_generated.reshape(-1), all_real
        )
        reconstruction = self.criterion(generated.class_probs, generated.bbox, targets)
        loss_reconstruction = sum(reconstruction.values())
        loss_g = adversarial_weight * loss_g_adv + loss_reconstruction
        self._backward(loss_g)
        self._configure_gradient_clipping(optimizer_g)
        optimizer_g.step()

        optimizer_d.zero_grad(set_to_none=True)
        discriminator_fake = self.discriminator(pixel_values, generated_layout.detach())
        discriminator_real = self.discriminator(pixel_values, target_layout.clone())
        loss_d_fake = nn.functional.hinge_embedding_loss(
            discriminator_fake.reshape(-1), all_fake
        )
        loss_d_real = nn.functional.hinge_embedding_loss(
            discriminator_real.reshape(-1), all_real
        )
        loss_d = adversarial_weight * (loss_d_real + loss_d_fake)
        self._backward(loss_d)
        self._configure_gradient_clipping(optimizer_d)
        optimizer_d.step()

        self.latest_step_trace = {
            "initial_layout": initial_layout.detach(),
            "class_probs": generated.class_probs.detach(),
            "bbox": generated.bbox.detach(),
            "discriminator_generated": discriminator_generated.detach(),
            "discriminator_fake": discriminator_fake.detach(),
            "discriminator_real": discriminator_real.detach(),
            "loss_g_adv": loss_g_adv.detach(),
            "loss_reconstruction": loss_reconstruction.detach(),
            "loss_g": loss_g.detach(),
            "loss_d_fake": loss_d_fake.detach(),
            "loss_d_real": loss_d_real.detach(),
            "loss_d": loss_d.detach(),
            "adversarial_weight": torch.tensor(
                adversarial_weight, device=pixel_values.device
            ),
        }
        self._step_index += 1
        return loss_g.detach()

    def on_train_epoch_end(self) -> None:
        """Advance both epoch schedulers after all manual updates in an epoch."""
        schedulers = self.lr_schedulers()
        if isinstance(schedulers, list):
            for scheduler in cast(
                list[torch.optim.lr_scheduler.MultiStepLR], schedulers
            ):
                scheduler.step()

    @staticmethod
    def _parameter_groups(
        module: nn.Module,
    ) -> tuple[list[nn.Parameter], list[nn.Parameter]]:
        """Partition ResNet parameters from the trainable head."""
        backbone: list[nn.Parameter] = []
        head: list[nn.Parameter] = []
        for name, parameter in module.named_parameters():
            (backbone if name.startswith("resnet_fpn") else head).append(parameter)

        return backbone, head

    def _backward(self, loss: Float[torch.Tensor, ""]) -> None:
        """Backpropagate through Lightning or a direct parity harness."""
        if self._trainer is None:
            loss.backward()
            return

        self.manual_backward(loss)

    def _configure_gradient_clipping(self, optimizer: torch.optim.Optimizer) -> None:
        if self._trainer is None:
            return

        self.configure_gradient_clipping(
            optimizer,
            gradient_clip_val=self.trainer.gradient_clip_val,
            gradient_clip_algorithm=self.trainer.gradient_clip_algorithm,
        )

    def configure_gradient_clipping(
        self,
        optimizer: torch.optim.Optimizer,
        gradient_clip_val: float | None = None,
        gradient_clip_algorithm: str | None = None,
    ) -> None:
        """Delegate the Trainer's clipping policy for the active optimizer."""
        super().configure_gradient_clipping(
            optimizer,
            gradient_clip_val=gradient_clip_val,
            gradient_clip_algorithm=gradient_clip_algorithm,
        )
__init__
__init__(
    *,
    config: TrainingConfigInput,
    discriminator_config: TrainingConfigInput | None = None,
    generator: DSGANModel | None = None,
    discriminator: DSGANDiscriminator | None = None,
    generator_backbone_weights: str | None = None,
    discriminator_backbone_weights: str | None = None,
    generator_learning_rate: float = 0.0001,
    discriminator_learning_rate: float = 0.001,
    generator_backbone_learning_rate: float = 1e-05,
    discriminator_backbone_learning_rate: float = 0.0001,
    adversarial_weight_ramp_steps: int = 100,
    scheduler_gamma: float = 0.8,
    scheduler_generator_milestones: tuple[int, ...] = (
        0,
        50,
        100,
        150,
        200,
        250,
    ),
    scheduler_discriminator_milestones: tuple[int, ...] = (
        0,
        25,
        50,
        75,
        100,
        125,
        150,
        175,
        200,
        225,
        250,
        275,
    ),
    seed: int = 0,
) -> None

Initialize the two-network DS-GAN training state.

Source code in models/ds-gan/src/ds_gan/training/lightning_module.py
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
def __init__(
    self,
    *,
    config: TrainingConfigInput,
    discriminator_config: TrainingConfigInput | None = None,
    generator: DSGANModel | None = None,
    discriminator: DSGANDiscriminator | None = None,
    generator_backbone_weights: str | None = None,
    discriminator_backbone_weights: str | None = None,
    generator_learning_rate: float = 1e-4,
    discriminator_learning_rate: float = 1e-3,
    generator_backbone_learning_rate: float = 1e-5,
    discriminator_backbone_learning_rate: float = 1e-4,
    adversarial_weight_ramp_steps: int = 100,
    scheduler_gamma: float = 0.8,
    scheduler_generator_milestones: tuple[int, ...] = (0, 50, 100, 150, 200, 250),
    scheduler_discriminator_milestones: tuple[int, ...] = (
        0,
        25,
        50,
        75,
        100,
        125,
        150,
        175,
        200,
        225,
        250,
        275,
    ),
    seed: int = 0,
) -> None:
    """Initialize the two-network DS-GAN training state."""
    super().__init__()
    resolved_config = _resolve_config(config)
    resolved_discriminator_config = (
        _resolve_config(discriminator_config)
        if discriminator_config is not None
        else None
    )
    self.save_hyperparameters(
        ignore=(
            "generator",
            "discriminator",
            "generator_backbone_weights",
            "discriminator_backbone_weights",
        )
    )
    self.ds_gan_config = resolved_config

    if resolved_discriminator_config is None:
        resolved_discriminator_config = DSGANConfig(
            dataset_name=resolved_config.dataset_name,
            backbone="resnet18",
            max_elem=resolved_config.max_elem,
            in_channels=resolved_config.in_channels,
            out_channels=resolved_config.out_channels,
            hidden_size=resolved_config.hidden_size,
            num_layers=2,
            output_size=resolved_config.output_size,
            image_size=cast(tuple[int, int], resolved_config.image_size),
            reference_canvas_size=cast(
                tuple[int, int], resolved_config.reference_canvas_size
            ),
            backbone_feature_size=resolved_config.backbone_feature_size,
        )

    self.generator = generator or DSGANModel(
        resolved_config, backbone_weights=generator_backbone_weights
    )
    self.discriminator = discriminator or DSGANDiscriminator(
        resolved_discriminator_config,
        backbone_weights=discriminator_backbone_weights,
    )
    self.criterion = DSGANSetCriterion()
    self.generator_learning_rate = generator_learning_rate
    self.discriminator_learning_rate = discriminator_learning_rate
    self.generator_backbone_learning_rate = generator_backbone_learning_rate
    self.discriminator_backbone_learning_rate = discriminator_backbone_learning_rate
    self.adversarial_weight_ramp_steps = adversarial_weight_ramp_steps
    self.scheduler_gamma = scheduler_gamma
    self.scheduler_generator_milestones = scheduler_generator_milestones
    self.scheduler_discriminator_milestones = scheduler_discriminator_milestones
    self.seed = seed
    self.automatic_optimization = False

    self._numpy_rng = np.random.RandomState(seed)
    self._torch_generator = torch.Generator(device="cpu")
    self._torch_generator.manual_seed(seed)
    self.latest_step_trace: dict[str, Shaped[torch.Tensor, "..."]] = {}
    self._step_index = 0
forward
forward(
    pixel_values: Float[Tensor, "batch 4 height width"],
    initial_layout: Float[Tensor, "batch elements 2 4"],
) -> DSGANModelOutput

Run the package generator.

Source code in models/ds-gan/src/ds_gan/training/lightning_module.py
175
176
177
178
179
180
181
182
183
184
185
def forward(
    self,
    pixel_values: Float[torch.Tensor, "batch 4 height width"],
    initial_layout: Float[torch.Tensor, "batch elements 2 4"],
) -> DSGANModelOutput:
    """Run the package generator."""
    output = self.generator(pixel_values=pixel_values, layout=initial_layout)
    if not isinstance(output, DSGANModelOutput):
        raise TypeError("DS-GAN generator must return DSGANModelOutput")

    return output
on_fit_start
on_fit_start() -> None

Share the data module's seeded stream with layout initialization.

Source code in models/ds-gan/src/ds_gan/training/lightning_module.py
187
188
189
190
191
192
193
194
195
def on_fit_start(self) -> None:
    """Share the data module's seeded stream with layout initialization."""
    datamodule = getattr(self.trainer, "datamodule", None)
    if datamodule is None:
        return

    generator = getattr(datamodule, "training_generator", None)
    if isinstance(generator, torch.Generator):
        self._torch_generator = generator
configure_optimizers
configure_optimizers() -> OptimizerLRScheduler

Return Adam optimizers and reference MultiStepLR schedulers.

Source code in models/ds-gan/src/ds_gan/training/lightning_module.py
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
def configure_optimizers(self) -> OptimizerLRScheduler:
    """Return Adam optimizers and reference MultiStepLR schedulers."""
    generator_backbone, generator_head = self._parameter_groups(self.generator)
    discriminator_backbone, discriminator_head = self._parameter_groups(
        self.discriminator
    )
    optimizer_g = torch.optim.Adam(
        [
            {"params": generator_head, "lr": self.generator_learning_rate},
            {
                "params": generator_backbone,
                "lr": self.generator_backbone_learning_rate,
            },
        ]
    )
    optimizer_d = torch.optim.Adam(
        [
            {"params": discriminator_head, "lr": self.discriminator_learning_rate},
            {
                "params": discriminator_backbone,
                "lr": self.discriminator_backbone_learning_rate,
            },
        ]
    )
    scheduler_g = torch.optim.lr_scheduler.MultiStepLR(
        optimizer_g,
        milestones=self.scheduler_generator_milestones,
        gamma=self.scheduler_gamma,
    )
    scheduler_d = torch.optim.lr_scheduler.MultiStepLR(
        optimizer_d,
        milestones=self.scheduler_discriminator_milestones,
        gamma=self.scheduler_gamma,
    )
    scheduler_configs: list[LRSchedulerConfig] = [
        cast(
            LRSchedulerConfig,
            {"scheduler": scheduler_g, "interval": "epoch", "frequency": 1},
        ),
        cast(
            LRSchedulerConfig,
            {"scheduler": scheduler_d, "interval": "epoch", "frequency": 1},
        ),
    ]
    return [optimizer_g, optimizer_d], scheduler_configs
training_step
training_step(
    batch: Mapping[str, Shaped[Tensor, "..."]],
    batch_idx: int,
) -> Float[torch.Tensor, ""]

Run one manually optimized generator-then-discriminator update.

Source code in models/ds-gan/src/ds_gan/training/lightning_module.py
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
def training_step(
    self, batch: Mapping[str, Shaped[torch.Tensor, "..."]], batch_idx: int
) -> Float[torch.Tensor, ""]:
    """Run one manually optimized generator-then-discriminator update."""
    del batch_idx
    optimizers = self.optimizers()
    if not isinstance(optimizers, list) or len(optimizers) != 2:
        raise RuntimeError("DS-GAN requires two Lightning optimizers")

    optimizer_g = cast(torch.optim.Optimizer, optimizers[0])
    optimizer_d = cast(torch.optim.Optimizer, optimizers[1])
    return self.step_with_optimizers(
        batch,
        optimizer_g,
        optimizer_d,
        numpy_rng=self._numpy_rng,
        torch_generator=self._torch_generator,
        epoch=self.current_epoch + 1,
    )
step_with_optimizers
step_with_optimizers(
    batch: Mapping[str, Shaped[Tensor, "..."]],
    optimizer_g: Optimizer,
    optimizer_d: Optimizer,
    *,
    initial_layout: Float[Tensor, "batch elements 2 4"]
    | None = None,
    numpy_rng: RandomState | None = None,
    torch_generator: Generator | None = None,
    epoch: int = 1,
) -> Float[torch.Tensor, ""]

Apply one exact G-then-D update and record its comparison trace.

Source code in models/ds-gan/src/ds_gan/training/lightning_module.py
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
def step_with_optimizers(
    self,
    batch: Mapping[str, Shaped[torch.Tensor, "..."]],
    optimizer_g: torch.optim.Optimizer,
    optimizer_d: torch.optim.Optimizer,
    *,
    initial_layout: Float[torch.Tensor, "batch elements 2 4"] | None = None,
    numpy_rng: np.random.RandomState | None = None,
    torch_generator: torch.Generator | None = None,
    epoch: int = 1,
) -> Float[torch.Tensor, ""]:
    """Apply one exact G-then-D update and record its comparison trace."""
    pixel_values = batch["pixel_values"]
    target_layout = batch["layout"]
    batch_size = pixel_values.shape[0]
    if initial_layout is None:
        resolved_numpy_rng = numpy_rng if numpy_rng is not None else np.random
        initial_layout = vendor_random_initial_layout(
            batch_size,
            self.ds_gan_config.max_elem,
            numpy_rng=cast(NumpyChoiceSource, resolved_numpy_rng),
            torch_generator=torch_generator,
            device=pixel_values.device,
        )

    initial_layout = initial_layout.to(pixel_values.device, pixel_values.dtype)
    all_real = torch.ones(batch_size, device=pixel_values.device)
    all_fake = torch.full((batch_size,), -1.0, device=pixel_values.device)
    target_labels = batch["labels"].long()
    target_boxes = batch["boxes"].float()
    targets = [
        {"labels": labels, "boxes": boxes}
        for labels, boxes in zip(target_labels, target_boxes, strict=True)
    ]
    adversarial_weight = min(
        1.0, max(0, epoch - 1) / self.adversarial_weight_ramp_steps
    )

    optimizer_g.zero_grad(set_to_none=True)
    generated = self.forward(pixel_values, initial_layout)
    if generated.bbox is None:
        raise RuntimeError("DS-GAN generator did not return bounding boxes")

    generated_layout = torch.stack((generated.class_probs, generated.bbox), dim=2)
    discriminator_generated = self.discriminator(pixel_values, generated_layout)
    loss_g_adv = nn.functional.hinge_embedding_loss(
        discriminator_generated.reshape(-1), all_real
    )
    reconstruction = self.criterion(generated.class_probs, generated.bbox, targets)
    loss_reconstruction = sum(reconstruction.values())
    loss_g = adversarial_weight * loss_g_adv + loss_reconstruction
    self._backward(loss_g)
    self._configure_gradient_clipping(optimizer_g)
    optimizer_g.step()

    optimizer_d.zero_grad(set_to_none=True)
    discriminator_fake = self.discriminator(pixel_values, generated_layout.detach())
    discriminator_real = self.discriminator(pixel_values, target_layout.clone())
    loss_d_fake = nn.functional.hinge_embedding_loss(
        discriminator_fake.reshape(-1), all_fake
    )
    loss_d_real = nn.functional.hinge_embedding_loss(
        discriminator_real.reshape(-1), all_real
    )
    loss_d = adversarial_weight * (loss_d_real + loss_d_fake)
    self._backward(loss_d)
    self._configure_gradient_clipping(optimizer_d)
    optimizer_d.step()

    self.latest_step_trace = {
        "initial_layout": initial_layout.detach(),
        "class_probs": generated.class_probs.detach(),
        "bbox": generated.bbox.detach(),
        "discriminator_generated": discriminator_generated.detach(),
        "discriminator_fake": discriminator_fake.detach(),
        "discriminator_real": discriminator_real.detach(),
        "loss_g_adv": loss_g_adv.detach(),
        "loss_reconstruction": loss_reconstruction.detach(),
        "loss_g": loss_g.detach(),
        "loss_d_fake": loss_d_fake.detach(),
        "loss_d_real": loss_d_real.detach(),
        "loss_d": loss_d.detach(),
        "adversarial_weight": torch.tensor(
            adversarial_weight, device=pixel_values.device
        ),
    }
    self._step_index += 1
    return loss_g.detach()
on_train_epoch_end
on_train_epoch_end() -> None

Advance both epoch schedulers after all manual updates in an epoch.

Source code in models/ds-gan/src/ds_gan/training/lightning_module.py
352
353
354
355
356
357
358
359
def on_train_epoch_end(self) -> None:
    """Advance both epoch schedulers after all manual updates in an epoch."""
    schedulers = self.lr_schedulers()
    if isinstance(schedulers, list):
        for scheduler in cast(
            list[torch.optim.lr_scheduler.MultiStepLR], schedulers
        ):
            scheduler.step()
configure_gradient_clipping
configure_gradient_clipping(
    optimizer: Optimizer,
    gradient_clip_val: float | None = None,
    gradient_clip_algorithm: str | None = None,
) -> None

Delegate the Trainer's clipping policy for the active optimizer.

Source code in models/ds-gan/src/ds_gan/training/lightning_module.py
391
392
393
394
395
396
397
398
399
400
401
402
def configure_gradient_clipping(
    self,
    optimizer: torch.optim.Optimizer,
    gradient_clip_val: float | None = None,
    gradient_clip_algorithm: str | None = None,
) -> None:
    """Delegate the Trainer's clipping policy for the active optimizer."""
    super().configure_gradient_clipping(
        optimizer,
        gradient_clip_val=gradient_clip_val,
        gradient_clip_algorithm=gradient_clip_algorithm,
    )

losses

DS-GAN reconstruction losses matching the reference criterion.

HungarianMatcher

Bases: Module

Match predicted layout slots to target elements.

Source code in models/ds-gan/src/ds_gan/training/losses.py
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
class HungarianMatcher(nn.Module):
    """Match predicted layout slots to target elements."""

    def __init__(
        self, cost_class: float = 2.0, cost_bbox: float = 5.0, cost_giou: float = 2.0
    ) -> None:
        """Initialize weighted matching costs."""
        super().__init__()
        if cost_class == 0 and cost_bbox == 0 and cost_giou == 0:
            raise ValueError("at least one matching cost must be non-zero")

        self.cost_class = cost_class
        self.cost_bbox = cost_bbox
        self.cost_giou = cost_giou

    @torch.no_grad()
    def forward(
        self,
        outputs: dict[str, Float[torch.Tensor, "batch elements channels"]],
        targets: Sequence[dict[str, Shaped[torch.Tensor, "elements ..."]]],
    ) -> list[tuple[Int[torch.Tensor, "matched"], Int[torch.Tensor, "matched"]]]:
        """Return per-example Hungarian assignments."""
        logits = outputs["pred_logits"]
        boxes = outputs["pred_boxes"]
        batch_size, num_queries = logits.shape[:2]
        probabilities = logits.flatten(0, 1).softmax(-1)
        target_labels = torch.cat([target["labels"] for target in targets])
        target_boxes = torch.cat([target["boxes"] for target in targets])
        class_cost = -probabilities[:, target_labels]
        bbox_cost = torch.cdist(boxes.flatten(0, 1), target_boxes, p=1)
        giou_cost = -_generalized_box_iou(
            _cxcywh_to_xyxy(boxes.flatten(0, 1)), _cxcywh_to_xyxy(target_boxes)
        )
        cost = (
            self.cost_bbox * bbox_cost
            + self.cost_class * class_cost
            + self.cost_giou * giou_cost
        )
        cost = cost.reshape(batch_size, num_queries, -1).cpu()
        sizes = [len(target["boxes"]) for target in targets]
        assignments: list[
            tuple[Int[torch.Tensor, "matched"], Int[torch.Tensor, "matched"]]
        ] = []
        offset = 0
        for index, size in enumerate(sizes):
            matrix = cost[index, :, offset : offset + size]
            row, column = linear_sum_assignment(matrix)
            assignments.append(
                (
                    torch.as_tensor(row, dtype=torch.long),
                    torch.as_tensor(column, dtype=torch.long),
                )
            )
            offset += size

        return assignments
__init__
__init__(
    cost_class: float = 2.0,
    cost_bbox: float = 5.0,
    cost_giou: float = 2.0,
) -> None

Initialize weighted matching costs.

Source code in models/ds-gan/src/ds_gan/training/losses.py
55
56
57
58
59
60
61
62
63
64
65
def __init__(
    self, cost_class: float = 2.0, cost_bbox: float = 5.0, cost_giou: float = 2.0
) -> None:
    """Initialize weighted matching costs."""
    super().__init__()
    if cost_class == 0 and cost_bbox == 0 and cost_giou == 0:
        raise ValueError("at least one matching cost must be non-zero")

    self.cost_class = cost_class
    self.cost_bbox = cost_bbox
    self.cost_giou = cost_giou
forward
forward(
    outputs: dict[
        str, Float[Tensor, "batch elements channels"]
    ],
    targets: Sequence[
        dict[str, Shaped[Tensor, "elements ..."]]
    ],
) -> list[
    tuple[
        Int[torch.Tensor, "matched"],
        Int[torch.Tensor, "matched"],
    ]
]

Return per-example Hungarian assignments.

Source code in models/ds-gan/src/ds_gan/training/losses.py
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
@torch.no_grad()
def forward(
    self,
    outputs: dict[str, Float[torch.Tensor, "batch elements channels"]],
    targets: Sequence[dict[str, Shaped[torch.Tensor, "elements ..."]]],
) -> list[tuple[Int[torch.Tensor, "matched"], Int[torch.Tensor, "matched"]]]:
    """Return per-example Hungarian assignments."""
    logits = outputs["pred_logits"]
    boxes = outputs["pred_boxes"]
    batch_size, num_queries = logits.shape[:2]
    probabilities = logits.flatten(0, 1).softmax(-1)
    target_labels = torch.cat([target["labels"] for target in targets])
    target_boxes = torch.cat([target["boxes"] for target in targets])
    class_cost = -probabilities[:, target_labels]
    bbox_cost = torch.cdist(boxes.flatten(0, 1), target_boxes, p=1)
    giou_cost = -_generalized_box_iou(
        _cxcywh_to_xyxy(boxes.flatten(0, 1)), _cxcywh_to_xyxy(target_boxes)
    )
    cost = (
        self.cost_bbox * bbox_cost
        + self.cost_class * class_cost
        + self.cost_giou * giou_cost
    )
    cost = cost.reshape(batch_size, num_queries, -1).cpu()
    sizes = [len(target["boxes"]) for target in targets]
    assignments: list[
        tuple[Int[torch.Tensor, "matched"], Int[torch.Tensor, "matched"]]
    ] = []
    offset = 0
    for index, size in enumerate(sizes):
        matrix = cost[index, :, offset : offset + size]
        row, column = linear_sum_assignment(matrix)
        assignments.append(
            (
                torch.as_tensor(row, dtype=torch.long),
                torch.as_tensor(column, dtype=torch.long),
            )
        )
        offset += size

    return assignments

DSGANSetCriterion

Bases: Module

Compute the reference class, box, and generalized-IoU losses.

Source code in models/ds-gan/src/ds_gan/training/losses.py
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
class DSGANSetCriterion(nn.Module):
    """Compute the reference class, box, and generalized-IoU losses."""

    def __init__(self, eos_weight: float = 0.1) -> None:
        """Initialize the fixed reference loss weights."""
        super().__init__()
        self.matcher = HungarianMatcher()
        self.register_buffer("empty_weight", torch.tensor((eos_weight, 0.8, 1.0, 1.0)))

    def forward(
        self,
        pred_logits: Float[torch.Tensor, "batch elements 4"],
        pred_boxes: Float[torch.Tensor, "batch elements 4"],
        targets: Sequence[dict[str, Shaped[torch.Tensor, "elements ..."]]],
    ) -> dict[str, Float[torch.Tensor, ""]]:
        """Return the three weighted reconstruction components."""
        outputs = {"pred_logits": pred_logits, "pred_boxes": pred_boxes}
        indices = self.matcher(outputs, targets)
        target_count = (
            torch.as_tensor(
                [sum(len(target["labels"]) for target in targets)],
                dtype=torch.float,
                device=pred_logits.device,
            )
            .clamp_min(1)
            .item()
        )
        batch_indices = torch.cat(
            [
                torch.full_like(source, index)
                for index, (source, _) in enumerate(indices)
            ]
        )
        source_indices = torch.cat([source for source, _ in indices])
        target_labels = torch.cat(
            [
                target["labels"][target_indices]
                for target, (_, target_indices) in zip(targets, indices, strict=True)
            ]
        )
        target_classes = torch.full(
            pred_logits.shape[:2],
            3,
            dtype=torch.long,
            device=pred_logits.device,
        )
        target_classes[batch_indices, source_indices] = target_labels
        loss_ce = torch.nn.functional.cross_entropy(
            pred_logits.transpose(1, 2),
            target_classes,
            cast(torch.Tensor, self.empty_weight),
        )
        source_boxes = pred_boxes[batch_indices, source_indices]
        target_boxes = torch.cat(
            [
                target["boxes"][target_indices]
                for target, (_, target_indices) in zip(targets, indices, strict=True)
            ]
        )
        loss_bbox = (
            torch.nn.functional.l1_loss(
                source_boxes, target_boxes, reduction="none"
            ).sum()
            / target_count
        )
        loss_giou = (
            1
            - torch.diag(
                _generalized_box_iou(
                    _cxcywh_to_xyxy(source_boxes), _cxcywh_to_xyxy(target_boxes)
                )
            )
        ).sum() / target_count
        return {"loss_ce": loss_ce, "loss_bbox": loss_bbox, "loss_giou": loss_giou}
__init__
__init__(eos_weight: float = 0.1) -> None

Initialize the fixed reference loss weights.

Source code in models/ds-gan/src/ds_gan/training/losses.py
113
114
115
116
117
def __init__(self, eos_weight: float = 0.1) -> None:
    """Initialize the fixed reference loss weights."""
    super().__init__()
    self.matcher = HungarianMatcher()
    self.register_buffer("empty_weight", torch.tensor((eos_weight, 0.8, 1.0, 1.0)))
forward
forward(
    pred_logits: Float[Tensor, "batch elements 4"],
    pred_boxes: Float[Tensor, "batch elements 4"],
    targets: Sequence[
        dict[str, Shaped[Tensor, "elements ..."]]
    ],
) -> dict[str, Float[torch.Tensor, ""]]

Return the three weighted reconstruction components.

Source code in models/ds-gan/src/ds_gan/training/losses.py
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
def forward(
    self,
    pred_logits: Float[torch.Tensor, "batch elements 4"],
    pred_boxes: Float[torch.Tensor, "batch elements 4"],
    targets: Sequence[dict[str, Shaped[torch.Tensor, "elements ..."]]],
) -> dict[str, Float[torch.Tensor, ""]]:
    """Return the three weighted reconstruction components."""
    outputs = {"pred_logits": pred_logits, "pred_boxes": pred_boxes}
    indices = self.matcher(outputs, targets)
    target_count = (
        torch.as_tensor(
            [sum(len(target["labels"]) for target in targets)],
            dtype=torch.float,
            device=pred_logits.device,
        )
        .clamp_min(1)
        .item()
    )
    batch_indices = torch.cat(
        [
            torch.full_like(source, index)
            for index, (source, _) in enumerate(indices)
        ]
    )
    source_indices = torch.cat([source for source, _ in indices])
    target_labels = torch.cat(
        [
            target["labels"][target_indices]
            for target, (_, target_indices) in zip(targets, indices, strict=True)
        ]
    )
    target_classes = torch.full(
        pred_logits.shape[:2],
        3,
        dtype=torch.long,
        device=pred_logits.device,
    )
    target_classes[batch_indices, source_indices] = target_labels
    loss_ce = torch.nn.functional.cross_entropy(
        pred_logits.transpose(1, 2),
        target_classes,
        cast(torch.Tensor, self.empty_weight),
    )
    source_boxes = pred_boxes[batch_indices, source_indices]
    target_boxes = torch.cat(
        [
            target["boxes"][target_indices]
            for target, (_, target_indices) in zip(targets, indices, strict=True)
        ]
    )
    loss_bbox = (
        torch.nn.functional.l1_loss(
            source_boxes, target_boxes, reduction="none"
        ).sum()
        / target_count
    )
    loss_giou = (
        1
        - torch.diag(
            _generalized_box_iou(
                _cxcywh_to_xyxy(source_boxes), _cxcywh_to_xyxy(target_boxes)
            )
        )
    ).sum() / target_count
    return {"loss_ce": loss_ce, "loss_bbox": loss_bbox, "loss_giou": loss_giou}

rng

Deterministic random-state helpers for DS-GAN training.

NumpyChoiceSource

Bases: Protocol

NumPy random source exposing the legacy weighted choice operation.

Source code in models/ds-gan/src/ds_gan/training/rng.py
16
17
18
19
20
21
22
23
24
25
26
class NumpyChoiceSource(Protocol):
    """NumPy random source exposing the legacy weighted choice operation."""

    def choice(
        self,
        a: int,
        *,
        size: tuple[int, int, int],
        p: Float[np.ndarray, "4"],
    ) -> Int[np.ndarray, "batch elements 1"]:
        """Draw weighted class ids."""
choice
choice(
    a: int,
    *,
    size: tuple[int, int, int],
    p: Float[ndarray, "4"],
) -> Int[np.ndarray, "batch elements 1"]

Draw weighted class ids.

Source code in models/ds-gan/src/ds_gan/training/rng.py
19
20
21
22
23
24
25
26
def choice(
    self,
    a: int,
    *,
    size: tuple[int, int, int],
    p: Float[np.ndarray, "4"],
) -> Int[np.ndarray, "batch elements 1"]:
    """Draw weighted class ids."""

explicit_setup_seed

explicit_setup_seed(
    seed: int, *, deterministic: bool = True
) -> None

Set every random source used by the reference training recipe.

Source code in models/ds-gan/src/ds_gan/training/rng.py
29
30
31
32
33
34
35
36
37
38
def explicit_setup_seed(seed: int, *, deterministic: bool = True) -> None:
    """Set every random source used by the reference training recipe."""
    random.seed(seed)
    np.random.seed(seed)
    torch.manual_seed(seed)
    if torch.cuda.is_available():
        torch.cuda.manual_seed_all(seed)

    if deterministic:
        torch.use_deterministic_algorithms(True, warn_only=True)

vendor_random_initial_layout

vendor_random_initial_layout(
    batch_size: int,
    max_elem: int,
    *,
    numpy_rng: NumpyChoiceSource,
    torch_generator: Generator | None,
    device: device | str = "cpu",
) -> Float[torch.Tensor, "batch elements 2 4"]

Reproduce the reference class and box draws with explicit RNG objects.

Source code in models/ds-gan/src/ds_gan/training/rng.py
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
def vendor_random_initial_layout(
    batch_size: int,
    max_elem: int,
    *,
    numpy_rng: NumpyChoiceSource,
    torch_generator: torch.Generator | None,
    device: torch.device | str = "cpu",
) -> Float[torch.Tensor, "batch elements 2 4"]:
    """Reproduce the reference class and box draws with explicit RNG objects."""
    class_ids = numpy_rng.choice(
        4,
        size=(batch_size, max_elem, 1),
        p=np.asarray((0.1, 0.8, 1.0, 1.0), dtype=np.float64) / 2.9,
    )
    class_tensor = torch.zeros(
        batch_size, max_elem, 4, device=device, dtype=torch.float32
    )
    class_tensor.scatter_(
        -1,
        torch.as_tensor(class_ids, device=device, dtype=torch.long),
        1.0,
    )
    box_xyxy = normal(
        0.5,
        0.15,
        size=(batch_size, max_elem, 1, 4),
        generator=torch_generator,
        device=device,
        dtype=torch.float32,
    )
    return torch.cat((class_tensor.unsqueeze(2), xyxy_to_xywh(box_xyxy)), dim=2)