|
| 1 | +import torch |
| 2 | +import argparse |
| 3 | +import requests |
| 4 | +import numpy as np |
| 5 | +import huggingface_hub |
| 6 | +import albumentations as A |
| 7 | +import matplotlib.pyplot as plt |
| 8 | + |
| 9 | +from PIL import Image |
| 10 | +import segmentation_models_pytorch as smp |
| 11 | + |
| 12 | + |
| 13 | +def convert_state_dict_to_smp(state_dict: dict): |
| 14 | + # fmt: off |
| 15 | + |
| 16 | + if "state_dict" in state_dict: |
| 17 | + state_dict = state_dict["state_dict"] |
| 18 | + |
| 19 | + new_state_dict = {} |
| 20 | + |
| 21 | + # Map the backbone components to the encoder |
| 22 | + keys = list(state_dict.keys()) |
| 23 | + for key in keys: |
| 24 | + if key.startswith("backbone"): |
| 25 | + new_key = key.replace("backbone", "encoder") |
| 26 | + new_state_dict[new_key] = state_dict.pop(key) |
| 27 | + |
| 28 | + |
| 29 | + # Map the linear_cX layers to MLP stages |
| 30 | + for i in range(4): |
| 31 | + base = f"decode_head.linear_c{i+1}.proj" |
| 32 | + new_state_dict[f"decoder.mlp_stage.{3-i}.linear.weight"] = state_dict.pop(f"{base}.weight") |
| 33 | + new_state_dict[f"decoder.mlp_stage.{3-i}.linear.bias"] = state_dict.pop(f"{base}.bias") |
| 34 | + |
| 35 | + # Map fuse_stage components |
| 36 | + fuse_base = "decode_head.linear_fuse" |
| 37 | + fuse_weights = { |
| 38 | + "decoder.fuse_stage.0.weight": state_dict.pop(f"{fuse_base}.conv.weight"), |
| 39 | + "decoder.fuse_stage.1.weight": state_dict.pop(f"{fuse_base}.bn.weight"), |
| 40 | + "decoder.fuse_stage.1.bias": state_dict.pop(f"{fuse_base}.bn.bias"), |
| 41 | + "decoder.fuse_stage.1.running_mean": state_dict.pop(f"{fuse_base}.bn.running_mean"), |
| 42 | + "decoder.fuse_stage.1.running_var": state_dict.pop(f"{fuse_base}.bn.running_var"), |
| 43 | + "decoder.fuse_stage.1.num_batches_tracked": state_dict.pop(f"{fuse_base}.bn.num_batches_tracked"), |
| 44 | + } |
| 45 | + new_state_dict.update(fuse_weights) |
| 46 | + |
| 47 | + # Map final layer components |
| 48 | + new_state_dict["segmentation_head.0.weight"] = state_dict.pop("decode_head.linear_pred.weight") |
| 49 | + new_state_dict["segmentation_head.0.bias"] = state_dict.pop("decode_head.linear_pred.bias") |
| 50 | + |
| 51 | + del state_dict["decode_head.conv_seg.weight"] |
| 52 | + del state_dict["decode_head.conv_seg.bias"] |
| 53 | + |
| 54 | + assert len(state_dict) == 0, f"Unmapped keys: {state_dict.keys()}" |
| 55 | + |
| 56 | + # fmt: on |
| 57 | + return new_state_dict |
| 58 | + |
| 59 | + |
| 60 | +def get_np_image(): |
| 61 | + url = "https://huggingface.co/datasets/hf-internal-testing/fixtures_ade20k/resolve/main/ADE_val_00000001.jpg" |
| 62 | + image = Image.open(requests.get(url, stream=True).raw) |
| 63 | + return np.array(image) |
| 64 | + |
| 65 | + |
| 66 | +def main(args): |
| 67 | + original_checkpoint = torch.load(args.path, map_location="cpu", weights_only=True) |
| 68 | + smp_state_dict = convert_state_dict_to_smp(original_checkpoint) |
| 69 | + |
| 70 | + config = original_checkpoint["meta"]["config"] |
| 71 | + num_classes = int(config.split("num_classes=")[1].split(",\n")[0]) |
| 72 | + decoder_dims = int(config.split("embed_dim=")[1].split(",\n")[0]) |
| 73 | + height, width = [ |
| 74 | + int(x) for x in config.split("crop_size=(")[1].split("), ")[0].split(",") |
| 75 | + ] |
| 76 | + model_size = args.path.split("segformer.")[1][:2] |
| 77 | + |
| 78 | + # Create the model |
| 79 | + model = smp.create_model( |
| 80 | + in_channels=3, |
| 81 | + classes=num_classes, |
| 82 | + arch="segformer", |
| 83 | + encoder_name=f"mit_{model_size}", |
| 84 | + encoder_weights=None, |
| 85 | + decoder_segmentation_channels=decoder_dims, |
| 86 | + ).eval() |
| 87 | + |
| 88 | + # Load the converted state dict |
| 89 | + model.load_state_dict(smp_state_dict, strict=True) |
| 90 | + |
| 91 | + # Preprocessing params |
| 92 | + preprocessing = A.Compose( |
| 93 | + [ |
| 94 | + A.Resize(height, width, p=1), |
| 95 | + A.Normalize( |
| 96 | + mean=[123.675, 116.28, 103.53], |
| 97 | + std=[58.395, 57.12, 57.375], |
| 98 | + max_pixel_value=1.0, |
| 99 | + p=1, |
| 100 | + ), |
| 101 | + ] |
| 102 | + ) |
| 103 | + |
| 104 | + # Prepare the input |
| 105 | + image = get_np_image() |
| 106 | + normalized_image = preprocessing(image=image)["image"] |
| 107 | + tensor = torch.tensor(normalized_image).permute(2, 0, 1).unsqueeze(0).float() |
| 108 | + |
| 109 | + # Forward pass |
| 110 | + with torch.no_grad(): |
| 111 | + mask = model(tensor) |
| 112 | + |
| 113 | + # Postprocessing |
| 114 | + mask = torch.nn.functional.interpolate( |
| 115 | + mask, size=(image.shape[0], image.shape[1]), mode="bilinear" |
| 116 | + ) |
| 117 | + mask = torch.argmax(mask, dim=1) |
| 118 | + mask = mask.squeeze().cpu().numpy() |
| 119 | + |
| 120 | + model_name = args.path.split("/")[-1].replace(".pth", "").replace(".", "-") |
| 121 | + |
| 122 | + model.save_pretrained(model_name) |
| 123 | + preprocessing.save_pretrained(model_name) |
| 124 | + |
| 125 | + # fmt: off |
| 126 | + plt.subplot(121), plt.axis('off'), plt.imshow(image), plt.title('Input Image') |
| 127 | + plt.subplot(122), plt.axis('off'), plt.imshow(mask), plt.title('Output Mask') |
| 128 | + plt.savefig(f"{model_name}/example_mask.png") |
| 129 | + # fmt: on |
| 130 | + |
| 131 | + if args.push_to_hub: |
| 132 | + repo_id = f"smp-hub/{model_name}" |
| 133 | + api = huggingface_hub.HfApi() |
| 134 | + api.create_repo(repo_id=repo_id, repo_type="model") |
| 135 | + api.upload_folder(folder_path=model_name, repo_id=repo_id) |
| 136 | + |
| 137 | + |
| 138 | +if __name__ == "__main__": |
| 139 | + parser = argparse.ArgumentParser() |
| 140 | + parser.add_argument( |
| 141 | + "--path", |
| 142 | + type=str, |
| 143 | + default="weights/trained_models/segformer.b2.512x512.ade.160k.pth", |
| 144 | + ) |
| 145 | + parser.add_argument("--push_to_hub", action="store_true") |
| 146 | + args = parser.parse_args() |
| 147 | + |
| 148 | + main(args) |
0 commit comments