Services

---

Main functionality

---

Main business logic implementation

Services

Main business logic implementation

Parameters:

Name	Type	Description	Default
tagger_net	Classifier	tagger model predictor	tagger

exterior_styles_net : Classifier exterior style model predictor

encoder_net : Encoder the encoder model predictor

message_broker_client : AsyncRabbitClient message broker client used to publish embeddings

Source code in src/services.py

class Services:
    """
    Main business logic implementation

    Parameters
    ----------
    tagger_net : Classifier
        tagger model predictor

    exterior_styles_net : Classifier
        exterior style model predictor

    encoder_net : Encoder
        the encoder model predictor

    message_broker_client : AsyncRabbitClient
        message broker client used to publish embeddings
    """

    def __init__(
            self,
            tagger_net: Classifier = tagger,
            exterior_styles_net: Classifier = exterior_styles_classifier,
            encoder_net: Encoder = encoder,
            message_broker_client: AsyncRabbitClient = rabbit_client,
            preprocessor: ImagePreprocessor = image_preprocessor,
    ):

        self.tagger = tagger_net
        self.exterior_styles = exterior_styles_net
        self.encoder = encoder_net
        self.message_broker_client = message_broker_client
        self.preprocessor = preprocessor
        self.nets = [self.tagger, self.exterior_styles, self.encoder]
        self.dependencies = self.nets + [self.message_broker_client]

    async def start(self) -> None:
        """Start the long-lived dependencies (message broker connection)."""
        await self.message_broker_client.start()

    async def stop(self) -> None:
        """
        Stop the dependencies and close the module-level aiohttp sessions (TF Serving + image
        downloader). Note this makes the service one-shot per process: a second start after a
        stop would reuse the closed sessions.
        """
        await self.message_broker_client.stop()
        await session.close()
        await download_session.close()

    async def health_check(self) -> Tuple[bool, List[Dependence]]:
        """
        Checks whether all dependencies are connected and ready to work
        """
        dependencies: list[Dependence] = []
        for net in self.nets:
            connected, message = net.check_connection()
            dependencies.append(Dependence(connected=connected, message=message, name=net.name,
                                           version=getattr(net, "version", None)))
        connected, message = await self.message_broker_client.async_check_connection()
        dependencies.append(Dependence(connected=connected, message=message, name=self.message_broker_client.name,
                                       version=getattr(self.message_broker_client, "version", None)))
        healthy = all(dependence.connected for dependence in dependencies)

        return healthy, dependencies

    def info(self) -> List[dict]:
        """
        Return information on each network by calling corresponding methods
        """

        return [net.info() for net in self.nets]

    async def generate_pred_dict(
            self, network: Predictor,
            preprocessed: List[PreprocessedImage],
            ids_as_keys: bool = False,
            **kwargs
    ) -> Union[List[dict], dict]:
        """
        Make predictions with a network for multiple preprocessed images and return dictionaries with
        ids and predictions.

        Parameters
        ----------
        network : api.predictors.Predictor
            network to make predictions with
        preprocessed : List[PreprocessedImage]
            preprocessed images (from :meth:`Services.prepare_inputs`). Successful entries carry
            a resized variant per model input size; failed entries become per-image 422 responses
        ids_as_keys : bool
            whether to use ids as keys in the returned dict or include them as a value of "id" field
        **kwargs :
            additional arguments for .predict method of the network

        Returns
        -------
        out : List of dict, dict
            predictions per image with id and status
        """
        valid = [p for p in preprocessed if p.ok]
        invalid = [p for p in preprocessed if not p.ok]
        if not valid:
            # Whole batch unusable: mirror the legacy all-invalid-b64 behaviour (422 with per-id
            # detail), enriched with the specific per-image failure reasons.
            raise Base64ValidationError(
                detail=[{"invalid_ids": [p.id for p in invalid],
                         "errors": [{"id": p.id, "kind": p.error.kind.value,
                                     "message": p.error.message} for p in invalid]}],
                msg="Could not process any of the input images",
                status_code=status.HTTP_422_UNPROCESSABLE_ENTITY)

        valid_instances = [p.image_for(network.input_shape) for p in valid]

        valid_responses = await network.predict(valid_instances, **kwargs)
        invalid_responses = [PredictorResponse(status_code=p.error.status_code,
                                               message=p.error.message,
                                               scores=[]) for p in invalid]

        raw_responses = valid_responses + invalid_responses
        ids = [p.id for p in valid] + [p.id for p in invalid]
        response = [item.dict() for item in raw_responses]  # TODO make use of the PredictorResponse
        if ids_as_keys:
            return {id_: {"results": {network.name: response[i]}}
                    for i, id_ in enumerate(ids)}

        results_per_image = [{"results": {network.name: pred}}
                             for pred in response]
        predictions = add_ids_to_dicts(ids=ids, dicts=results_per_image)

        return self.postprocess_prediction_dicts(predictions)

    async def prepare_inputs(
            self,
            images: List[ImageSource],
            networks: List[Predictor]
    ) -> List[PreprocessedImage]:
        """
        Preprocess request images once for every network that will consume them.

        Parameters
        ----------
        images : List[ImageSource]
            request items, each carrying either a base64 ``image`` or a ``url``
        networks : List[Predictor]
            the networks the batch will be fed to; one resized variant is produced per distinct
            ``input_shape``, all from a single download + decode

        Returns
        -------
        out : List[PreprocessedImage]
            one entry per input item, in the same order
        """
        shapes = {net.input_shape for net in networks}
        return await self.preprocessor.preprocess(images, shapes=shapes)

    @timing
    async def predict_tags(
            self,
            images: List[ImageSource],
            custom_thresholds: Optional[Union[float, list, dict]]
    ) -> List[dict]:
        """
        Call the Tagger network with given request images and get corresponding predictions. For
        more information on parameters, refer to Services.generate_pred_dict()
        """
        preprocessed = await self.prepare_inputs(images, networks=[self.tagger])
        return await self.generate_pred_dict(network=self.tagger,
                                             preprocessed=preprocessed,
                                             custom_thresholds=custom_thresholds)

    @timing
    async def predict_exterior_styles(
            self,
            images: List[ImageSource]
    ) -> List[dict]:
        """
        Call the ExteriorStyles network with given request images and get corresponding
        predictions. For more information on parameters, refer to Services.generate_pred_dict()
        """
        preprocessed = await self.prepare_inputs(images, networks=[self.exterior_styles])
        return await self.generate_pred_dict(network=self.exterior_styles,
                                             preprocessed=preprocessed)

    @timing
    async def encode(
            self,
            images: List[ImageSource],
    ) -> List[dict]:
        """
        Encode given request images with the encoder network. For more information on parameters,
        refer to Services.generate_pred_dict()
        """
        preprocessed = await self.prepare_inputs(images, networks=[self.encoder])
        return await self.generate_pred_dict(network=self.encoder,
                                             preprocessed=preprocessed)

    @timing
    async def classify_dht(
            self,
            preprocessed: List[PreprocessedImage],
            custom_thresholds: Optional[Union[float, list, dict]]
    ) -> Tuple:
        """
        Call Tagger network, then call the Exterior Styles for the samples with "facade"
        predicted tag. Return all tags in a single dict for each sample for each network.

        The facade subset reuses the SAME :class:`PreprocessedImage` objects, so no image is
        downloaded, decoded or resized twice.
        """
        dht_preds = await self.generate_pred_dict(network=self.tagger,
                                                  preprocessed=preprocessed,
                                                  custom_thresholds=custom_thresholds,
                                                  ids_as_keys=True)
        facade_ids = {k for k, v in dht_preds.items()
                      if "facade" in v["results"][self.tagger.name]["scores"]}
        facade_preprocessed = [p for p in preprocessed if p.id in facade_ids]
        style_preds = None
        if facade_preprocessed:
            style_preds = await self.generate_pred_dict(network=self.exterior_styles,
                                                        preprocessed=facade_preprocessed,
                                                        filtered=False,
                                                        ids_as_keys=True)
            for id_ in facade_ids:
                dht_preds[id_]["results"][self.exterior_styles.name] = style_preds[id_]["results"][
                    self.exterior_styles.name].copy()
                dht_preds[id_]["results"][self.exterior_styles.name]["scores"] = argmax_dict(
                    style_preds[id_]["results"][self.exterior_styles.name]["scores"])

        return dht_preds, style_preds

    @timing
    async def process_house_images(
            self,
            images: List[ImageSource],
            custom_thresholds: Optional[Union[float, list, dict]],
            publish_embeddings: bool = False,
            core_listing_id: Any = None,
            embeddings_compression: CompressionMethod = CompressionMethod.GZIP
    ) -> Dict:
        """
        Process all images of a single house via Image Processing Networks

        The images are preprocessed ONCE for every network involved (tagger, exterior styles and —
        when publishing embeddings — the encoder), so each photo is downloaded, decoded and
        resized a single time and shared across all passes.

        Parameters
        ----------
        images : List[ImageSource]
            request house images, each carrying either a base64 ``image`` or a ``url``
        custom_thresholds : float, list, dict, optional
            per-class thresholds for the Tagger
        publish_embeddings : bool
            whether to also encode the images and publish their embeddings to the message broker
        core_listing_id : Any
            identifier of the house (used when publishing embeddings)
        embeddings_compression : CompressionMethod
            compression method to use when publishing embeddings

        Returns
        -------
        out : dict
            predictions for given house. Contains tagger's tags per image, information on present
            features, averaged scores for exterior styles and (if requested) publishing results
        """
        networks = [self.tagger, self.exterior_styles]
        if publish_embeddings:
            networks.append(self.encoder)
        preprocessed = await self.prepare_inputs(images, networks=networks)

        dht_preds, style_preds = await self.classify_dht(preprocessed=preprocessed,
                                                         custom_thresholds=custom_thresholds)
        per_image = add_ids_to_dicts(ids=dht_preds.keys(), dicts=list(dht_preds.values()))
        avg_exterior_style = self.get_average_style(style_preds)
        house_status = any(d["results"][self.tagger.name]["status"] for d in dht_preds.values())
        present_features = [tag for im in dht_preds.values()
                            for tag in im["results"][self.tagger.name]["scores"]]
        unique_features = sorted(set(present_features))
        tagged = list(filter(lambda x: x["results"][self.tagger.name]["status"], per_image))
        house_info = {"exterior_styles": avg_exterior_style,
                      "prob_ranking": self.rank_images_by_class(tagged, classes=unique_features)}

        publish_results = None
        if publish_embeddings:
            publish_results = await self.publish_embeddings(core_listing_id=core_listing_id,
                                                            preprocessed=preprocessed,
                                                            compression=embeddings_compression)

        predictions = {"results_per_image": self.postprocess_prediction_dicts(per_image),
                       "house_info": house_info,
                       "status": house_status,
                       "publish_results": publish_results}
        return predictions

    def get_average_style(self, style_predictions_per_image: dict) -> List[dict]:
        """
        Calculate average style of the house given predictions of the ExteriorStyle Classifier

        Parameters
        ----------
        style_predictions_per_image : dict
            per image predictions for exterior styles

        Returns
        -------
        out : List of dict
            average score for each possible class
        """
        avg_exterior_style = []
        if not style_predictions_per_image:
            return avg_exterior_style

        successful = list(filter(
            lambda x: x["results"][self.exterior_styles.name]["status"],
            style_predictions_per_image.values()))
        num_facades = len(successful)
        if num_facades:
            avg_exterior_style = [{"name": k,
                                   "probability": sum(
                                       i["results"][f"{self.exterior_styles.name}"]["scores"][k]
                                       for i in successful) / num_facades} for k in
                                  self.exterior_styles.output_names]

        return avg_exterior_style

    def rank_images_by_class(
            self,
            tagger_dicts: List[dict],
            classes: Iterable[str]
    ) -> List[dict]:
        """
        Rank images by Tagger's confidence that given scene exists there

        Parameters
        ----------
        tagger_dicts : List[dict]
            valid predictions per_image
        classes : Iterable[str]
            scenes to rank

        Returns
        -------
        out : List of Dict
             Scenes and list of images containing them sorted by probability (highest first)
        """

        if not classes:
            return []
        raw_scores = {cls: {d["id"]: d["results"][self.tagger.name]["scores"].get(cls, 0)
                            for d in tagger_dicts}
                      for cls in classes}
        sorted_tuples = {cls: sorted(raw_scores[cls].items(), key=lambda x: x[1],
                                     reverse=True) for cls in classes}
        ranking = [{"name": cls,
                    "ordering": [k for k, v in sorted_tuples[cls] if v > 0]}
                   for cls in classes]

        return ranking

    @staticmethod
    def postprocess_prediction_dicts(results_per_image: List[dict]) -> List[dict]:
        """
        Postprocess network prediction dictionaries

        Parameters
        ----------
        results_per_image: List
            prediction dictionaries containing scores in format {"feature" : probability}

        Returns
        -------
        out : List
            dictionaries containing scores in format {"name" : feature, "probability" : probability}
        """
        processed = results_per_image.copy()
        for result in processed:
            for network_results in result["results"].values():
                scores_per_class = network_results["scores"]
                if isinstance(scores_per_class, dict):  # in case of encoder & similar nets
                    network_results["scores"] = [{"name": cls_name,
                                                  "probability": probability}
                                                 for cls_name, probability in
                                                 scores_per_class.items()]

        return processed

    @timing
    async def prepare_embeddings(
            self,
            preprocessed: List[PreprocessedImage]
    ) -> list[PhotoEmbedding]:
        """
        Run the input data through Encoder model, filter and preprocess the output for publishing

        Parameters
        ----------
        preprocessed : List[PreprocessedImage]
            preprocessed images (from :meth:`Services.prepare_inputs`, with the encoder among the
            target networks); only successful ones are embedded

        Returns
        -------
        out : list[PhotoEmbedding]
            Vector embeddings and their ids
        """
        valid = [p for p in preprocessed if p.ok]
        photo_embeddings: list[PhotoEmbedding] = []
        if not valid:
            return photo_embeddings
        instances = [p.image_for(self.encoder.input_shape) for p in valid]
        encoder_output = await self.encoder.predict_batch(data=instances)

        # Filter out the successful ones
        for preprocessed_image, response in zip(valid, encoder_output):
            if not response.status:
                continue
            photo_embeddings.append(PhotoEmbedding(photo_id=preprocessed_image.id,
                                                   embedding=response.scores))

        return photo_embeddings

    @timing
    async def publish_embeddings(self,
                                 core_listing_id: Any,
                                 preprocessed: List[PreprocessedImage],
                                 compression: CompressionMethod) -> PublishResults:
        """
        Calculate embeddings and publish them to message broker

        Parameters
        ----------
        core_listing_id : Any
            identifier of a house
        preprocessed : List[PreprocessedImage]
            preprocessed images (from :meth:`Services.prepare_inputs`, with the encoder among the
            target networks) to embed
        compression: CompressionMethod = CompressionMethod.UNCOMPRESSED
            Compression method to use. Defaults to None

        Returns
        -------
        out : PublishResults
            Results of the publishing
        """
        embeddings = await self.prepare_embeddings(preprocessed=preprocessed)
        if not embeddings:
            return PublishResults(published=False, error="No embeddings to publish. There might be problems with the "
                                                         "encoder models or the input images.")
        payload = ListingEmbeddings(core_listing_id=core_listing_id, embeddings=embeddings)

        return await self.message_broker_client.publish(payload=payload.dict(), compression=compression)

    async def trim_image(self, item: ImageSource, norm_thresh: float, initial_thresh: float) -> str:
        """
        Trim the background from a single image, provided as either a base64 ``image`` or a ``url``.

        The image is used at full resolution (never the inference resize), so trimming operates on
        the original pixels.

        Parameters
        ----------
        item : ImageSource
            request item exposing ``id`` and either ``image`` (base64) or ``url``
        norm_thresh : float
            norm threshold to consider vectors the same
        initial_thresh : float
            identification thresh

        Returns
        -------
        result : str
            trimmed image as a urlsafe-base64 PNG string
        """
        # A bad/undecodable image (or failed download) raises Base64ValidationError (422), handled
        # by the exception-handling middleware. The image keeps its native mode (alpha/palette
        # preserved) so trimming sees the exact original pixels.
        image_array = await self.preprocessor.load_original_image(item)
        trimmed_image = trim_image_with_thresh(img=image_array, norm_thresh=norm_thresh,
                                               initial_thresh=initial_thresh)
        return pil_to_b64(image=Image.fromarray(trimmed_image))

