commit 7ac2f13a7557cd855fbe5066af541d63f98a44ae Author: Daekeun Date: Tue Aug 18 18:06:14 2026 +0900 first commit diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..572e70e --- /dev/null +++ b/.gitignore @@ -0,0 +1,32 @@ +# Python +__pycache__/ +*.py[cod] +*$py.class +*.so +.Python +*.egg-info/ +.eggs/ +dist/ +build/ + +# Virtual env +.venv/ +venv/ +env/ + +# Env / secrets +.env +.env.* + +# IDE / OS +.idea/ +.vscode/ +.DS_Store +Thumbs.db + +# Jupyter +.ipynb_checkpoints/ + +# ClearML / logs +.clearml/ +*.log diff --git a/README.md b/README.md new file mode 100644 index 0000000..c0fa79c --- /dev/null +++ b/README.md @@ -0,0 +1,17 @@ +# Market-1501 Person Attribute Recognition + +이미지 URL과 사람 영역(`xywh`)으로 나이, 성별, 의상 색상 등 Market-1501 속성을 추론합니다. + +`person_attr.py`와 `net_last.pth`를 같은 디렉터리에 두고 실행하세요. + +```bash +pip install -r requirements.txt +uv run python person_attr.py \ + --image_url "https://example.com/image.jpg" \ + --xywh "404,290,74,193" +``` + +- `--image_url`: 이미지 URL 또는 로컬 경로 +- `--xywh`: 좌상단 기준 `x,y,w,h` + +결과는 ClearML Task(`Person_Attribute_Recognition` / `person_attribute`)에 `final_result` artifact로 업로드됩니다. diff --git a/__pycache__/person_attribute.cpython-311.pyc b/__pycache__/person_attribute.cpython-311.pyc new file mode 100644 index 0000000..3f66709 Binary files /dev/null and b/__pycache__/person_attribute.cpython-311.pyc differ diff --git a/net_last.pth b/net_last.pth new file mode 100644 index 0000000..7456093 Binary files /dev/null and b/net_last.pth differ diff --git a/person_attr.py b/person_attr.py new file mode 100644 index 0000000..b120330 --- /dev/null +++ b/person_attr.py @@ -0,0 +1,306 @@ +"""Market-1501 Person Attribute Recognition — standalone ClearML Agent. + +Deploy as two files in the same directory: + person_attribute.py + net_last.pth +""" + +from __future__ import annotations + +import argparse +import os +from io import BytesIO +from typing import Any + +import requests +import torch +from clearml import Task +from PIL import Image +from torch import nn +from torch.nn import init +from torchvision import models +from torchvision import transforms as T + +# --------------------------------------------------------------------------- +# Market-1501 label / attribute constants (order must match training) +# --------------------------------------------------------------------------- + +NUM_LABEL = 30 + +MARKET_LABELS = [ + 'young', + 'teenager', + 'adult', + 'old', + 'backpack', + 'bag', + 'handbag', + 'clothes', + 'down', + 'up', + 'hair', + 'hat', + 'gender', + 'upblack', + 'upwhite', + 'upred', + 'uppurple', + 'upyellow', + 'upgray', + 'upblue', + 'upgreen', + 'downblack', + 'downwhite', + 'downpink', + 'downpurple', + 'downyellow', + 'downgray', + 'downblue', + 'downgreen', + 'downbrown', +] + +MARKET_ATTRIBUTES = { + 'bag': ['carrying bag', ['no', 'yes']], + 'upred': ['color of upper-body clothing', [None, 'red']], + 'upblue': ['color of upper-body clothing', [None, 'blue']], + 'hat': ['wearing hat', ['no', 'yes']], + 'downgreen': ['color of lower-body clothing', [None, 'green']], + 'downbrown': ['color of lower-body clothing', [None, 'brown']], + 'upyellow': ['color of upper-body clothing', [None, 'yellow']], + 'up': ['sleeve length', ['long sleeve', 'short sleeve']], + 'upgreen': ['color of upper-body clothing', [None, 'green']], + 'handbag': ['carrying handbag', ['no', 'yes']], + 'downgray': ['color of lower-body clothing', [None, 'gray']], + 'clothes': ['type of lower-body clothing', ['dress', 'pants']], + 'adult': ['age', [None, 'adult']], + 'downblack': ['color of lower-body clothing', [None, 'black']], + 'backpack': ['carrying backpack', ['no', 'yes']], + 'downwhite': ['color of lower-body clothing', [None, 'white']], + 'upblack': ['color of upper-body clothing', [None, 'black']], + 'gender': ['gender', ['male', 'female']], + 'downyellow': ['color of lower-body clothing', [None, 'yellow']], + 'downpink': ['color of lower-body clothing', [None, 'pink']], + 'old': ['age', [None, 'old']], + 'down': ['length of lower-body clothing', ['long lower body clothing', 'short']], + 'uppurple': ['color of upper-body clothing', [None, 'purple']], + 'downpurple': ['color of lower-body clothing', [None, 'purple']], + 'young': ['age', [None, 'young']], + 'teenager': ['age', [None, 'teenager']], + 'hair': ['hair length', ['short hair', 'long hair']], + 'downblue': ['color of lower-body clothing', [None, 'blue']], + 'upgray': ['color of upper-body clothing', [None, 'gray']], + 'upwhite': ['color of upper-body clothing', [None, 'white']], +} + +TRANSFORMS = T.Compose([ + T.Resize(size=(288, 144)), + T.ToTensor(), + T.Normalize([0.485, 0.456, 0.406], [0.229, 0.224, 0.225]), +]) + +PACKAGE_DIR = os.path.dirname(os.path.abspath(__file__)) +DEFAULT_CHECKPOINT = os.path.join(PACKAGE_DIR, 'net_last.pth') + + +# --------------------------------------------------------------------------- +# Model (inference only) +# --------------------------------------------------------------------------- + +def _weights_init_kaiming(m): + classname = m.__class__.__name__ + if classname.find('Conv') != -1: + init.kaiming_normal_(m.weight.data, a=0, mode='fan_in') + elif classname.find('Linear') != -1: + init.kaiming_normal_(m.weight.data, a=0, mode='fan_out') + init.constant_(m.bias.data, 0.0) + elif classname.find('BatchNorm1d') != -1: + init.normal_(m.weight.data, 1.0, 0.02) + init.constant_(m.bias.data, 0.0) + + +def _weights_init_classifier(m): + classname = m.__class__.__name__ + if classname.find('Linear') != -1: + init.normal_(m.weight.data, std=0.001) + init.constant_(m.bias.data, 0.0) + + +class ClassBlock(nn.Module): + def __init__(self, input_dim, class_num=1, activ='sigmoid', num_bottleneck=512): + super().__init__() + add_block = nn.Sequential( + nn.Linear(input_dim, num_bottleneck), + nn.BatchNorm1d(num_bottleneck), + nn.LeakyReLU(0.1), + nn.Dropout(p=0.5), + ) + add_block.apply(_weights_init_kaiming) + + classifier_layers = [nn.Linear(num_bottleneck, class_num)] + if activ == 'sigmoid': + classifier_layers.append(nn.Sigmoid()) + elif activ == 'softmax': + classifier_layers.append(nn.Softmax()) + elif activ != 'none': + raise AssertionError(f'Unsupported activation: {activ}') + classifier = nn.Sequential(*classifier_layers) + classifier.apply(_weights_init_classifier) + + self.add_block = add_block + self.classifier = classifier + + def forward(self, x): + x = self.add_block(x) + return self.classifier(x) + + +class PersonAttributeModel(nn.Module): + """ResNet50 + nFC, 30 Market-1501 attribute heads.""" + + def __init__(self, class_num: int = NUM_LABEL): + super().__init__() + self.class_num = class_num + backbone = models.resnet50(weights=None) + backbone.avgpool = nn.AdaptiveAvgPool2d((1, 1)) + backbone.fc = nn.Sequential() + self.features = backbone + self.num_ftrs = 2048 + for c in range(self.class_num): + self.__setattr__( + f'class_{c}', + ClassBlock(input_dim=self.num_ftrs, class_num=1, activ='sigmoid'), + ) + + def forward(self, x): + x = self.features(x) + x = x.view(x.size(0), -1) + pred_label = [self.__getattr__(f'class_{c}')(x) for c in range(self.class_num)] + return torch.cat(pred_label, dim=1) + + +# --------------------------------------------------------------------------- +# Processor +# --------------------------------------------------------------------------- + +class PersonAttribute: + """image_url + xywh(top-left) → Market person attributes.""" + + def __init__(self, checkpoint_path: str | None = None, device: str | None = None): + self.checkpoint_path = checkpoint_path or DEFAULT_CHECKPOINT + self.device = torch.device( + device if device else ('cuda' if torch.cuda.is_available() else 'cpu') + ) + self.model = self.load_model() + + def load_model(self) -> PersonAttributeModel: + if not os.path.isfile(self.checkpoint_path): + raise FileNotFoundError(f'Checkpoint not found: {self.checkpoint_path}') + model = PersonAttributeModel(NUM_LABEL) + state = torch.load(self.checkpoint_path, map_location=self.device) + model.load_state_dict(state) + model.to(self.device) + model.eval() + return model + + @staticmethod + def parse_xywh(xywh: str) -> tuple[int, int, int, int]: + parts = str(xywh).split(',') + if len(parts) != 4: + raise ValueError(f"xywh must be 'x,y,w,h', got: {xywh}") + x, y, w, h = (int(v.strip()) for v in parts) + if w <= 0 or h <= 0: + raise ValueError(f'xywh width/height must be positive, got: {xywh}') + return x, y, w, h + + @staticmethod + def load_image(image_source: str) -> Image.Image: + if image_source.startswith(('http://', 'https://')): + response = requests.get(image_source, timeout=60) + response.raise_for_status() + return Image.open(BytesIO(response.content)).convert('RGB') + if not os.path.isfile(image_source): + raise FileNotFoundError(f'Image not found: {image_source}') + return Image.open(image_source).convert('RGB') + + @staticmethod + def crop_person(image: Image.Image, xywh: tuple[int, int, int, int]) -> Image.Image: + # Top-left based xywh (same as existing pipeline / start.py) + x, y, w, h = xywh + return image.crop((x, y, x + w, y + h)) + + @staticmethod + def preprocess(image: Image.Image) -> torch.Tensor: + return TRANSFORMS(image).unsqueeze(dim=0) + + def decode(self, pred: torch.Tensor) -> dict[str, str]: + pred = pred.squeeze(dim=0) + results: dict[str, str] = {} + for idx, label_key in enumerate(MARKET_LABELS): + name, choice = MARKET_ATTRIBUTES[label_key] + value = choice[int(pred[idx].item())] + if value: + results[name] = value + return results + + def predict(self, image: Image.Image) -> dict[str, str]: + src = self.preprocess(image).to(self.device) + with torch.no_grad(): + out = self.model.forward(src) + pred = torch.gt(out, torch.ones_like(out) / 2) + return self.decode(pred) + + def process(self, image_url: str, xywh: str) -> dict[str, str]: + if not image_url or not str(image_url).strip(): + raise ValueError('image_url is required') + box = self.parse_xywh(xywh) + image = self.load_image(str(image_url).strip()) + crop = self.crop_person(image, box) + return self.predict(crop) + + @staticmethod + def format_result(attributes: dict[str, Any], status: str = 'PASS') -> dict[str, Any]: + return {'output': attributes, 'status': status} + + def run(self, image_url: str, xywh: str) -> dict[str, Any]: + attributes = self.process(image_url, xywh) + return self.format_result(attributes) + + +# --------------------------------------------------------------------------- +# ClearML entry point +# --------------------------------------------------------------------------- + +def upload_final_result(task: Task, result: dict) -> None: + task.upload_artifact(name='final_result', artifact_object=result) + + +def main() -> None: + # uv run python person_attr.py --image_url "https://acai.ketidev.kr:20443/detect/image/202606/20260619_145116_image.jpg" --xywh "404,290,74,193" + parser = argparse.ArgumentParser(description='Person Attribute Recognition (Market)') + parser.add_argument('--image_url', type=str, required=True) + parser.add_argument('--xywh', type=str, required=True) + args = parser.parse_args() + + task = Task.init( + project_name='Person_Attribute_Recognition', + task_name='person_attribute', + ) + task.connect(parser) + + engine = PersonAttribute() + attributes = engine.process(args.image_url, args.xywh) + result = PersonAttribute.format_result(attributes) + + print('\n' + '=' * 50) + print(' Person Attribute Recognition 결과 ') + print('=' * 50) + for name, value in attributes.items(): + print(f'{name}: {value}') + print('=' * 50) + + upload_final_result(task, result) + + +if __name__ == '__main__': + main() diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..a29fe27 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,5 @@ +torch +torchvision +Pillow +requests +clearml