Skip to content

Application schemas

File for storing the application models

SingleImageModel

Bases: BaseModel

Schema for a single instance in the PredictRequest

Source code in src/app/schemas.py 
42
43
44
45
46
47
class SingleImageModel(BaseModel):
    """
    Schema for a single instance in the PredictRequest
    """
    id: int = Field(description="Identifier of the image", examples=[1])
    image: str = Field(description="Image as a string (url, base64 etc)", examples=[b64_image])

TrimImageRequest

Bases: SingleImageModel

Schema for trim_image endpoint

Source code in src/app/schemas.py 
50
51
52
53
54
55
56
57
58
59
class TrimImageRequest(SingleImageModel):
    """
    Schema for trim_image endpoint
    """
    norm_thresh: Optional[confloat(ge=0.0)] = Field(default=1.5,
                                                    description="Threshold of vectors norm to "
                                                                "consider vectors as a same")
    initial_thresh: Optional[confloat(ge=0.0)] = Field(default=0.001,
                                                       description="Threshold of vectors norm "
                                                                   "to identify trimming direction")

PredictBatchRequest

Bases: BaseModel

Batch of images with id and image field each

Source code in src/app/schemas.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
class PredictBatchRequest(BaseModel):
    """
    Batch of images with id and image field each
    """
    images: conlist(item_type=SingleImageModel, min_length=1) = \
        Field(description="List of images to infer the CNN on")
    core_listing_id: Optional[int] = Field(default=None, description="Identifier of the house the images belong to",
                                           examples=[1])
    thresholds: Optional[Union[confloat(ge=0, le=1), Dict[str, confloat(ge=0, le=1)]]] = Field(default=None,
                                                                                               description="Probability threshold to use when extracting tags via multilabel network. "
                                                                                                           "If not given, defaults from the config file will be used",
                                                                                               examples=[{
                                                                                                   "interior": 0.5}])

    @field_validator("images")
    def ids_are_unique(cls, v):
        """Check if all ids are unique in a list of elements"""
        ids = [item.id for item in v]
        validate_unique_ids(ids=ids, field_name='id')
        return v

ids_are_unique(v)

Check if all ids are unique in a list of elements

Source code in src/app/schemas.py 
76
77
78
79
80
81
@field_validator("images")
def ids_are_unique(cls, v):
    """Check if all ids are unique in a list of elements"""
    ids = [item.id for item in v]
    validate_unique_ids(ids=ids, field_name='id')
    return v

PredictHomeRequest

Bases: PredictBatchRequest

Extended model for the endpoint that processes all house images

Source code in src/app/schemas.py 
84
85
86
87
88
89
90
91
class PredictHomeRequest(PredictBatchRequest):
    """Extended model for the endpoint that processes all house images"""
    publish_embeddings: bool = Field(default=False,
                                     description="Whether to send vector embeddings to the message broker queue")
    embeddings_compression: CompressionMethod = Field(default=CompressionMethod.GZIP,
                                                      description="Compression method to use when sending "
                                                                  "embeddings to the message broker queue",
                                                      examples=[CompressionMethod.GZIP])

Prediction

Bases: BaseModel

Model for prediction of a single image

Source code in src/app/schemas.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
class Prediction(BaseModel):
    """
    Model for prediction of a single image
    """
    scores: Union[List[float], Dict[str, float], List[dict]] = \
        Field(description="Dictionary with present labels and their probabilities",
              examples=[[{"name": "class1", "probability": 0.75},
                         {"name": "class4", "probability": 0.6}]]
              )
    message: Optional[str] = Field(default=None, description="Error message (if any) from the final network",
                                   examples=[''])
    status: bool = Field(description="Boolean showing whether the image is successfully processed", examples=[True])

TaggedImageModel

Bases: BaseModel

Model for a single tagged image

Source code in src/app/schemas.py 
108
109
110
111
112
113
114
115
116
117
118
class TaggedImageModel(BaseModel):
    """
    Model for a single tagged image
    """
    id: int = Field(description="Identifier of the image", examples=[1])
    results: Dict[str, Prediction] = Field(description="Results per network",
                                           examples=[{"net1": [{"name": "class1", "probability": 0.55}],
                                                      "net2": [0.25,
                                                               0.75,
                                                               0.1,
                                                               0.4]}])

PredictBatchResponse

Bases: BaseModel

Response for prediction request on a bunch of (unrelated) images

Source code in src/app/schemas.py 
121
122
123
124
125
126
class PredictBatchResponse(BaseModel):
    """
    Response for prediction request on a bunch of (unrelated) images
    """
    predictions: List[TaggedImageModel] = Field(description="List of prediction dicts for each "
                                                            "passed image")

HouseFeatures

Bases: BaseModel

Model for extracted home features