start() async

Start the long-lived dependencies (message broker connection).

Source code in src/services.py

async def start(self) -> None:
    """Start the long-lived dependencies (message broker connection)."""
    await self.message_broker_client.start()

stop() async

Stop the dependencies and close the module-level aiohttp sessions (TF Serving + image downloader). Note this makes the service one-shot per process: a second start after a stop would reuse the closed sessions.

Source code in src/services.py

async def stop(self) -> None:
    """
    Stop the dependencies and close the module-level aiohttp sessions (TF Serving + image
    downloader). Note this makes the service one-shot per process: a second start after a
    stop would reuse the closed sessions.
    """
    await self.message_broker_client.stop()
    await session.close()
    await download_session.close()

health_check() async

Checks whether all dependencies are connected and ready to work

Source code in src/services.py

async def health_check(self) -> Tuple[bool, List[Dependence]]:
    """
    Checks whether all dependencies are connected and ready to work
    """
    dependencies: list[Dependence] = []
    for net in self.nets:
        connected, message = net.check_connection()
        dependencies.append(Dependence(connected=connected, message=message, name=net.name,
                                       version=getattr(net, "version", None)))
    connected, message = await self.message_broker_client.async_check_connection()
    dependencies.append(Dependence(connected=connected, message=message, name=self.message_broker_client.name,
                                   version=getattr(self.message_broker_client, "version", None)))
    healthy = all(dependence.connected for dependence in dependencies)

    return healthy, dependencies

