bps

gistfile1.txt

import logging
log = logging.getLogger(__name__)

from api.request_util import extract_extension_for_key, get_kafka_comsumer
from s3_upload import helpers

from api.restplus import api
from api.body_parts_segmentation.logic.body_parts_segmentation import human_parsing_processing
from flask import request, jsonify
from flask_restplus import Resource

# create dedicated namespace for FashionAI - Body parts segmentation client
ns = api.namespace('body_parts_segmentation_client', description='Operations for Body parts segmentation client')

from kafka import KafkaConsumer
from json import loads


def get_human_parsing_process(message):
    """

    :param message:
    :return:
    """

    image_person_url = message['imagePersonURL']
    owner_id = message['ownerId']
    print("handling {}".format(image_person_url))
    body_segments, schp_segments = human_parsing_processing(image_person_url)

    # For saving with the proper colours
    schp_segments = schp_segments[:, :, ::-1]

    body_segments = body_segments[:, :, ::-1]

    key, format_key = extract_extension_for_key(image_person_url=image_person_url,
                                                user_id=owner_id,
                                                module="combine_features")
    body_segments_url = helpers.upload_numpy_array_to_s3(body_segments, key, format_key)

    key, format_key = extract_extension_for_key(image_person_url=image_person_url,
                                                user_id=owner_id,
                                                module="SCHP")
    schp_segments_url = helpers.upload_numpy_array_to_s3(schp_segments, key, format_key)

    results_json = [
        {
            "ownerId": int(owner_id),
            "combineFeaturesURL": str(body_segments_url),
            "schpURL": str(schp_segments_url)
        }
    ]

    print(results_json)


def get_bps_message():
    kafka_domain, kafka_port = get_kafka_comsumer()
    print('Connecting to Kafka consumer at port: {} domain server {}'.format(kafka_port, kafka_domain))
    # Make sure the broker is available
    consumer = KafkaConsumer(
        'bpsIn',
        bootstrap_servers=['localhost:9092'],
        auto_offset_reset='earliest',
        enable_auto_commit=True,
        group_id='bps-group',
        value_deserializer=lambda x: loads(x.decode('utf-8')))

    try:
        for message in consumer:
            print("why")
            message = message.value
            print("ua sao la vay eee {}".format(message))
    #         # get_human_parsing_process(message)
    #
    except Exception as inst:
        print(inst)

get_bps_message()