Source code in src/app/schemas.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
class HouseFeatures(BaseModel):
    """
    Model for extracted home features
    """

    exterior_styles: List[dict] = Field(description="Dictionary of Exterior Styles as keys and their averaged "
                                                    "probability over all facade images as values",
                                        examples=[[{"name": "style1", "probability": 0.1},
                                                   {"name": "style2", "probability": 0.25},
                                                   {"name": "style3", "probability": 0.65}]])
    prob_ranking: List[dict] = Field(description="Dictionary of present classes and the array of ids of images "
                                                 "containing them, sorted in descending order by probability",
                                     examples=[[{"name": "feature1", "ordering": [0, 1, 2, 3]},
                                                {"name": "feature5", "ordering": [2, 4, 3]}]]
                                     )

PredictHomeResponse

Bases: BaseModel

A model for house info prediction endpoint response

Source code in src/app/schemas.py 
146
147
148
149
150
151
152
153
154
155
156
class PredictHomeResponse(BaseModel):
    """
    A model for house info prediction endpoint response
    """
    core_listing_id: Optional[int] = Field(default=None, description="Identifier of a house", examples=[1])
    status: bool = Field(description="Whether at least one image is processed successfully", examples=[True])
    house_info: HouseFeatures = Field(description="House-specific information extracted from the "
                                                  "passed images")
    results_per_image: List[TaggedImageModel] = Field(description="List of predictions for "
                                                                  "successfully processed images")
    publish_results: PublishResults | None = Field(None, description="Information about the published message if any")

Dependence

Bases: BaseModel

Dependencies model

Source code in src/app/schemas.py 
159
160
161
162
163
164
165
166
167
168
169
170
class Dependence(BaseModel):
    """
    Dependencies model
    """

    name: str = Field(description="Name of the dependency", examples=["some_dependency"])
    version: Optional[int] = Field(default=None, description="Version of the dependency", examples=[1])
    connected: bool = Field(description="Boolean representing whether API is succesfully "
                                        "connected to said dependency", examples=[True])
    status: Optional[str] = Field(default=None, description="Current status of the dependence returned by the "
                                                            "final endpoint", examples=[True])
    message: Optional[str] = Field(default="success", description="Message based on current status", examples=[''])

HealthCheckResponse

Bases: BaseModel

HealthCheck endpoint response model

Source code in src/app/schemas.py 
173
174
175
176
177
178
179
class HealthCheckResponse(BaseModel):
    """
    HealthCheck endpoint response model
    """

    healthy: bool = Field(description="Boolean showing whether the service is healthy", examples=[True])
    dependencies: List[Dependence] = Field(description="List of dicts for each dependency")

InfoResponse

Bases: BaseModel

Response for info endpoint returning name/url/output info on each network

Source code in src/app/schemas.py 
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
class InfoResponse(BaseModel):
    """
    Response for info endpoint returning name/url/output info on each network
    """
    networks: List[dict]
    model_config = ConfigDict(json_schema_extra={
        'example': {
            "networks": [
                {"name": "network1",
                 "network": {
                     "predict_url": "localhost:8501/v1/models/network1/versions/2:predict",
                     "version": 2, "num_outputs": 101,
                     "raw_output_names": ["class1", "class2", "class3"],
                     "is_multilabel": True},
                 "outputs": {"num_classes": 3,
                             "all_classes": ["class1", "class2", "class3"],
                             "default_thresholds": {"class1": 0.5,
                                                    "class2": 0.65,
                                                    "class3": 0.7}}},
                {"name": "network2",
                 "network": {
                     "predict_url": "localhost:8501/v1/models/network2/versions/2:predict",
                     "version": 2, "num_outputs": 2,
                     "raw_output_names": ["class1", "class2"],
                     "outputs": {"num_classes": 2,
                                 "all_classes": ["class1", "class2"]}}},
                {"name": "network3", "network": {
                    "predict_url": "localhost:8501/v1/models/network3/versions/1:predict",
                    "version": 1}}]}})

validate_unique_ids(ids, field_name='id')

Validate whether provided elements are unique

Parameters:

Name Type Description Default
ids Collection

Elements to validate

 required
field_name str = 'id'

Name of the field. Used in the error message

 'id'

Returns:

Type Description
None

Raises:

Type Description
ValueError : if elements are non-unique
Source code in src/app/schemas.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
def validate_unique_ids(ids: Collection, field_name: str = 'id') -> None:
    """
    Validate whether provided elements are unique

    Parameters
    ----------
    ids : Collection
        Elements to validate
    field_name : str = 'id'
        Name of the field. Used in the error message

    Returns
    -------
    None

    Raises
    ------
    ValueError : if elements are non-unique
    """
    if not len(set(ids)) == len(ids):
        # Validation failed, build an error message
        counts = Counter(ids)
        non_unique = [id_ for id_, count in counts.items() if count > 1]

        raise ValueError(f"Please provide unique {field_name} for each element. Non unique ids: {non_unique}")