info()

Return information on each network by calling corresponding methods

Source code in src/services.py

def info(self) -> List[dict]:
    """
    Return information on each network by calling corresponding methods
    """

    return [net.info() for net in self.nets]

generate_pred_dict(network, preprocessed, ids_as_keys=False, **kwargs) async

Make predictions with a network for multiple preprocessed images and return dictionaries with ids and predictions.

Parameters:

Name	Type	Description	Default
network	Predictor	network to make predictions with	required
preprocessed	List[PreprocessedImage]	preprocessed images (from :meth:Services.prepare_inputs). Successful entries carry a resized variant per model input size; failed entries become per-image 422 responses	required
ids_as_keys	bool	whether to use ids as keys in the returned dict or include them as a value of "id" field	False
**kwargs		additional arguments for .predict method of the network	{}

Returns:

Name	Type	Description
out	List of dict, dict	predictions per image with id and status

Source code in src/services.py

async def generate_pred_dict(
        self, network: Predictor,
        preprocessed: List[PreprocessedImage],
        ids_as_keys: bool = False,
        **kwargs
) -> Union[List[dict], dict]:
    """
    Make predictions with a network for multiple preprocessed images and return dictionaries with
    ids and predictions.

    Parameters
    ----------
    network : api.predictors.Predictor
        network to make predictions with
    preprocessed : List[PreprocessedImage]
        preprocessed images (from :meth:`Services.prepare_inputs`). Successful entries carry
        a resized variant per model input size; failed entries become per-image 422 responses
    ids_as_keys : bool
        whether to use ids as keys in the returned dict or include them as a value of "id" field
    **kwargs :
        additional arguments for .predict method of the network

    Returns
    -------
    out : List of dict, dict
        predictions per image with id and status
    """
    valid = [p for p in preprocessed if p.ok]
    invalid = [p for p in preprocessed if not p.ok]
    if not valid:
        # Whole batch unusable: mirror the legacy all-invalid-b64 behaviour (422 with per-id
        # detail), enriched with the specific per-image failure reasons.
        raise Base64ValidationError(
            detail=[{"invalid_ids": [p.id for p in invalid],
                     "errors": [{"id": p.id, "kind": p.error.kind.value,
                                 "message": p.error.message} for p in invalid]}],
            msg="Could not process any of the input images",
            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY)

    valid_instances = [p.image_for(network.input_shape) for p in valid]

    valid_responses = await network.predict(valid_instances, **kwargs)
    invalid_responses = [PredictorResponse(status_code=p.error.status_code,
                                           message=p.error.message,
                                           scores=[]) for p in invalid]

    raw_responses = valid_responses + invalid_responses
    ids = [p.id for p in valid] + [p.id for p in invalid]
    response = [item.dict() for item in raw_responses]  # TODO make use of the PredictorResponse
    if ids_as_keys:
        return {id_: {"results": {network.name: response[i]}}
                for i, id_ in enumerate(ids)}

    results_per_image = [{"results": {network.name: pred}}
                         for pred in response]
    predictions = add_ids_to_dicts(ids=ids, dicts=results_per_image)

    return self.postprocess_prediction_dicts(predictions)

