|
| 1 | +# Copyright 2026 TIER IV, Inc. |
| 2 | +# |
| 3 | +# Licensed under the Apache License, Version 2.0 (the "License"); |
| 4 | +# you may not use this file except in compliance with the License. |
| 5 | +# You may obtain a copy of the License at |
| 6 | +# |
| 7 | +# http://www.apache.org/licenses/LICENSE-2.0 |
| 8 | +# |
| 9 | +# Unless required by applicable law or agreed to in writing, software |
| 10 | +# distributed under the License is distributed on an "AS IS" BASIS, |
| 11 | +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| 12 | +# See the License for the specific language governing permissions and |
| 13 | +# limitations under the License. |
| 14 | + |
| 15 | +from __future__ import annotations |
| 16 | + |
| 17 | +from dataclasses import dataclass |
| 18 | +from typing import Sequence, Any, Mapping |
| 19 | + |
| 20 | +import polars as pl |
| 21 | +from pydantic import BaseModel, ConfigDict |
| 22 | + |
| 23 | +from autoware_ml.databases.schemas.base_schemas import ( |
| 24 | + BaseFieldSchema, |
| 25 | + DatasetTableColumn, |
| 26 | + DataModelInterface, |
| 27 | +) |
| 28 | + |
| 29 | + |
| 30 | +@dataclass(frozen=True) |
| 31 | +class CategoryMappingDatasetSchema(BaseFieldSchema): |
| 32 | + """ |
| 33 | + Dataclass to define polars schema for columns related to category mapping. |
| 34 | + """ |
| 35 | + |
| 36 | + CATEGORY_NAMES = DatasetTableColumn("category_names", pl.List(pl.String)) |
| 37 | + CATEGORY_INDICES = DatasetTableColumn("category_indices", pl.List(pl.Int32)) |
| 38 | + |
| 39 | + |
| 40 | +class CategoryMappingDataModel(BaseModel, DataModelInterface): |
| 41 | + """ |
| 42 | + Category mapping data model that can be shared by multiple datasets. |
| 43 | +
|
| 44 | + Attributes: |
| 45 | + category_name: Category name. |
| 46 | + category_index: Category index. |
| 47 | + """ |
| 48 | + |
| 49 | + model_config = ConfigDict(frozen=True, strict=True, arbitrary_types_allowed=True) |
| 50 | + |
| 51 | + category_names: Sequence[str] |
| 52 | + category_indices: Sequence[int] |
| 53 | + |
| 54 | + def model_post_init(self, __context: Any) -> None: |
| 55 | + """Validate that all attributes are of the same length.""" |
| 56 | + |
| 57 | + assert len(self.category_names) == len(self.category_indices), ( |
| 58 | + "All attributes must be of the same length" |
| 59 | + ) |
| 60 | + |
| 61 | + def to_dictionary(self) -> Mapping[str, Any]: |
| 62 | + """ |
| 63 | + Convert the category mapping data model to a dictionary. |
| 64 | +
|
| 65 | + Returns: |
| 66 | + Mapping[str, Any]: Dictionary representation of the category mapping data model. |
| 67 | + """ |
| 68 | + |
| 69 | + return self.model_dump() |
| 70 | + |
| 71 | + @classmethod |
| 72 | + def load_from_dictionary(cls, data_model: Mapping[str, Any]) -> CategoryMappingDataModel: |
| 73 | + """ |
| 74 | + Load the category mapping data model and decode it to the corresponding CategoryMappingDataModel |
| 75 | + from a dictionary, which is deserialized from a Polars dataframe. |
| 76 | +
|
| 77 | + Args: |
| 78 | + data_model: Dictionary representation of the category mapping data model, which is |
| 79 | + deserialized from a Polars dataframe. |
| 80 | + """ |
| 81 | + return cls( |
| 82 | + category_names=data_model[CategoryMappingDatasetSchema.CATEGORY_NAMES.name], |
| 83 | + category_indices=data_model[CategoryMappingDatasetSchema.CATEGORY_INDICES.name], |
| 84 | + ) |
0 commit comments