gisaf-backend/src/gisaf/models/category.py

from typing import Any, ClassVar

from sqlalchemy import String
from pydantic import computed_field, ConfigDict
from sqlmodel import Field, Relationship, SQLModel, JSON, TEXT, select

from gisaf.models.metadata import gisaf_survey
from gisaf.database import db_session, pandas_query

mapbox_type_mapping = {
    'Point': 'symbol',
    'Line': 'line',
    'Polygon': 'fill',
}

class BaseModel(SQLModel):
    @classmethod
    async def get_df(cls):
        async with db_session() as session:
            query = select(cls)
            return await session.run_sync(pandas_query, query)


class CategoryGroup(BaseModel, table=True):
    metadata = gisaf_survey
    __tablename__ = 'category_group'
    name: str | None = Field(sa_type=String(4), default=None, primary_key=True)
    major: str
    long_name: str
    categories: list['Category'] = Relationship(back_populates='category_group')

    class Admin:
        menu = 'Other'
        flask_admin_model_view = 'CategoryGroupModelView'


class CategoryModelType(BaseModel, table=True):
    metadata = gisaf_survey
    __tablename__ = 'category_model_type'
    name: str | None = Field(default=None, primary_key=True)

    class Admin:
        menu = 'Other'
        flask_admin_model_view = 'MyModelViewWithPrimaryKey'


class CategoryBase(BaseModel):
    model_config = ConfigDict(protected_namespaces=())  # type: ignore
    class Admin:
        menu = 'Other'
        flask_admin_model_view = 'CategoryModelView'

    name: str | None = Field(default=None, primary_key=True)
    domain: ClassVar[str] = 'V'
    description: str | None
    group: str = Field(min_length=4, max_length=4,
                       foreign_key="category_group.name", index=True)
    minor_group_1: str = Field(min_length=4, max_length=4, default='----')
    minor_group_2: str = Field(min_length=4, max_length=4, default='----')
    status: str = Field(min_length=1, max_length=1)
    custom: bool | None
    auto_import: bool = True
    gis_type: str = Field(max_length=50,
                          foreign_key='category_model_type.name',
                          default='Point')
    long_name: str | None = Field(max_length=50)
    style: str | None = Field(sa_type=TEXT)
    symbol: str | None = Field(max_length=1)
    mapbox_type_custom: str | None = Field(max_length=32)
    mapbox_paint: dict[str, Any] | None = Field(sa_type=JSON(none_as_null=True))
    mapbox_layout: dict[str, Any] | None = Field(sa_type=JSON(none_as_null=True))
    viewable_role: str | None
    extra: dict[str, Any] | None = Field(sa_type=JSON(none_as_null=True))

    @computed_field
    @property
    def layer_name(self) -> str:
        """
        ISO compliant layer name (see ISO 13567)
        :return: str
        """
        return '{self.domain}-{self.group:4s}-{self.minor_group_1:4s}-{self.minor_group_2:4s}-{self.status:1s}'.format(self=self)

    @computed_field
    @property
    def table_name(self) -> str:
        """
        Table name
        :return:
        """
        if self.minor_group_2 == '----':
            return '{self.domain}_{self.group:4s}_{self.minor_group_1:4s}'.format(self=self)
        else:
            return '{self.domain}_{self.group:4s}_{self.minor_group_1:4s}_{self.minor_group_2:4s}'.format(self=self)

    @computed_field
    @property
    def raw_survey_table_name(self) -> str:
        """
        Table name
        :return:
        """
        if self.minor_group_2 == '----':
            return 'RAW_{self.domain}_{self.group:4s}_{self.minor_group_1:4s}'.format(self=self)
        else:
            return 'RAW_{self.domain}_{self.group:4s}_{self.minor_group_1:4s}_{self.minor_group_2:4s}'.format(self=self)

    @computed_field
    @property
    def mapbox_type(self) -> str:
        return self.mapbox_type_custom or mapbox_type_mapping[self.gis_type]


class Category(CategoryBase, table=True):
    metadata = gisaf_survey
    name: str | None = Field(default=None, primary_key=True)
    category_group: CategoryGroup = Relationship(back_populates="categories")


class CategoryRead(CategoryBase):
    name: str