prepare_inputs(images, networks) async

Preprocess request images once for every network that will consume them.

Parameters:

Name	Type	Description	Default
images	List[ImageSource]	request items, each carrying either a base64 image or a url	required
networks	List[Predictor]	the networks the batch will be fed to; one resized variant is produced per distinct input_shape, all from a single download + decode	required

Returns:

Name	Type	Description
out	List[PreprocessedImage]	one entry per input item, in the same order

Source code in src/services.py

async def prepare_inputs(
        self,
        images: List[ImageSource],
        networks: List[Predictor]
) -> List[PreprocessedImage]:
    """
    Preprocess request images once for every network that will consume them.

    Parameters
    ----------
    images : List[ImageSource]
        request items, each carrying either a base64 ``image`` or a ``url``
    networks : List[Predictor]
        the networks the batch will be fed to; one resized variant is produced per distinct
        ``input_shape``, all from a single download + decode

    Returns
    -------
    out : List[PreprocessedImage]
        one entry per input item, in the same order
    """
    shapes = {net.input_shape for net in networks}
    return await self.preprocessor.preprocess(images, shapes=shapes)

predict_tags(images, custom_thresholds) async

Call the Tagger network with given request images and get corresponding predictions. For more information on parameters, refer to Services.generate_pred_dict()

