| 226 | |
| 227 | class SuperNet(torch.nn.Module): |
| 228 | def __init__(self, state_dict: Dict[str, torch.Tensor]): |
| 229 | super().__init__() |
| 230 | state_dict = OrderedDict((k, state_dict[k]) for k in sorted(state_dict.keys())) |
| 231 | self.layers = torch.nn.ModuleList(state_dict.values()) |
| 232 | self.mapping = dict(enumerate(state_dict.keys())) |
| 233 | self.rev_mapping = {v: k for k, v in enumerate(state_dict.keys())} |
| 234 | |
| 235 | # .processor for unet, .self_attn for text encoder |
| 236 | self.split_keys = [".processor", ".self_attn"] |
| 237 | |
| 238 | # we add a hook to state_dict() and load_state_dict() so that the |
| 239 | # naming fits with `unet.attn_processors` |
| 240 | def map_to(module, state_dict, *args, **kwargs): |
| 241 | new_state_dict = {} |
| 242 | for key, value in state_dict.items(): |
| 243 | num = int(key.split(".")[1]) # 0 is always "layers" |
| 244 | new_key = key.replace(f"layers.{num}", module.mapping[num]) |
| 245 | new_state_dict[new_key] = value |
| 246 | |
| 247 | return new_state_dict |
| 248 | |
| 249 | def remap_key(key, state_dict): |
| 250 | for k in self.split_keys: |
| 251 | if k in key: |
| 252 | return key.split(k)[0] + k |
| 253 | return key.split('.')[0] |
| 254 | |
| 255 | def map_from(module, state_dict, *args, **kwargs): |
| 256 | all_keys = list(state_dict.keys()) |
| 257 | for key in all_keys: |
| 258 | replace_key = remap_key(key, state_dict) |
| 259 | new_key = key.replace(replace_key, f"layers.{module.rev_mapping[replace_key]}") |
| 260 | state_dict[new_key] = state_dict[key] |
| 261 | del state_dict[key] |
| 262 | |
| 263 | self._register_state_dict_hook(map_to) |
| 264 | self._register_load_state_dict_pre_hook(map_from, with_module=True) |
| 265 | |
| 266 | |
| 267 | class Zero123PlusPipeline(diffusers.StableDiffusionPipeline): |