Source code in src/services.py

@timing
async def predict_tags(
        self,
        images: List[ImageSource],
        custom_thresholds: Optional[Union[float, list, dict]]
) -> List[dict]:
    """
    Call the Tagger network with given request images and get corresponding predictions. For
    more information on parameters, refer to Services.generate_pred_dict()
    """
    preprocessed = await self.prepare_inputs(images, networks=[self.tagger])
    return await self.generate_pred_dict(network=self.tagger,
                                         preprocessed=preprocessed,
                                         custom_thresholds=custom_thresholds)

predict_exterior_styles(images) async

Call the ExteriorStyles network with given request images and get corresponding predictions. For more information on parameters, refer to Services.generate_pred_dict()

Source code in src/services.py

@timing
async def predict_exterior_styles(
        self,
        images: List[ImageSource]
) -> List[dict]:
    """
    Call the ExteriorStyles network with given request images and get corresponding
    predictions. For more information on parameters, refer to Services.generate_pred_dict()
    """
    preprocessed = await self.prepare_inputs(images, networks=[self.exterior_styles])
    return await self.generate_pred_dict(network=self.exterior_styles,
                                         preprocessed=preprocessed)

encode(images) async

Encode given request images with the encoder network. For more information on parameters, refer to Services.generate_pred_dict()

Source code in src/services.py

@timing
async def encode(
        self,
        images: List[ImageSource],
) -> List[dict]:
    """
    Encode given request images with the encoder network. For more information on parameters,
    refer to Services.generate_pred_dict()
    """
    preprocessed = await self.prepare_inputs(images, networks=[self.encoder])
    return await self.generate_pred_dict(network=self.encoder,
                                         preprocessed=preprocessed)

classify_dht(preprocessed, custom_thresholds) async

Call Tagger network, then call the Exterior Styles for the samples with "facade" predicted tag. Return all tags in a single dict for each sample for each network.

The facade subset reuses the SAME :class:PreprocessedImage objects, so no image is downloaded, decoded or resized twice.

Source code in src/services.py

@timing
async def classify_dht(
        self,
        preprocessed: List[PreprocessedImage],
        custom_thresholds: Optional[Union[float, list, dict]]
) -> Tuple:
    """
    Call Tagger network, then call the Exterior Styles for the samples with "facade"
    predicted tag. Return all tags in a single dict for each sample for each network.

    The facade subset reuses the SAME :class:`PreprocessedImage` objects, so no image is
    downloaded, decoded or resized twice.
    """
    dht_preds = await self.generate_pred_dict(network=self.tagger,
                                              preprocessed=preprocessed,
                                              custom_thresholds=custom_thresholds,
                                              ids_as_keys=True)
    facade_ids = {k for k, v in dht_preds.items()
                  if "facade" in v["results"][self.tagger.name]["scores"]}
    facade_preprocessed = [p for p in preprocessed if p.id in facade_ids]
    style_preds = None
    if facade_preprocessed:
        style_preds = await self.generate_pred_dict(network=self.exterior_styles,
                                                    preprocessed=facade_preprocessed,
                                                    filtered=False,
                                                    ids_as_keys=True)
        for id_ in facade_ids:
            dht_preds[id_]["results"][self.exterior_styles.name] = style_preds[id_]["results"][
                self.exterior_styles.name].copy()
            dht_preds[id_]["results"][self.exterior_styles.name]["scores"] = argmax_dict(
                style_preds[id_]["results"][self.exterior_styles.name]["scores"])

    return dht_preds, style_preds

process_house_images(images, custom_thresholds, publish_embeddings=False, core_listing_id=None, embeddings_compression=CompressionMethod.GZIP) async

Process all images of a single house via Image Processing Networks

The images are preprocessed ONCE for every network involved (tagger, exterior styles and — when publishing embeddings — the encoder), so each photo is downloaded, decoded and resized a single time and shared across all passes.

Parameters:

Name	Type	Description	Default
images	List[ImageSource]	request house images, each carrying either a base64 image or a url	required
custom_thresholds	(float, list, dict)	per-class thresholds for the Tagger	required
publish_embeddings	bool	whether to also encode the images and publish their embeddings to the message broker	False
core_listing_id	Any	identifier of the house (used when publishing embeddings)	None
embeddings_compression	CompressionMethod	compression method to use when publishing embeddings	GZIP

Returns:

Name	Type	Description
out	dict	predictions for given house. Contains tagger's tags per image, information on present features, averaged scores for exterior styles and (if requested) publishing results

Source code in src/services.py

@timing
async def process_house_images(
        self,
        images: List[ImageSource],
        custom_thresholds: Optional[Union[float, list, dict]],
        publish_embeddings: bool = False,
        core_listing_id: Any = None,
        embeddings_compression: CompressionMethod = CompressionMethod.GZIP
) -> Dict:
    """
    Process all images of a single house via Image Processing Networks

    The images are preprocessed ONCE for every network involved (tagger, exterior styles and —
    when publishing embeddings — the encoder), so each photo is downloaded, decoded and
    resized a single time and shared across all passes.

    Parameters
    ----------
    images : List[ImageSource]
        request house images, each carrying either a base64 ``image`` or a ``url``
    custom_thresholds : float, list, dict, optional
        per-class thresholds for the Tagger
    publish_embeddings : bool
        whether to also encode the images and publish their embeddings to the message broker
    core_listing_id : Any
        identifier of the house (used when publishing embeddings)
    embeddings_compression : CompressionMethod
        compression method to use when publishing embeddings

    Returns
    -------
    out : dict
        predictions for given house. Contains tagger's tags per image, information on present
        features, averaged scores for exterior styles and (if requested) publishing results
    """
    networks = [self.tagger, self.exterior_styles]
    if publish_embeddings:
        networks.append(self.encoder)
    preprocessed = await self.prepare_inputs(images, networks=networks)

    dht_preds, style_preds = await self.classify_dht(preprocessed=preprocessed,
                                                     custom_thresholds=custom_thresholds)
    per_image = add_ids_to_dicts(ids=dht_preds.keys(), dicts=list(dht_preds.values()))
    avg_exterior_style = self.get_average_style(style_preds)
    house_status = any(d["results"][self.tagger.name]["status"] for d in dht_preds.values())
    present_features = [tag for im in dht_preds.values()
                        for tag in im["results"][self.tagger.name]["scores"]]
    unique_features = sorted(set(present_features))
    tagged = list(filter(lambda x: x["results"][self.tagger.name]["status"], per_image))
    house_info = {"exterior_styles": avg_exterior_style,
                  "prob_ranking": self.rank_images_by_class(tagged, classes=unique_features)}

    publish_results = None
    if publish_embeddings:
        publish_results = await self.publish_embeddings(core_listing_id=core_listing_id,
                                                        preprocessed=preprocessed,
                                                        compression=embeddings_compression)

    predictions = {"results_per_image": self.postprocess_prediction_dicts(per_image),
                   "house_info": house_info,
                   "status": house_status,
                   "publish_results": publish_results}
    return predictions

get_average_style(style_predictions_per_image)

Calculate average style of the house given predictions of the ExteriorStyle Classifier

Parameters:

Name	Type	Description	Default
style_predictions_per_image	dict	per image predictions for exterior styles	required

Returns:

Name	Type	Description
out	List of dict	average score for each possible class

Source code in src/services.py

def get_average_style(self, style_predictions_per_image: dict) -> List[dict]:
    """
    Calculate average style of the house given predictions of the ExteriorStyle Classifier

    Parameters
    ----------
    style_predictions_per_image : dict
        per image predictions for exterior styles

    Returns
    -------
    out : List of dict
        average score for each possible class
    """
    avg_exterior_style = []
    if not style_predictions_per_image:
        return avg_exterior_style

    successful = list(filter(
        lambda x: x["results"][self.exterior_styles.name]["status"],
        style_predictions_per_image.values()))
    num_facades = len(successful)
    if num_facades:
        avg_exterior_style = [{"name": k,
                               "probability": sum(
                                   i["results"][f"{self.exterior_styles.name}"]["scores"][k]
                                   for i in successful) / num_facades} for k in
                              self.exterior_styles.output_names]

    return avg_exterior_style

rank_images_by_class(tagger_dicts, classes)

Rank images by Tagger's confidence that given scene exists there

Parameters:

Name	Type	Description	Default
tagger_dicts	List[dict]	valid predictions per_image	required
classes	Iterable[str]	scenes to rank	required

Returns:

Name	Type	Description
out	List of Dict	Scenes and list of images containing them sorted by probability (highest first)

Source code in src/services.py

def rank_images_by_class(
        self,
        tagger_dicts: List[dict],
        classes: Iterable[str]
) -> List[dict]:
    """
    Rank images by Tagger's confidence that given scene exists there

    Parameters
    ----------
    tagger_dicts : List[dict]
        valid predictions per_image
    classes : Iterable[str]
        scenes to rank

    Returns
    -------
    out : List of Dict
         Scenes and list of images containing them sorted by probability (highest first)
    """

    if not classes:
        return []
    raw_scores = {cls: {d["id"]: d["results"][self.tagger.name]["scores"].get(cls, 0)
                        for d in tagger_dicts}
                  for cls in classes}
    sorted_tuples = {cls: sorted(raw_scores[cls].items(), key=lambda x: x[1],
                                 reverse=True) for cls in classes}
    ranking = [{"name": cls,
                "ordering": [k for k, v in sorted_tuples[cls] if v > 0]}
               for cls in classes]

    return ranking

postprocess_prediction_dicts(results_per_image) staticmethod

Postprocess network prediction dictionaries

Parameters:

Name	Type	Description	Default
results_per_image	List[dict]	prediction dictionaries containing scores in format {"feature" : probability}	required

Returns:

Name	Type	Description
out	List	dictionaries containing scores in format {"name" : feature, "probability" : probability}

Source code in src/services.py

@staticmethod
def postprocess_prediction_dicts(results_per_image: List[dict]) -> List[dict]:
    """
    Postprocess network prediction dictionaries

    Parameters
    ----------
    results_per_image: List
        prediction dictionaries containing scores in format {"feature" : probability}

    Returns
    -------
    out : List
        dictionaries containing scores in format {"name" : feature, "probability" : probability}
    """
    processed = results_per_image.copy()
    for result in processed:
        for network_results in result["results"].values():
            scores_per_class = network_results["scores"]
            if isinstance(scores_per_class, dict):  # in case of encoder & similar nets
                network_results["scores"] = [{"name": cls_name,
                                              "probability": probability}
                                             for cls_name, probability in
                                             scores_per_class.items()]

    return processed

prepare_embeddings(preprocessed) async

Run the input data through Encoder model, filter and preprocess the output for publishing

Parameters:

Name	Type	Description	Default
preprocessed	List[PreprocessedImage]	preprocessed images (from :meth:Services.prepare_inputs, with the encoder among the target networks); only successful ones are embedded	required

Returns:

Name	Type	Description
out	list[PhotoEmbedding]	Vector embeddings and their ids

Source code in src/services.py

@timing
async def prepare_embeddings(
        self,
        preprocessed: List[PreprocessedImage]
) -> list[PhotoEmbedding]:
    """
    Run the input data through Encoder model, filter and preprocess the output for publishing

    Parameters
    ----------
    preprocessed : List[PreprocessedImage]
        preprocessed images (from :meth:`Services.prepare_inputs`, with the encoder among the
        target networks); only successful ones are embedded

    Returns
    -------
    out : list[PhotoEmbedding]
        Vector embeddings and their ids
    """
    valid = [p for p in preprocessed if p.ok]
    photo_embeddings: list[PhotoEmbedding] = []
    if not valid:
        return photo_embeddings
    instances = [p.image_for(self.encoder.input_shape) for p in valid]
    encoder_output = await self.encoder.predict_batch(data=instances)

    # Filter out the successful ones
    for preprocessed_image, response in zip(valid, encoder_output):
        if not response.status:
            continue
        photo_embeddings.append(PhotoEmbedding(photo_id=preprocessed_image.id,
                                               embedding=response.scores))

    return photo_embeddings

publish_embeddings(core_listing_id, preprocessed, compression) async

Calculate embeddings and publish them to message broker

Parameters:

Name	Type	Description	Default
core_listing_id	Any	identifier of a house	required
preprocessed	List[PreprocessedImage]	preprocessed images (from :meth:Services.prepare_inputs, with the encoder among the target networks) to embed	required
compression	CompressionMethod	Compression method to use. Defaults to None	required

Returns:

Name	Type	Description
out	PublishResults	Results of the publishing

Source code in src/services.py

@timing
async def publish_embeddings(self,
                             core_listing_id: Any,
                             preprocessed: List[PreprocessedImage],
                             compression: CompressionMethod) -> PublishResults:
    """
    Calculate embeddings and publish them to message broker

    Parameters
    ----------
    core_listing_id : Any
        identifier of a house
    preprocessed : List[PreprocessedImage]
        preprocessed images (from :meth:`Services.prepare_inputs`, with the encoder among the
        target networks) to embed
    compression: CompressionMethod = CompressionMethod.UNCOMPRESSED
        Compression method to use. Defaults to None

    Returns
    -------
    out : PublishResults
        Results of the publishing
    """
    embeddings = await self.prepare_embeddings(preprocessed=preprocessed)
    if not embeddings:
        return PublishResults(published=False, error="No embeddings to publish. There might be problems with the "
                                                     "encoder models or the input images.")
    payload = ListingEmbeddings(core_listing_id=core_listing_id, embeddings=embeddings)

    return await self.message_broker_client.publish(payload=payload.dict(), compression=compression)

trim_image(item, norm_thresh, initial_thresh) async

Trim the background from a single image, provided as either a base64 image or a url.

The image is used at full resolution (never the inference resize), so trimming operates on the original pixels.

Parameters:

Name	Type	Description	Default
item	ImageSource	request item exposing id and either image (base64) or url	required
norm_thresh	float	norm threshold to consider vectors the same	required
initial_thresh	float	identification thresh	required

Returns:

Name	Type	Description
result	str	trimmed image as a urlsafe-base64 PNG string

Source code in src/services.py

async def trim_image(self, item: ImageSource, norm_thresh: float, initial_thresh: float) -> str:
    """
    Trim the background from a single image, provided as either a base64 ``image`` or a ``url``.

    The image is used at full resolution (never the inference resize), so trimming operates on
    the original pixels.

    Parameters
    ----------
    item : ImageSource
        request item exposing ``id`` and either ``image`` (base64) or ``url``
    norm_thresh : float
        norm threshold to consider vectors the same
    initial_thresh : float
        identification thresh

    Returns
    -------
    result : str
        trimmed image as a urlsafe-base64 PNG string
    """
    # A bad/undecodable image (or failed download) raises Base64ValidationError (422), handled
    # by the exception-handling middleware. The image keeps its native mode (alpha/palette
    # preserved) so trimming sees the exact original pixels.
    image_array = await self.preprocessor.load_original_image(item)
    trimmed_image = trim_image_with_thresh(img=image_array, norm_thresh=norm_thresh,
                                           initial_thresh=initial_thresh)
    return pil_to_b64(image=Image.fromarray(trimmed_image))

---

Middlewares for pre- and post- processing of requests/responses

LoggingMiddleware

Bases: BaseHTTPMiddleware

Basic logging middleware inherited from starlette.BaseHTTPMiddleware

Source code in src/app/middlewares.py

class LoggingMiddleware(BaseHTTPMiddleware):
    """
    Basic logging middleware inherited from starlette.BaseHTTPMiddleware
    """

    def __init__(self, app, logger):
        """
        Init object with src and logger

        Parameters
        ----------
        app : fast.FastAPI
            application object where middleware need to be added
        logger : logging.loger
            already configured logger for logging requests
        """
        super().__init__(app)
        self.logger = logger

    async def dispatch(self, request: Request, call_next):
        """
        Overriding BaseHTTPMiddleware.dispatch method to implement logging logic

        Parameters
        ----------
        request : starlette.middleware.base.Request
            current request
        call_next : starlette.middleware.base.RequestResponseEndpoint
            call function

        Returns
        -------
        streaming_response : starlette.middleware.base.StreamingResponse
            streaming response for the endpoint
        """
        guid = generate_guid()
        self.logger.info(f"rid={guid} start request path={request.url.path}")
        start_time = time()

        streaming_response = await call_next(request)
        status_code = streaming_response.status_code
        process_time = (time() - start_time) * 1000
        formatted_process_time = '{0:.2f}'.format(process_time)

        # collect errors and log also error messages
        if status_code != status.HTTP_200_OK:
            response_body = [
                section async for section in streaming_response.body_iterator
            ]
            streaming_response.body_iterator = iterate_in_threadpool(
                iter(response_body)
            )
            msg = response_body[0].decode()
            self.logger.info(
                f"rid={guid} completed_in={formatted_process_time}ms"
                f" status_code={status_code}"
                f" response = {response_body},"
                f" message = {msg}"
            )
        # for 200 OK requests logging only event
        else:
            self.logger.info(
                f"rid={guid} completed_in={formatted_process_time}ms status_code={status_code}"
            )

        return streaming_response

__init__(app, logger)

Init object with src and logger

Parameters:

Name	Type	Description	Default
app	FastAPI	application object where middleware need to be added	required
logger	loger	already configured logger for logging requests	required

Source code in src/app/middlewares.py

def __init__(self, app, logger):
    """
    Init object with src and logger

    Parameters
    ----------
    app : fast.FastAPI
        application object where middleware need to be added
    logger : logging.loger
        already configured logger for logging requests
    """
    super().__init__(app)
    self.logger = logger

dispatch(request, call_next) async

Overriding BaseHTTPMiddleware.dispatch method to implement logging logic

Parameters:

Name	Type	Description	Default
request	Request	current request	required
call_next	RequestResponseEndpoint	call function	required

Returns:

Name	Type	Description
streaming_response	StreamingResponse	streaming response for the endpoint

Source code in src/app/middlewares.py

async def dispatch(self, request: Request, call_next):
    """
    Overriding BaseHTTPMiddleware.dispatch method to implement logging logic

    Parameters
    ----------
    request : starlette.middleware.base.Request
        current request
    call_next : starlette.middleware.base.RequestResponseEndpoint
        call function

    Returns
    -------
    streaming_response : starlette.middleware.base.StreamingResponse
        streaming response for the endpoint
    """
    guid = generate_guid()
    self.logger.info(f"rid={guid} start request path={request.url.path}")
    start_time = time()

    streaming_response = await call_next(request)
    status_code = streaming_response.status_code
    process_time = (time() - start_time) * 1000
    formatted_process_time = '{0:.2f}'.format(process_time)

    # collect errors and log also error messages
    if status_code != status.HTTP_200_OK:
        response_body = [
            section async for section in streaming_response.body_iterator
        ]
        streaming_response.body_iterator = iterate_in_threadpool(
            iter(response_body)
        )
        msg = response_body[0].decode()
        self.logger.info(
            f"rid={guid} completed_in={formatted_process_time}ms"
            f" status_code={status_code}"
            f" response = {response_body},"
            f" message = {msg}"
        )
    # for 200 OK requests logging only event
    else:
        self.logger.info(
            f"rid={guid} completed_in={formatted_process_time}ms status_code={status_code}"
        )

    return streaming_response

ExceptionHandlerMiddleware

Bases: BaseHTTPMiddleware

A middleware to handle errors

Source code in src/app/middlewares.py

class ExceptionHandlerMiddleware(BaseHTTPMiddleware):
    """A middleware to handle errors"""

    async def dispatch(self, request: Request, call_next):
        """Try to process the request. If failed, return details about the exception"""

        try:
            return await call_next(request)
        except DetailedExceptionBase as e:
            info = self.extract_info(e)
            info["additional_information"] = e.detail
            return JSONResponse(status_code=e.status_code, content={"detail": [info]})
        except Exception as e:  # noqa: BLE001 - last-resort 500 for any unhandled error
            info = self.extract_info(e)
            return JSONResponse(status_code=500, content={"detail": [info]})

    @staticmethod
    def extract_info(error: BaseException):
        """Extract the type and the message of an error and return as a dict"""
        return {"type": type(error).__name__,
                "msg": str(error)}

dispatch(request, call_next) async

Try to process the request. If failed, return details about the exception

Source code in src/app/middlewares.py

async def dispatch(self, request: Request, call_next):
    """Try to process the request. If failed, return details about the exception"""

    try:
        return await call_next(request)
    except DetailedExceptionBase as e:
        info = self.extract_info(e)
        info["additional_information"] = e.detail
        return JSONResponse(status_code=e.status_code, content={"detail": [info]})
    except Exception as e:  # noqa: BLE001 - last-resort 500 for any unhandled error
        info = self.extract_info(e)
        return JSONResponse(status_code=500, content={"detail": [info]})

extract_info(error) staticmethod

Extract the type and the message of an error and return as a dict

Source code in src/app/middlewares.py

@staticmethod
def extract_info(error: BaseException):
    """Extract the type and the message of an error and return as a dict"""
    return {"type": type(error).__name__,
            "msg": str(error)}

validation_exception_handler(_request, exc) async

Handler for FastAPI request-validation errors (registered in main.py).

Mirrors FastAPI's default 422 body but truncates long echoed input values, so a failed validation of a base64-image payload does not dump megabytes back at the client. Registered as an exception handler (rather than rewriting 422 responses in middleware) so it only ever touches genuine validation errors and leaves response headers of other 422s intact.

Source code in src/app/middlewares.py

async def validation_exception_handler(_request: Request, exc: RequestValidationError) -> JSONResponse:
    """
    Handler for FastAPI request-validation errors (registered in ``main.py``).

    Mirrors FastAPI's default 422 body but truncates long echoed ``input`` values, so a failed
    validation of a base64-image payload does not dump megabytes back at the client. Registered
    as an exception handler (rather than rewriting 422 responses in middleware) so it only ever
    touches genuine validation errors and leaves response headers of other 422s intact.
    """
    errors = jsonable_encoder(exc.errors())
    for error in errors:
        if "input" in error:
            error["input"] = _truncate_error_input(error["input"])
    return JSONResponse(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
                        content={"detail": errors})