diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..8d4185b --- /dev/null +++ b/.env.example @@ -0,0 +1,9 @@ +LOGS_FILE="object_effects" +# Explicit API roots (e.g. https://host/urban_api) are preserved; bare origins use /api. +URBAN_API="https://urban-api.testing" +MCP_URBAN_API="https://urban-api.testing" +PROMETHEUS_PORT=9464 +SERVICE_AUTH_SERVER_URL=https://keycloak.example.com +SERVICE_AUTH_REALM=IDU +SERVICE_AUTH_CLIENT_ID=object-effects +SERVICE_AUTH_CLIENT_SECRET=change-me diff --git a/.github/workflows/build_and_deploy.yml b/.github/workflows/build_and_deploy.yml new file mode 100644 index 0000000..6c9da6a --- /dev/null +++ b/.github/workflows/build_and_deploy.yml @@ -0,0 +1,61 @@ +name: build_and_deploy +on: workflow_dispatch +env: + IMAGE_NAME: ${{secrets.REGISTRY}}/object_effects + CONTAINER_NAME: object_effects + +jobs: + build: + runs-on: 65_runner + outputs: + now: ${{steps.date.outputs.NOW}} + steps: + - name: Set current date as env variable + id: date + run: echo "NOW=$(date +'%Y-%m-%dT%H-%M-%S')" >> $GITHUB_OUTPUT + - name: checkout + uses: actions/checkout@v4 + - name: copy_env + env: + ENV_PATH: ${{secrets.ENV_PATH}} + run: cp "$ENV_PATH"/.env.development ./ + - name: build + env: + NOW: ${{steps.date.outputs.now}} + run: docker build -t "$IMAGE_NAME":"$NOW" . + - name: push_to_registry + env: + NOW: ${{steps.date.outputs.now}} + run: docker push "$IMAGE_NAME":"$NOW" + stop_container: + runs-on: 65_runner + needs: build + steps: + - name: stop_container + run: docker rm -f "$CONTAINER_NAME" + run_container: + runs-on: 65_runner + needs: [build, stop_container] + # Required so the SERVICE_AUTH_* vars/secrets of the "production" environment resolve: + # without it they come out empty and docker-compose.actions.yml aborts on ${VAR:?...}. + environment: production + env: + NOW: ${{needs.build.outputs.now}} + steps: + - name: set env + run: echo "IMAGE=$IMAGE_NAME:$NOW" >> $GITHUB_ENV + - name: checkout + uses: actions/checkout@v4 + - name: copy_env + env: + ENV_PATH: ${{secrets.ENV_PATH}} + run: cp "$ENV_PATH"/.env.development ./ + - name: run + env: + SERVICE_AUTH_SERVER_URL: ${{ vars.SERVICE_AUTH_SERVER_URL }} + SERVICE_AUTH_REALM: ${{ vars.SERVICE_AUTH_REALM }} + SERVICE_AUTH_CLIENT_ID: ${{ vars.SERVICE_AUTH_CLIENT_ID }} + SERVICE_AUTH_CLIENT_SECRET: ${{ secrets.SERVICE_AUTH_CLIENT_SECRET }} + URBAN_API: ${{ vars.URBAN_API }} +# run: docker run -d --name "$CONTAINER_NAME" --env-file ./.env.development -p 8210:8000 "$IMAGE_NAME":"$NOW" + run: docker compose -f docker-compose.actions.yml up -d diff --git a/.github/workflows/ci-dev.yml b/.github/workflows/ci-dev.yml new file mode 100644 index 0000000..96472c4 --- /dev/null +++ b/.github/workflows/ci-dev.yml @@ -0,0 +1,27 @@ +name: Kubernetes dev release + +on: + push: + branches: + - dev + +permissions: + contents: read + +jobs: + release: + uses: IDUclub/urban-assistant-deploy/.github/workflows/reusable-application-release.yaml@main + with: + service: object-effects + test-command: | + python -m pip install --disable-pip-version-check uv==0.12.10 + uv python install 3.11 + uv venv --python 3.11 + uv pip install --python .venv/bin/python \ + --requirement requirements.txt \ + --requirement requirements-auth.txt \ + pytest \ + pytest-asyncio + cp .env.example .env.test + APP_ENV=test .venv/bin/python -m pytest --verbose tests + secrets: inherit diff --git a/.gitignore b/.gitignore index 4b74b34..46a37a2 100644 --- a/.gitignore +++ b/.gitignore @@ -120,7 +120,8 @@ celerybeat.pid *.sage.py # Environments -.env* +.env.development +.env.production .venv env/ venv/ @@ -160,4 +161,8 @@ cython_debug/ .idea/ # Notebooks -*.ipynb \ No newline at end of file +*.ipynb + +# Agents +CLAUDE.md +AGENTS.md diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml new file mode 100644 index 0000000..54e2bba --- /dev/null +++ b/.pre-commit-config.yaml @@ -0,0 +1,13 @@ +repos: + - repo: https://github.com/psf/black + rev: 26.3.1 + hooks: + - id: black + language_version: python3.11 + + - repo: https://github.com/pycqa/isort + rev: 8.0.1 + hooks: + - id: isort + name: isort (python) + args: ["--profile", "black"] \ No newline at end of file diff --git a/Dockerfile b/Dockerfile index 27d3806..1a574f0 100644 --- a/Dockerfile +++ b/Dockerfile @@ -13,13 +13,15 @@ ENV PYTHONUNBUFFERED=1 # Enables env file ENV APP_ENV=development - +# add pyppi mirror to config +COPY pip.conf /etc/xdg/pip/pip.conf # Install pip requirements COPY requirements.txt . -RUN python -m pip install -r requirements.txt +COPY requirements-auth.txt . +RUN python -m pip install -r requirements.txt -r requirements-auth.txt WORKDIR /app COPY . /app # During debugging, this entry point will be overridden. For more information, please refer to https://aka.ms/vscode-docker-python-debug -CMD ["gunicorn", "--bind", "0.0.0.0:80", "-k", "uvicorn.workers.UvicornWorker", "--workers", "4", "app.main:app"] \ No newline at end of file +CMD ["gunicorn", "--bind", "0.0.0.0:80", "-k", "uvicorn.workers.UvicornWorker", "--workers", "1", "app.main:app"] diff --git a/README.md b/README.md index ab0a4ee..8dd256b 100644 --- a/README.md +++ b/README.md @@ -1,2 +1,6 @@ # ObjectEffectsAPI Repository for evaluation effects by ObjectNat library + +### Logs + +`GET /logs` downloads the application log file without authorization. diff --git a/app/__dev_runner__.py b/app/__dev_runner__.py new file mode 100644 index 0000000..d85f2ff --- /dev/null +++ b/app/__dev_runner__.py @@ -0,0 +1,6 @@ +import uvicorn + +from app.main import app + +if __name__ == "__main__": + uvicorn.run(app, host="127.0.0.1", port=8020) diff --git a/app/__version__.py b/app/__version__.py new file mode 100644 index 0000000..e3ed1f4 --- /dev/null +++ b/app/__version__.py @@ -0,0 +1 @@ +APP_VERSION = "0.4.0" diff --git a/app/common/api_handler/api_handler.py b/app/common/api_handler/api_handler.py index 6f430d1..2ebcb6d 100644 --- a/app/common/api_handler/api_handler.py +++ b/app/common/api_handler/api_handler.py @@ -1,13 +1,16 @@ import aiohttp +from idu_service_auth import KeycloakTokenClient +from app.common.api_handler.urban_api_url import normalize_urban_api_url from app.common.exceptions.http_exception_wrapper import http_exception class APIHandler: def __init__( - self, - base_url: str, + self, + base_url: str, + service_auth: KeycloakTokenClient, ) -> None: """Initialisation function @@ -17,11 +20,18 @@ def __init__( None """ - self.base_url = base_url + self.base_url = normalize_urban_api_url(base_url) + self.service_auth = service_auth + + async def _service_headers(self, headers: dict | None) -> dict[str, str]: + """Preserve request context while always replacing caller auth with M2M.""" + outgoing = dict(headers or {}) + outgoing.update(await self.service_auth.get_authorization_headers()) + return outgoing @staticmethod async def _check_response_status( - response: aiohttp.ClientResponse + response: aiohttp.ClientResponse, ) -> list | dict | None: """Function handles response @@ -38,8 +48,15 @@ async def _check_response_status( elif response.status == 500: if response.content_type == "application/json": response_info = await response.json() - if "reset by peer" in await response_info["error"]: + if "reset by peer" in response_info: return None + else: + raise http_exception( + 500, + "Couldn't get data from API", + _input=repr(response.url), + _detail=response_info, + ) else: response_info = await response.text() raise http_exception( @@ -57,11 +74,11 @@ async def _check_response_status( ) async def get( - self, - endpoint_url: str, - headers: dict | None = None, - params: dict | None = None, - session: aiohttp.ClientSession | None = None, + self, + endpoint_url: str, + headers: dict | None = None, + params: dict | None = None, + session: aiohttp.ClientSession | None = None, ) -> dict | list: """Function to get data from api @@ -82,14 +99,20 @@ async def get( params=params, session=session, ) - url = self.base_url + endpoint_url - async with session.get( - url=url, - headers=headers, - params=params - ) as response: + headers = await self._service_headers(headers) + endpoint = endpoint_url.lstrip("/").removeprefix("api/") + url = f"{self.base_url}/{endpoint}" + async with session.get(url=url, headers=headers, params=params) as response: result = await self._check_response_status(response) + if isinstance(result, list): + return result + elif isinstance(result, dict): + return result if not result: + if isinstance(result, list): + return result + elif isinstance(result, dict): + return result return await self.get( endpoint_url=endpoint_url, headers=headers, @@ -99,13 +122,13 @@ async def get( return result async def post( - self, - endpoint_url: str, - headers: dict | None = None, - params: dict | None = None, - data: dict | None = None, - session: aiohttp.ClientSession | None = None, - ) -> dict | list: + self, + endpoint_url: str, + headers: dict | None = None, + params: dict | None = None, + data: dict | None = None, + session: aiohttp.ClientSession | None = None, + ) -> dict | list: """Function to post data from api Args: @@ -127,7 +150,9 @@ async def post( data=data, session=session, ) - url = self.base_url + endpoint_url + headers = await self._service_headers(headers) + endpoint = endpoint_url.lstrip("/").removeprefix("api/") + url = f"{self.base_url}/{endpoint}" async with session.post( url=url, headers=headers, @@ -136,7 +161,7 @@ async def post( ) as response: result = await self._check_response_status(response) if not result: - return await self.post( + return await self.post( endpoint_url=endpoint_url, headers=headers, params=params, @@ -145,12 +170,12 @@ async def post( return result async def put( - self, - endpoint_url: str, - headers: dict | None = None, - params: dict | None = None, - data: dict | None = None, - session: aiohttp.ClientSession | None = None, + self, + endpoint_url: str, + headers: dict | None = None, + params: dict | None = None, + data: dict | None = None, + session: aiohttp.ClientSession | None = None, ) -> dict | list: """Function to post data from api @@ -173,16 +198,18 @@ async def put( data=data, session=session, ) - url = self.base_url + endpoint_url + headers = await self._service_headers(headers) + endpoint = endpoint_url.lstrip("/").removeprefix("api/") + url = f"{self.base_url}/{endpoint}" async with session.put( - url=url, - headers=headers, - params=params, - data=data, + url=url, + headers=headers, + params=params, + data=data, ) as response: result = await self._check_response_status(response) if not result: - return await self.put( + return await self.put( endpoint_url=endpoint_url, headers=headers, params=params, @@ -191,12 +218,12 @@ async def put( return result async def delete( - self, - endpoint_url: str, - headers: dict | None = None, - params: dict | None = None, - data: dict | None = None, - session: aiohttp.ClientSession | None = None, + self, + endpoint_url: str, + headers: dict | None = None, + params: dict | None = None, + data: dict | None = None, + session: aiohttp.ClientSession | None = None, ) -> dict | list: """Function to post data from api @@ -219,16 +246,18 @@ async def delete( data=data, session=session, ) - url = self.base_url + endpoint_url + headers = await self._service_headers(headers) + endpoint = endpoint_url.lstrip("/").removeprefix("api/") + url = f"{self.base_url}/{endpoint}" async with session.delete( - url=url, - headers=headers, - params=params, - data=data, + url=url, + headers=headers, + params=params, + data=data, ) as response: result = await self._check_response_status(response) if not result: - return await self.delete( + return await self.delete( endpoint_url=endpoint_url, headers=headers, params=params, diff --git a/app/common/api_handler/urban_api_url.py b/app/common/api_handler/urban_api_url.py new file mode 100644 index 0000000..cda0900 --- /dev/null +++ b/app/common/api_handler/urban_api_url.py @@ -0,0 +1,19 @@ +"""Urban API roots for direct connections and load-balancer mounts.""" + +from urllib.parse import urlsplit, urlunsplit + + +def normalize_urban_api_url(base_url: str) -> str: + """Preserve explicit API roots; use /api only for an origin without a path.""" + url = urlsplit(base_url.strip()) + if ( + url.scheme not in {"http", "https"} + or not url.netloc + or url.query + or url.fragment + ): + raise ValueError( + "Urban API URL must be an HTTP(S) base URL without query or fragment" + ) + path = url.path.rstrip("/") or "/api" + return urlunsplit((url.scheme, url.netloc, path, "", "")) diff --git a/app/effects/dto/__init__.py b/app/common/auth/__init__.py similarity index 100% rename from app/effects/dto/__init__.py rename to app/common/auth/__init__.py diff --git a/app/common/auth/bearer.py b/app/common/auth/bearer.py new file mode 100644 index 0000000..788136f --- /dev/null +++ b/app/common/auth/bearer.py @@ -0,0 +1,14 @@ +from typing import Optional + +from fastapi import Depends +from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer + +http_bearer = HTTPBearer() + + +async def verify_bearer_token( + credentials: HTTPAuthorizationCredentials = Depends(http_bearer), +) -> str | None: + + token = credentials.credentials + return token if token not in ["''", '""'] or not token else None diff --git a/app/common/auth/service_auth.py b/app/common/auth/service_auth.py new file mode 100644 index 0000000..9ec342b --- /dev/null +++ b/app/common/auth/service_auth.py @@ -0,0 +1,98 @@ +from __future__ import annotations + +from fastapi import Header, HTTPException, Security, status +from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer +from fastmcp.exceptions import AuthorizationError, ToolError +from fastmcp.server.auth import AccessToken +from fastmcp.server.auth.providers.jwt import JWTVerifier +from fastmcp.server.dependencies import get_http_headers +from idu_service_auth import KeycloakTokenClient, KeycloakTokenConfig + +from app.common.config.config import Config + +USER_ID_HEADER = "X-User-Id" +SERVICE_ACCOUNT_PREFIX = "service-account-" +bearer_scheme = HTTPBearer(auto_error=True) + + +def build_service_auth(config: Config) -> KeycloakTokenClient: + return KeycloakTokenClient( + KeycloakTokenConfig( + auth_server_url=config.get("SERVICE_AUTH_SERVER_URL"), + realm=config.get("SERVICE_AUTH_REALM"), + client_id=config.get("SERVICE_AUTH_CLIENT_ID"), + client_secret=config.get("SERVICE_AUTH_CLIENT_SECRET"), + background_refresh=True, + ) + ) + + +def build_service_token_verifier(config: Config) -> "ServiceTokenVerifier": + return ServiceTokenVerifier(config) + + +async def get_current_user_id( + _credentials: HTTPAuthorizationCredentials = Security(bearer_scheme), + x_user_id: str | None = Header(default=None, alias=USER_ID_HEADER), +) -> str: + if not x_user_id or not x_user_id.strip(): + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail=f"{USER_ID_HEADER} header is required", + ) + return x_user_id.strip() + + +async def require_service_token( + credentials: HTTPAuthorizationCredentials = Security(bearer_scheme), +) -> None: + """Require a verified Keycloak service-account token.""" + + from app.dependencies import service_token_verifier + + try: + access_token = await service_token_verifier.verify_token( + credentials.credentials + ) + except Exception as exc: + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Invalid service token", + ) from exc + if access_token is None: + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Invalid service token", + ) + + +def get_mcp_user_id() -> str: + user_id = get_http_headers(include_all=True).get("x-user-id", "").strip() + if not user_id: + raise ToolError(f"{USER_ID_HEADER} header is required") + return user_id + + +class ServiceTokenVerifier(JWTVerifier): + """Verify Keycloak JWTs and accept only client-credentials accounts.""" + + def __init__(self, config: Config) -> None: + server_url = config.get("SERVICE_AUTH_SERVER_URL").rstrip("/") + realm = config.get("SERVICE_AUTH_REALM") + issuer = f"{server_url}/realms/{realm}" + super().__init__( + jwks_uri=f"{issuer}/protocol/openid-connect/certs", + issuer=issuer, + algorithm="RS256", + ) + + async def verify_token(self, token: str) -> AccessToken | None: + access_token = await super().verify_token(token) + if access_token is None: + return None + username = access_token.claims.get("preferred_username", "") + if not isinstance(username, str) or not username.startswith( + SERVICE_ACCOUNT_PREFIX + ): + raise AuthorizationError("A service-account token is required") + return access_token diff --git a/app/common/config/__init__.py b/app/common/config/__init__.py new file mode 100644 index 0000000..cca5d9b --- /dev/null +++ b/app/common/config/__init__.py @@ -0,0 +1 @@ +from .config import Config diff --git a/app/common/config/config.py b/app/common/config/config.py new file mode 100644 index 0000000..271e678 --- /dev/null +++ b/app/common/config/config.py @@ -0,0 +1,63 @@ +import os +from pathlib import Path + + +class Config: + """ + Class for loading environment variables from .env.{APP_ENV} file + """ + + def __init__(self): + app_env = os.getenv("APP_ENV") + if not app_env: + raise ValueError("APP_ENV variable is not present") + env_file = Path().absolute() / f".env.{app_env}" + if not env_file.is_file(): + raise FileNotFoundError(f"Couldn't find file with .env.{app_env} name") + self._load_env_file(env_file) + + @staticmethod + def _load_env_file(env_file: Path) -> None: + """ + Function loads variables from env file, existing environment variables take precedence + Args: + env_file (Path): path to env file + """ + + for line in env_file.read_text(encoding="utf-8").splitlines(): + line = line.strip() + if not line or line.startswith("#") or "=" not in line: + continue + key, _, value = line.partition("=") + key = key.strip() + value = value.strip().strip("'\"") + if key and key not in os.environ: + os.environ[key] = value + + @staticmethod + def get(key: str) -> str: + """ + Function gets environment variable value + Args: + key (str): name of environment variable + Returns: + str: value of environment variable + Raises: + ValueError: if environment variable is not set + """ + + value = os.getenv(key) + if value: + return value + raise ValueError(f"No such env: {key}") + + @staticmethod + def set(key: str, value: str) -> None: + """ + Function sets value for environment variable + Args: + key (str): name of environment variable + value (str): new value for environment variable + """ + + os.environ[key] = value diff --git a/app/common/exceptions/http_exception_wrapper.py b/app/common/exceptions/http_exception_wrapper.py index c9957a8..57248b0 100644 --- a/app/common/exceptions/http_exception_wrapper.py +++ b/app/common/exceptions/http_exception_wrapper.py @@ -3,10 +3,5 @@ def http_exception(status_code: int, msg: str, _input, _detail) -> HTTPException: return HTTPException( - status_code=status_code, - detail={ - "msg": msg, - "input": _input, - "detail": _detail - } + status_code=status_code, detail={"msg": msg, "input": _input, "detail": _detail} ) diff --git a/app/common/middlewares/__init__.py b/app/common/middlewares/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/common/middlewares/exception_handler.py b/app/common/middlewares/exception_handler.py new file mode 100644 index 0000000..527f72c --- /dev/null +++ b/app/common/middlewares/exception_handler.py @@ -0,0 +1,90 @@ +"""Exception handling middleware is defined here.""" + +import traceback + +from fastapi import FastAPI, Request +from starlette.middleware.base import BaseHTTPMiddleware +from starlette.responses import JSONResponse + +from app.common.middlewares.middleware_utils import _normalize_path +from app.observability.metrics import Metrics + + +class ExceptionHandlerMiddleware( + BaseHTTPMiddleware +): # pylint: disable=too-few-public-methods + """Handle exceptions, so they become http response code 500 - Internal Server Error if not handled as HTTPException + previously. + Attributes: + app (FastAPI): The FastAPI application instance. + """ + + def __init__(self, app: FastAPI, metrics: Metrics): + """ + Universal exception handler middleware init function. + Args: + app (FastAPI): The FastAPI application instance. + """ + + super().__init__(app) + self.metrics = metrics + + @staticmethod + async def prepare_request_info(request: Request) -> dict: + """ + Function prepares request input data + Args: + request (Request): Request instance. + Returns: + dict: Request input data. + """ + + request_info = { + "method": request.method, + "url": str(request.url), + "path_params": dict(request.path_params), + "query_params": dict(request.query_params), + "headers": dict(request.headers), + } + + try: + request_info["body"] = await request.json() + return request_info + except: + try: + request_info["body"] = str(await request.body()) + return request_info + except: + request_info["body"] = "Could not read request body" + return request_info + + async def dispatch(self, request: Request, call_next): + """ + Dispatch function for sending errors to user from API + Args: + request (Request): The incoming request object. + call_next: function to extract. + """ + + try: + return await call_next(request) + except Exception as e: + request_info = await self.prepare_request_info(request) + self.metrics.http.errors.add( + 1, + { + "method": request.method, + "path": _normalize_path(request), + "error_type": type(e).__name__, + }, + ) + return JSONResponse( + status_code=500, + content={ + "message": "Internal server error", + "error_type": e.__class__.__name__, + "request": request_info, + "detail": str(e), + "traceback": traceback.format_exc().splitlines(), + }, + ) diff --git a/app/common/middlewares/middleware_utils.py b/app/common/middlewares/middleware_utils.py new file mode 100644 index 0000000..5e3f5d0 --- /dev/null +++ b/app/common/middlewares/middleware_utils.py @@ -0,0 +1,12 @@ +from fastapi import Request + + +def _normalize_path(request: Request) -> str: + """ + Normalize path to avoid high-cardinality metrics. + """ + + route = request.scope.get("route") + if route and hasattr(route, "path"): + return route.path + return request.url.path diff --git a/app/common/middlewares/prometheus_handler.py b/app/common/middlewares/prometheus_handler.py new file mode 100644 index 0000000..b7cf9c1 --- /dev/null +++ b/app/common/middlewares/prometheus_handler.py @@ -0,0 +1,41 @@ +"""Observability middleware is defined here.""" + +import time + +from fastapi import FastAPI, Request +from starlette.middleware.base import BaseHTTPMiddleware + +from app.common.middlewares.middleware_utils import _normalize_path +from app.observability.metrics import Metrics + + +class ObservabilityMiddleware(BaseHTTPMiddleware): + + def __init__(self, app: FastAPI, metrics: Metrics): + """Obervability middleware class for http metrics with prometheus + + Args: + app (FastAPI): FastAPI app instance + metrics (Metrics): Metrics with http field connectable with prometheus + """ + super().__init__(app) + self._http_metrics = metrics.http + + async def dispatch(self, request: Request, call_next): + + path = _normalize_path(request) + method = request.method + self._http_metrics.requests_started.add(1, {"method": method, "path": path}) + self._http_metrics.inflight_requests.add(1) + start = time.monotonic() + response = await call_next(request) + duration = time.monotonic() - start + self._http_metrics.requests_finished.add( + 1, + {"method": method, "path": path, "status_code": response.status_code}, + ) + self._http_metrics.request_processing_duration.record( + duration, {"method": method, "path": path} + ) + self._http_metrics.inflight_requests.add(-1) + return response diff --git a/app/common/modules/__init__.py b/app/common/modules/__init__.py new file mode 100644 index 0000000..d639e53 --- /dev/null +++ b/app/common/modules/__init__.py @@ -0,0 +1,11 @@ +from .attribute_parser import attribute_parser +from .data_restorator import data_restorator +from .effects_api_gateway import EffectsAPIGateway +from .matrix_builder import matrix_builder +from .name_mappings import ( + ATTRIBUTES_MAP, + BUILDINGS_DROP_COLUMNS, + EFFECTS_MAP, + SERVICE_DROP_COLUMNS, +) +from .objectnat_calculator import objectnat_calculator diff --git a/app/effects/modules/attribute_parser.py b/app/common/modules/attribute_parser.py similarity index 69% rename from app/effects/modules/attribute_parser.py rename to app/common/modules/attribute_parser.py index 1aa4e87..581e4ee 100644 --- a/app/effects/modules/attribute_parser.py +++ b/app/common/modules/attribute_parser.py @@ -1,10 +1,7 @@ -import json import asyncio -import pandas as pd import geopandas as gpd - -from app.dependencies import http_exception +import pandas as pd class AttributeParser: @@ -14,7 +11,7 @@ class AttributeParser: @staticmethod async def parse_all_from_buildings( - living_buildings: pd.DataFrame | gpd.GeoDataFrame, + living_buildings: pd.DataFrame | gpd.GeoDataFrame, ) -> gpd.GeoDataFrame: """ Function purses living building area for buildings from nested response @@ -34,37 +31,48 @@ async def parse_all_from_buildings( if living_buildings["storeys_count"].isna().all(): living_buildings["storeys_count"] = await asyncio.to_thread( living_buildings["physical_objects"].apply, - lambda x: x[0].get("properties").get("Количество этажей") + lambda x: x[0].get("properties").get("Количество этажей"), ) living_buildings["building_id"] = await asyncio.to_thread( living_buildings["physical_objects"].apply, lambda x: x[0]["physical_object_id"], ) living_buildings = living_buildings.drop( - ['object_geometry_id', 'territory', 'address', 'osm_id', 'physical_objects', 'services'], + [ + "object_geometry_id", + "territory", + "address", + "osm_id", + "physical_objects", + "services", + ], axis=1, ) return living_buildings @staticmethod def _parse_service_capacity( - services:gpd.GeoDataFrame, + services: gpd.GeoDataFrame, service_default_capacity: int ) -> gpd.GeoDataFrame: """ Function parses capacity attributes from nested response Args: services (gpd.GeoDataFrame): nested response from api as feature collection + service_default_capacity (int): default capacity to fill Returns: gpd.GeoDataFrame: service capacity with parsed storeys data. Can be empty """ - services["capacity"] = services["services"].apply(lambda x: x[0].get("capacity")) + services["capacity"] = ( + services["services"] + .apply(lambda x: x[0].get("capacity")) + .fillna(service_default_capacity) + .astype(int) + ) return services @staticmethod - def _parse_service_id( - services: gpd.GeoDataFrame - ) -> gpd.GeoDataFrame: + def _parse_service_id(services: gpd.GeoDataFrame) -> gpd.GeoDataFrame: """ Function parses service id from nested response Args: @@ -73,20 +81,23 @@ def _parse_service_id( gpd.GeoDataFrame: service id with parsed storeys data. Can be empty """ - services["service_id"] = services["services"].apply(lambda x: x[0].get("service_id")) + services["service_id"] = services["services"].apply( + lambda x: x[0].get("service_id") + ) return services async def parse_all_from_services( - self, - services: gpd.GeoDataFrame, + self, services: gpd.GeoDataFrame, service_default_capacity: int ) -> gpd.GeoDataFrame: """ Function parses all required data from service request data Args: services (gpd.GeoDataFrame): nested response from api as feature collection + service_default_capacity(int): service default capacity value Returns: gpd.GeoDataFrame: service capacity with parsed storeys data. Can be empty """ + services = services.copy() if services.empty: return services @@ -96,12 +107,21 @@ async def parse_all_from_services( ) services = await asyncio.to_thread( self._parse_service_capacity, - services=services + services=services, + service_default_capacity=service_default_capacity, ) services = services.drop( - ['object_geometry_id', 'territory', 'address', 'osm_id', 'physical_objects', 'services'], - axis=1 + [ + "object_geometry_id", + "territory", + "address", + "osm_id", + "physical_objects", + "services", + ], + axis=1, ) return services + attribute_parser = AttributeParser() diff --git a/app/effects/modules/data_restorator.py b/app/common/modules/data_restorator.py similarity index 65% rename from app/effects/modules/data_restorator.py rename to app/common/modules/data_restorator.py index 1e7a95f..719cf23 100644 --- a/app/effects/modules/data_restorator.py +++ b/app/common/modules/data_restorator.py @@ -1,11 +1,10 @@ from typing import Literal +import geopandas as gpd import numpy as np import pandas as pd -import geopandas as gpd -from objectnat import get_balanced_buildings -from app.dependencies import http_exception +from app.common.exceptions.http_exception_wrapper import http_exception class DataRestorator: @@ -15,7 +14,7 @@ class DataRestorator: @staticmethod def _restore_stores( - buildings: gpd.GeoDataFrame, + buildings: gpd.GeoDataFrame, ) -> gpd.GeoDataFrame: """ Function to restore stores from db, have to include columns stores_count @@ -36,7 +35,7 @@ def _restore_stores( @staticmethod def _restore_target_population( - buildings: gpd.GeoDataFrame, + buildings: gpd.GeoDataFrame, ) -> int: """ Function estimates target population for territory @@ -48,13 +47,35 @@ def _restore_target_population( local_crs = buildings.estimate_utm_crs() buildings = buildings.to_crs(local_crs) - return int(sum(buildings.area * buildings["storeys_count"]) * 0.8/33) + return int(sum(buildings.area * buildings["storeys_count"]) * 0.8 / 33) + + @staticmethod + def _balance_population( + buildings: gpd.GeoDataFrame, + population: int, + ) -> gpd.GeoDataFrame: + """ + Function distributes population between buildings proportionally to their living area + Args: + buildings (gpd.GeoDataFrame): living buildings data with "living_area" attribute + population (int): total population to distribute + Returns: + gpd.GeoDataFrame: buildings data with restored "population" attribute + """ + + shares = buildings["living_area"] / buildings["living_area"].sum() + buildings["population"] = np.floor(shares * population).astype(int) + remainder = int(population - buildings["population"].sum()) + if remainder > 0: + top = (shares * population).mod(1).nlargest(remainder).index + buildings.loc[top, "population"] += 1 + return buildings # ToDo delete crs transformation def _restore_population( - self, - buildings: gpd.GeoDataFrame, - target_population: int | None = None, + self, + buildings: gpd.GeoDataFrame, + target_population: int | None = None, ): """ Function fills population data with objectnat population restoration @@ -70,18 +91,21 @@ def _restore_population( target_population = self._restore_target_population(buildings) local_crs = buildings.estimate_utm_crs() buildings = buildings.to_crs(local_crs) + buildings["storeys_count"] = buildings["storeys_count"].apply( + lambda x: int(round(x)) + ) buildings["living_area"] = buildings.area * buildings["storeys_count"] * 0.8 buildings["living_area"] = buildings["living_area"].astype(int) - balanced_buildings = get_balanced_buildings( - living_buildings=buildings, + buildings = self._balance_population( + buildings=buildings, population=int(target_population), ) - return balanced_buildings.to_crs(4326) + return buildings.to_crs(4326) @staticmethod def _generate_demand_per_building( - buildings: gpd.GeoDataFrame, - target_demand: int |float, + buildings: gpd.GeoDataFrame, + target_demand: int | float, ) -> pd.DataFrame | gpd.GeoDataFrame: """ Function generates random demands by probability with population data per building @@ -95,18 +119,20 @@ def _generate_demand_per_building( p = buildings["population"] / buildings["population"].sum() rng = np.random.default_rng(seed=0) r = pd.Series(0, p.index) - choice = np.unique(rng.choice(p.index, int(target_demand), p=p.values), return_counts=True) + choice = np.unique( + rng.choice(p.index, int(target_demand), p=p.values), return_counts=True + ) choice = r.add(pd.Series(choice[1], choice[0]), fill_value=0) buildings["demand"] = choice.astype(int) return buildings # Todo review provision model or at least create capacity solver def restore_demands( - self, - buildings: gpd.GeoDataFrame, - service_normative: int, - service_normative_type: Literal["unit", "capacity"], - target_population: int | None = None, + self, + buildings: gpd.GeoDataFrame, + service_normative: int, + service_normative_type: Literal["unit", "capacity"], + target_population: int | None = None, ) -> gpd.GeoDataFrame: """ Function restores demands in buildings by population for service @@ -126,24 +152,24 @@ def restore_demands( target_population=target_population, ) if service_normative_type == "capacity": - target_total_demand = buildings["population"].sum() / 1000 * service_normative + target_total_demand = ( + buildings["population"].sum() / 1000 * service_normative + ) buildings = self._generate_demand_per_building( - buildings=buildings, - target_demand=target_total_demand + buildings=buildings, target_demand=target_total_demand ) return buildings + elif service_normative_type == "unit": + buildings["demand"] = buildings["population"].astype(int).copy() + return buildings else: raise http_exception( - status_code=400, + status_code=500, msg="Service demand normative not found", _input={ "service_normative_type": service_normative_type, }, - _detail={ - "available_demand_type": [ - "num", "capacity" - ] - } + _detail={"available_demand_type": ["unit", "capacity"]}, ) diff --git a/app/common/modules/effects_api_gateway.py b/app/common/modules/effects_api_gateway.py new file mode 100644 index 0000000..52b7f34 --- /dev/null +++ b/app/common/modules/effects_api_gateway.py @@ -0,0 +1,395 @@ +import asyncio + +import geopandas as gpd +import pandas as pd +from shapely.geometry import shape + +from app.common.api_handler.api_handler import APIHandler +from app.common.auth.service_auth import USER_ID_HEADER +from app.common.exceptions.http_exception_wrapper import http_exception + + +class EffectsAPIGateway: + + def __init__(self, api_handler: APIHandler) -> None: + self.api_handler = api_handler + + async def get_project_id_by_scenario(self, scenario_id: int, token: str) -> int: + """ + Function retrieves project ID based on scenario ID from Urban API. + Args: + scenario_id (int): Scenario ID from Urban API. + token (str): User access token. + Returns: + int: Project ID from Urban API. + Raises: + Any: HTTP from Urban API. + """ + + proj_resp = await self.api_handler.get( + f"/api/v1/scenarios/{scenario_id}", + headers={USER_ID_HEADER: token} if token else None, + ) + return proj_resp["project"]["project_id"] + + async def get_service_normative( + self, + territory_id: int, + context_ids: list[int], + service_type_id: int, + token: str, + ) -> dict[str, int | str]: + """ + Function retrieves normative data from urban_api + Args: + territory_id: territory id to get normative from + context_ids: context id to get normative from + service_type_id: service to get normative from + token: auth token to get normative from + Returns: + dict[str, int | str]: normative data with normative value and normative type (Literal["time", "dist"]) + Raises: + 400, http exception id not found + """ + + if len(context_ids) == 1: + response = await self.api_handler.get( + f"/api/v1/territory/{context_ids[0]}/normatives", + headers={USER_ID_HEADER: token} if token else None, + ) + request_ter_id = context_ids[0] + else: + response = await self.api_handler.get( + f"/api/v1/territory/{territory_id}/normatives", + headers={USER_ID_HEADER: token} if token else None, + ) + request_ter_id = territory_id + response_df = pd.DataFrame.from_records(response) + response_df["service_type_id"] = response_df["service_type"].apply( + lambda x: x["id"] if x else None + ) + service_type = response_df[ + response_df["service_type_id"] == service_type_id + ].copy() + if len(service_type) < 1: + raise http_exception( + 400, + msg="Service type id not found in urban_db for provided territory/context ids. ", + _input={ + "territory_id": territory_id, + "context_ids": context_ids, + "service_type_id": service_type_id, + }, + _detail={ + "Available service ids": response_df["service_type_id"].to_list() + }, + ) + + service_type = ( + service_type[service_type["year"] == service_type["year"].max()] + .iloc[0] + .to_dict() + ) + + if service_type["service_type"]["id"] == service_type_id: + if not pd.isna(service_type["radius_availability_meters"]): + service_type["normative_value"] = service_type[ + "radius_availability_meters" + ] + service_type["normative_type"] = "dist" + if not pd.isna(service_type.get("services_per_1000_normative")): + service_type["capacity_type"] = "unit" + else: + service_type["capacity_type"] = "capacity" + return service_type + elif not pd.isna(service_type["time_availability_minutes"]): + service_type["normative_value"] = service_type[ + "time_availability_minutes" + ] + service_type["normative_type"] = "time" + if not pd.isna(service_type.get("services_per_1000_normative")): + service_type["capacity_type"] = "unit" + else: + service_type["capacity_type"] = "capacity" + return service_type + else: + raise http_exception( + status_code=404, + msg="Service type normative not found in urban_db. ", + _input={ + "territory_id": territory_id, + "context_ids": context_ids, + "request_ter_id": request_ter_id, + "service_type_id": service_type_id, + }, + _detail={ + "Available service ids": [ + service_type["id"] for service_type in response + ] + }, + ) + raise http_exception( + status_code=404, + msg="Service type normative not found in urban_db. Try another year or service type.", + _input={ + "territory_id": territory_id, + "context_ids": context_ids, + "request_ter_id": request_ter_id, + "service_type_id": service_type_id, + }, + _detail={"Available service ids": response_df["service_type_id"].to_list()}, + ) + + async def get_project_data( + self, project_id: int, token: str + ) -> dict[str, int | dict]: + """ + Function retrieves project territory data from urban_api + Args: + project_id: project id to get territory from + token: authentication token to retrieve data + Returns: + dict with "geometry" field as dict with "type" and "coordinates" fields and field "base_scenario_id" + """ + + response = await self.api_handler.get( + endpoint_url=f"/api/v1/projects/{project_id}", + headers={USER_ID_HEADER: token} if token else None, + ) + + return response + + async def get_scenario_buildings( + self, scenario_id: int, token: str + ) -> gpd.GeoDataFrame: + """ + Function retrieves scenario buildings data from urban_api + Args: + scenario_id: scenario id to get buildings from + token: authentication token to retrieve data + Returns: + gpd.GeoDataFrame: buildings layer, can be empty + """ + + buildings = await self.api_handler.get( + endpoint_url=f"/api/v1/scenarios/{scenario_id}/geometries_with_all_objects", + params={"physical_object_type_id": 4}, + headers={USER_ID_HEADER: token} if token else None, + ) + buildings_gdf = gpd.GeoDataFrame.from_features(buildings) + if buildings_gdf.empty: + return buildings_gdf + buildings_gdf.set_crs(4326, inplace=True) + return buildings_gdf + + async def get_project_context_buildings( + self, scenario_id: int, token: str + ) -> gpd.GeoDataFrame: + """ + Function retrieves scenario context buildings data from urban_api + Args: + scenario_id: scenario id to get buildings from + token: Authorization token to retrieve data + Returns: + gpd.GeoDataFrame: buildings layer + Raises: + 404, http exception living buildings not found + """ + + context_buildings = await self.api_handler.get( + endpoint_url=f"/api/v1/scenarios/{scenario_id}/context/geometries_with_all_objects", + params={ + "physical_object_type_id": 4, + }, + headers={USER_ID_HEADER: token} if token else None, + ) + context_buildings_gdf = gpd.GeoDataFrame.from_features(context_buildings) + if context_buildings_gdf.empty: + return context_buildings_gdf + context_buildings_gdf.set_crs(4326, inplace=True) + return context_buildings_gdf + + async def get_scenario_services( + self, scenario_id: int, service_type_id: int, token: str + ) -> gpd.GeoDataFrame: + """ + Function retrieves scenario services data from urban_api + Args: + scenario_id: scenario id to get services from + service_type_id: service to get services from + token: Authorization token to retrieve data + Returns: + gpd.GeoDataFrame: services layer, can be empty + """ + + services = await self.api_handler.get( + endpoint_url=f"/api/v1/scenarios/{scenario_id}/geometries_with_all_objects", + params={ + "service_type_id": service_type_id, + }, + headers={USER_ID_HEADER: token} if token else None, + ) + services_gdf = gpd.GeoDataFrame.from_features(services) + if services_gdf.empty: + return services_gdf + services_gdf.set_crs(4326, inplace=True) + return services_gdf + + async def get_project_context_services( + self, + scenario_id: int, + service_type_id: int, + token: str, + ) -> gpd.GeoDataFrame: + """ + Function retrieves scenario context services data from urban_api + Args: + scenario_id: scenario id to get services from + service_type_id: service to get services from + token: Authorization token to retrieve data + Returns: + gpd.GeoDataFrame: context services layer. Can be empty + """ + + context_services = await self.api_handler.get( + endpoint_url=f"/api/v1/scenarios/{scenario_id}/context/geometries_with_all_objects", + params={ + "service_type_id": service_type_id, + }, + headers={USER_ID_HEADER: token} if token else None, + ) + context_services_gdf = gpd.GeoDataFrame.from_features(context_services) + if context_services_gdf.empty: + return context_services_gdf + context_services_gdf.set_crs(4326, inplace=True) + return context_services_gdf + + async def get_scenario_population_data( + self, scenario_id: int | None, token: str + ) -> int | None: + """ + Function retrieves population data from urban_api + Args: + scenario_id: scenario id to get population data from + token: Authorization token to retrieve data + Returns: + int | none: population data layer, if < 1 returns None + """ + + population = await self.api_handler.get( + endpoint_url=f"/api/v1/scenarios/{scenario_id}/indicators_values", + params={ + "indicator_ids": 1, + }, + headers={USER_ID_HEADER: token} if token else None, + ) + + if len(population) < 1 or (value := population[0]["value"]) < 1: + return None + return value + + async def get_context_population( + self, territory_ids_list: list[int], token: str + ) -> int: + """ + Function retrieves territory population data from urban_api by territory id + Args: + territory_ids_list: list[int]: territory ids list to get population data from + token: Authorization token to retrieve data + Returns: + gpd.GeoDataFrame: territory population data layer + """ + + task_list = [ + self.api_handler.get( + endpoint_url=f"/api/v1/territory/{territory_id}/indicator_values", + params={"indicator_ids": 1}, + headers={USER_ID_HEADER: token} if token else None, + ) + for territory_id in territory_ids_list + ] + + result = await asyncio.gather(*task_list) + return sum([item[0]["value"] for item in result]) + + async def get_project_territory( + self, project_id: int, token: str + ) -> gpd.GeoDataFrame: + """ + Function retrieves territory data from urban_api + Args: + project_id: project id to get territory data from + token: Authorization token to retrieve data + Returns: + gpd.GeoDataFrame: territory data layer + """ + + territory = await self.api_handler.get( + endpoint_url=f"/api/v1/projects/{project_id}/territory", + headers={USER_ID_HEADER: token} if token else None, + ) + territory_gdf = gpd.GeoDataFrame( + geometry=[shape(territory["geometry"])], crs=4326 + ) + return territory_gdf + + async def get_default_capacity(self, service_type_id: int) -> int: + """ + Function retrieves default capacity data from urban_api + Args: + service_type_id (int): service type id to get default capacity data from + Returns: + int: default capacity value + """ + + service_types = await self.api_handler.get(endpoint_url="/api/v1/service_types") + service_types_df = pd.DataFrame.from_records(service_types).fillna(0) + return service_types_df[ + service_types_df["service_type_id"] == service_type_id + ].iloc[0]["capacity_modeled"] + + async def get_services_with_context( + self, scenario_id: int, service_type_id: int, token: str | None = None + ) -> gpd.GeoDataFrame: + """ + Function retrieves service by service_type_id for scenario ID from urban api with context. + Args: + scenario_id (int): Scenario ID from Urban API. + service_type_id (int): Service type ID from Urban API. + token (str | None): Auth token to retrieve data from Urban API. Default to None + Returns: + gpd.GeoDataFrame: layer with services in 4326 crs. + """ + + services = await self.api_handler.get( + endpoint_url=f"/api/v1/scenarios/{scenario_id}/context/services_with_geometry", + params={ + "service_type_id": service_type_id, + "include_scenario_objects": True, + }, + headers={USER_ID_HEADER: token} if token else None, + ) + return gpd.GeoDataFrame.from_features(services, crs=4326) + + async def get_physical_objects_with_context( + self, scenario_id: int, physical_object_type_id: int, token: str | None = None + ): + """ + Function retrieves physical objects by physical_object_type_id for scenario ID from urban api with context. + Args: + scenario_id (int): Scenario ID from Urban API. + physical_object_type_id (int): Physical object type ID from Urban API. + token (str | None): Auth token to retrieve data from Urban API. Default to None + Returns: + gpd.GeoDataFrame: layer with physical_objects in 4326 crs. + """ + + physical_objects = await self.api_handler.get( + endpoint_url=f"/api/v1/scenarios/{scenario_id}/context/services_with_geometry", + params={ + "physical_object_type_id": physical_object_type_id, + "include_scenario_objects": True, + }, + headers={USER_ID_HEADER: token} if token else None, + ) + return gpd.GeoDataFrame.from_features(physical_objects, crs=4326) diff --git a/app/effects/modules/matrix_builder.py b/app/common/modules/matrix_builder.py similarity index 66% rename from app/effects/modules/matrix_builder.py rename to app/common/modules/matrix_builder.py index 821b0c7..eb153a1 100644 --- a/app/effects/modules/matrix_builder.py +++ b/app/common/modules/matrix_builder.py @@ -1,8 +1,8 @@ from typing import Literal +import geopandas as gpd import numpy as np import pandas as pd -import geopandas as gpd from scipy.spatial import KDTree @@ -10,10 +10,10 @@ class MatrixBuilder: @staticmethod def calculate_availability_matrix( - buildings: gpd.GeoDataFrame, - services: gpd.GeoDataFrame, - normative_value: int, - normative_type: Literal["time", "dist"] + buildings: gpd.GeoDataFrame, + services: gpd.GeoDataFrame, + normative_value: int, + normative_type: Literal["time", "dist"], ) -> pd.DataFrame: """ Calculated availability matrix with walk simulation @@ -27,20 +27,26 @@ def calculate_availability_matrix( """ if normative_type == "time": - normative_value = (normative_value * 1000/60 * 40 )/1.41 + normative_value = (normative_value * 1000 / 60 * 40) / 1.41 else: normative_value = (normative_value * 3) / 1.41 local_crs = buildings.estimate_utm_crs() buildings = buildings.to_crs(local_crs).set_index(buildings.index, drop=True) services = services.to_crs(local_crs).set_index(services.index, drop=True) - buildings_points = [geometry.coords[0] for geometry in buildings.geometry.centroid] - services_points = [geometry.coords[0] for geometry in services.geometry.centroid] + buildings_points = [ + geometry.coords[0] for geometry in buildings.geometry.centroid + ] + services_points = [ + geometry.coords[0] for geometry in services.geometry.centroid + ] buildings_kd_tree = KDTree(buildings_points) services_kd_tree = KDTree(services_points) distances = buildings_kd_tree.sparse_distance_matrix( - other=services_kd_tree, - max_distance=normative_value * 3) - matrix = pd.DataFrame.sparse.from_spmatrix(distances, index=buildings.index, columns=services.index) + other=services_kd_tree, max_distance=normative_value * 3 + ) + matrix = pd.DataFrame.sparse.from_spmatrix( + distances, index=buildings.index, columns=services.index + ) matrix = matrix.sparse.to_dense() matrix.replace(0.0, np.nan, inplace=True) return matrix diff --git a/app/common/modules/name_mappings.py b/app/common/modules/name_mappings.py new file mode 100644 index 0000000..7511959 --- /dev/null +++ b/app/common/modules/name_mappings.py @@ -0,0 +1,42 @@ +ATTRIBUTES_MAP = { + "storeys_count": "Количество этажей", + "population": "Население (чел)", + "demand": "Спрос (чел)", + "demand_left": "Неудовлетворённый спрос (чел)", + "distance": "Расстояние (м)", + "avg_dist": "Средняя доступность до сервиса (м)", + "capacity": "Вместимость (чел)", + "capacity_left": "Профицит мест (чел)", + "living_area": "Жилая площадь (кв.м)", + "service_load": "Нагрузка на сервис", + "min_dist": "Минмиальное расстояне до сервиса (м)", + "building_index": "ID здания", + "service_index": "ID сервиса", + "supplied_demands_within": "Удовлетворённый спрос в нормативной доступности (чел)", + "supplied_demands_without": "Удовлетворённый спрос вне нормативной доступности (чел)", + "carried_capacity_within": "Обеспечено в радиусе нормативной доступности (чел)", + "carried_capacity_without": "Обеспечено вне радиуса нормативной доступности (чел)", + "provision_value": "Оценка обеспеченности", + "supplied_demands_within_before": "Удовлетворённый спрос в нормативной доступности (до) (чел)", + "us_demands_within_before": "Неудовлетворённый спрос в нормативной доступности (до) (чел)", + "supplied_demands_without_before": "Удовлетворённый спрос вне нормативной доступности (до) (чел)", + "us_demands_without_before": "Неудовлетворённый спрос вне нормативной доступности (до) (чел)", + "supplied_demands_within_after": "Удовлетворённый спрос в нормативной доступности (после) (чел)", + "us_demands_within_after": "Неудовлетворённый спрос в нормативной доступности (после) (чел)", + "supplied_demands_without_after": "Удовлетворённый спрос вне нормативной доступности (после) (чел)", + "us_demands_without_after": "Неудовлетворённый спрос вне нормативной доступности (после) (чел)", + "is_scenario_object": "Сценарный объект", +} + +EFFECTS_MAP = { + "absolute_total": "Абсолютный эффект (чел)", + "index_total": "Индексный эффект", + "absolute_scenario_project": "Абсолютный эффект на территории проекта", + "index_scenario_project": "Индексный эффект на территории проекта", + "absolute_within": "Абсолютный эффект в нормативной доступности", + "demand": "Спрос (чел)", + "is_project": "Проектный объект", +} + +SERVICE_DROP_COLUMNS = ["is_locked"] +BUILDINGS_DROP_COLUMNS = SERVICE_DROP_COLUMNS + ["is_project"] diff --git a/app/effects/modules/objectnat_calculator.py b/app/common/modules/objectnat_calculator.py similarity index 52% rename from app/effects/modules/objectnat_calculator.py rename to app/common/modules/objectnat_calculator.py index 449a4e2..95779e8 100644 --- a/app/effects/modules/objectnat_calculator.py +++ b/app/common/modules/objectnat_calculator.py @@ -1,21 +1,16 @@ -import json -from typing import Literal - -import pandas as pd import geopandas as gpd +import pandas as pd from objectnat import get_service_provision -from app.dependencies import http_exception - class ObjectNatCalculator: @staticmethod def evaluate_provision( - buildings: gpd.GeoDataFrame, - services: gpd.GeoDataFrame, - matrix: pd.DataFrame, - service_normative: int + buildings: gpd.GeoDataFrame, + services: gpd.GeoDataFrame, + matrix: pd.DataFrame, + service_normative: int, ) -> dict[str, gpd.GeoDataFrame]: """ Function calculates provision and writes results as dict with fields "buildings", "services" and "links" @@ -32,7 +27,7 @@ def evaluate_provision( buildings=buildings, services=services, adjacency_matrix=matrix, - threshold=int(service_normative*1000/60 * 40), + threshold=int(service_normative * 1000 / 60 * 40), ) return { @@ -43,11 +38,11 @@ def evaluate_provision( @staticmethod def _calculate_index( - supplied_demand_after: pd.Series, - supplied_demand_before: pd.Series, - unsupplied_demand_after: pd.Series, - unsupplied_demand_before: pd.Series, - total_demand: int + supplied_demand_after: pd.Series, + supplied_demand_before: pd.Series, + unsupplied_demand_after: pd.Series, + unsupplied_demand_before: pd.Series, + total_demand: int, ) -> pd.Series: """ Function calculates index effects marks for provided objects @@ -60,21 +55,18 @@ def _calculate_index( """ result = ( - ( - supplied_demand_after - supplied_demand_before - ) - ( - unsupplied_demand_after - unsupplied_demand_before - ) - ) / total_demand + (supplied_demand_after - supplied_demand_before) + - (unsupplied_demand_after - unsupplied_demand_before) + ) / total_demand return result # ToDo fix is_project attribute @staticmethod def _calculate_absolute( - supplied_demand_after: pd.Series, - supplied_demand_before: pd.Series, - unsupplied_demand_after: pd.Series, - unsupplied_demand_before: pd.Series, + supplied_demand_after: pd.Series, + supplied_demand_before: pd.Series, + unsupplied_demand_after: pd.Series, + unsupplied_demand_before: pd.Series, ) -> pd.Series: """ Function calculates absolute effects marks for provided objects @@ -85,17 +77,16 @@ def _calculate_absolute( unsupplied_demand_before (pd.Series): unsupplied demand for base scenario """ - result = ( - supplied_demand_after-supplied_demand_before - ).apply(lambda x: max(0, x)) - ( - unsupplied_demand_after-unsupplied_demand_before - ).apply(lambda x: max(0, x) - ) + result = (supplied_demand_after - supplied_demand_before).apply( + lambda x: max(0, x) + ) - (unsupplied_demand_after - unsupplied_demand_before).apply( + lambda x: max(0, x) + ) return result def _calculate_effects( - self, - effects: pd.DataFrame | gpd.GeoDataFrame, + self, + effects: pd.DataFrame | gpd.GeoDataFrame, ) -> pd.DataFrame | gpd.GeoDataFrame: """ Function calculates provision effects @@ -105,42 +96,45 @@ def _calculate_effects( pd.Series: effects results """ + # ToDo fix calculation without/before effects = effects.copy() - supplied_demand_within_before = effects["supplyed_demands_within_before"].fillna(0) - supplied_demand_without_before = effects["supplyed_demands_without_before"].fillna(0) - supplied_demand_within_after = effects["supplyed_demands_within_after"].fillna(0) - supplied_demand_without_after = effects["supplyed_demands_without_after"].fillna(0) + supplied_demand_within_before = effects[ + "supplied_demands_within_before" + ].fillna(0) + supplied_demand_without_before = effects[ + "supplied_demands_without_before" + ].fillna(0) + supplied_demand_within_after = effects["supplied_demands_within_after"].fillna( + 0 + ) + supplied_demand_without_after = effects[ + "supplied_demands_without_after" + ].fillna(0) unsupplied_demand_within_before = effects["us_demands_within_before"].fillna(0) unsupplied_demand_within_after = effects["us_demands_within_after"].fillna(0) - total_supplied_demands_before = supplied_demand_without_before + supplied_demand_within_before - total_supplied_demands_after = supplied_demand_without_after + supplied_demand_within_after - total_us_demands_before = effects["demand"] - total_supplied_demands_before - total_us_demands_after = effects["demand"] - total_supplied_demands_after + total_supplied_demands_before = supplied_demand_without_before + total_supplied_demands_after = supplied_demand_without_after + total_us_demands_before = effects["us_demands_without_before"].fillna(0) + total_us_demands_after = effects["us_demands_without_after"].fillna(0) total_demand = int(effects["demand"].sum()) - project_total_supplied_demands_before = effects[ - effects["is_project"] - ]["supplyed_demands_without_before"].fillna(0) + effects[ - effects["is_project"] - ]["supplyed_demands_within_before"].fillna(0) - - project_total_supplied_demands_after = effects[ - effects["is_project"] - ]["supplyed_demands_without_after"].fillna(0) + effects[ - effects["is_project"] - ]["supplyed_demands_within_after"].fillna(0) - - project_total_us_demands_before = effects[ - effects["is_project"] - ]["demand"].fillna(0) - effects[ - effects["is_project"] - ]["supplyed_demands_within_before"].fillna(0) - - project_total_us_demands_after = effects[ - effects["is_project"] - ]["demand"].fillna(0) - effects[ - effects["is_project"] - ]["supplyed_demands_within_after"].fillna(0) + effects.dropna(subset="is_project", inplace=True) + + project_total_supplied_demands_before = effects[effects["is_project"]][ + "supplied_demands_without_before" + ].fillna(0) + + project_total_supplied_demands_after = effects[effects["is_project"]][ + "supplied_demands_without_after" + ].fillna(0) + + project_total_us_demands_before = effects[effects["is_project"]][ + "us_demands_without_before" + ].fillna(0) + + project_total_us_demands_after = effects[effects["is_project"]][ + "us_demands_without_after" + ].fillna(0) project_total_demand = int(effects[effects["is_project"]]["demand"].sum()) @@ -157,32 +151,38 @@ def _calculate_effects( unsupplied_demand_before=total_us_demands_before, total_demand=total_demand, ) - effects["absolute_scenario_project"] = self._calculate_absolute( - supplied_demand_before=project_total_supplied_demands_before, - supplied_demand_after=project_total_supplied_demands_after, - unsupplied_demand_after=project_total_us_demands_after, - unsupplied_demand_before=project_total_us_demands_before, + effects["absolute_scenario_project"] = None + effects.loc[effects["is_project"], ["absolute_scenario_project"]] = ( + self._calculate_absolute( + supplied_demand_before=project_total_supplied_demands_before, + supplied_demand_after=project_total_supplied_demands_after, + unsupplied_demand_after=project_total_us_demands_after, + unsupplied_demand_before=project_total_us_demands_before, + ) ) - effects["index_scenario_project"] = self._calculate_index( - supplied_demand_after=project_total_supplied_demands_after, - supplied_demand_before=project_total_supplied_demands_before, - unsupplied_demand_after=project_total_us_demands_after, - unsupplied_demand_before=project_total_us_demands_before, - total_demand=project_total_demand, + effects["index_scenario_project"] = None + effects.loc[effects["is_project"], ["index_scenario_project"]] = ( + self._calculate_index( + supplied_demand_after=project_total_supplied_demands_after, + supplied_demand_before=project_total_supplied_demands_before, + unsupplied_demand_after=project_total_us_demands_after, + unsupplied_demand_before=project_total_us_demands_before, + total_demand=project_total_demand, + ) ) effects["absolute_within"] = self._calculate_absolute( supplied_demand_before=supplied_demand_within_before, supplied_demand_after=supplied_demand_within_after, unsupplied_demand_before=unsupplied_demand_within_before, - unsupplied_demand_after=unsupplied_demand_within_after + unsupplied_demand_after=unsupplied_demand_within_after, ) return effects # ToDo split function def estimate_effects( - self, - provision_before: gpd.GeoDataFrame, - provision_after:gpd.GeoDataFrame, + self, + provision_before: gpd.GeoDataFrame, + provision_after: gpd.GeoDataFrame, ) -> pd.DataFrame | gpd.GeoDataFrame: """ Main function which calculates provision and estimates effects @@ -193,45 +193,56 @@ def estimate_effects( gpd.GeoDataFrame: layer with effects, provision before and after attributes """ - provision_before["supplyed_demands_within_before"] = provision_before["supplyed_demands_within"].copy() + provision_before["supplied_demands_within_before"] = provision_before[ + "supplied_demands_within" + ].copy() - provision_before[ - "us_demands_within_before" - ] = provision_before["demand"] - provision_before["supplyed_demands_within_before"] + provision_before["us_demands_within_before"] = ( + provision_before["demand"] + - provision_before["supplied_demands_within_before"] + ) - provision_before[ - "supplyed_demands_without_before" - ] = provision_before["supplyed_demands_without"] + provision_before["supplied_demands_without_before"] = ( + provision_before["supplied_demands_without"] + + provision_before["supplied_demands_within_before"] + ) - provision_before[ - "us_demands_without_before" - ] = provision_before["demand"] - provision_before["supplyed_demands_without_before"] + provision_before["us_demands_without_before"] = ( + provision_before["demand"] + - provision_before["supplied_demands_within_before"] + ) - provision_after["supplyed_demands_within_after"] = provision_after["supplyed_demands_within"].copy() + provision_after["supplied_demands_within_after"] = provision_after[ + "supplied_demands_within" + ].copy() - provision_after[ - "us_demands_within_after" - ] = provision_after["demand"] - provision_after["supplyed_demands_within_after"] + provision_after["us_demands_within_after"] = ( + provision_after["demand"] - provision_after["supplied_demands_within_after"] + ) - provision_after[ - "supplyed_demands_without_after" - ] = provision_after["supplyed_demands_without"].copy() + provision_after["supplied_demands_without_after"] = ( + provision_after["supplied_demands_within_after"] + + provision_after["supplied_demands_without"].copy() + ) - provision_after[ - "us_demands_without_after" - ] = provision_after["demand"] - provision_after["supplyed_demands_without_after"] + provision_after["us_demands_without_after"] = ( + provision_after["demand"] + - provision_after["supplied_demands_without_after"] + ) effects = provision_after.merge( - provision_before, - how="outer", - on=["building_id"] + provision_before, how="outer", on=["building_id"] ) effects["geometry"] = effects.apply( - lambda x: x["geometry_x"] if not pd.isna(x["geometry_x"]) else x["geometry_y"], - axis=1 + lambda x: ( + x["geometry_x"] if not pd.isna(x["geometry_x"]) else x["geometry_y"] + ), + axis=1, ) effects.drop(columns=["geometry_x", "geometry_y"], inplace=True) - effects["demand"] = effects["demand_x"].fillna(0) + effects["demand_y"].fillna(0) + effects["demand"] = effects["demand_x"].fillna(0) + effects["demand_y"].fillna( + 0 + ) effects.drop("is_project_y", axis=1, inplace=True) effects.rename(columns={"is_project_x": "is_project"}, inplace=True) effects = self._calculate_effects(effects) @@ -244,10 +255,12 @@ def estimate_effects( "index_scenario_project", "absolute_within", "demand", - "is_project" + "is_project", ] ] - effects = gpd.GeoDataFrame(effects, geometry="geometry", crs=provision_before.crs) + effects = gpd.GeoDataFrame( + effects, geometry="geometry", crs=provision_before.crs + ) return effects diff --git a/app/dependencies.py b/app/dependencies.py index 39a28a2..876acd2 100644 --- a/app/dependencies.py +++ b/app/dependencies.py @@ -1,30 +1,42 @@ import sys -from datetime import datetime from loguru import logger -from iduconfig import Config -from app.common.exceptions.http_exception_wrapper import http_exception from app.common.api_handler.api_handler import APIHandler - +from app.common.auth.service_auth import ( + build_service_auth, + build_service_token_verifier, +) +from app.common.config.config import Config +from app.common.exceptions.http_exception_wrapper import http_exception +from app.common.modules.effects_api_gateway import EffectsAPIGateway +from app.effects.effects_service import EffectsService +from app.provision.provision_service import ProvisionService logger.remove() logger.add(sys.stderr, level="INFO") log_level = "INFO" log_format = "{time:YYYY-MM-DD HH:mm:ss.SSS} | {level: <8} | {message}" -logger.add( - sys.stderr, - format=log_format, - level=log_level, - colorize=True -) +logger.add(sys.stderr, format=log_format, level=log_level, colorize=True) config = Config() +service_auth = build_service_auth(config) +service_token_verifier = build_service_token_verifier(config) logger.add( - f"{config.get('LOGS_FILE')}.log", + ".log", format=log_format, level="INFO", ) -urban_api_handler = APIHandler(config.get("URBAN_API")) +urban_api_handler = APIHandler(config.get("URBAN_API"), service_auth) +urban_api_mcp_handler = APIHandler(config.get("MCP_URBAN_API"), service_auth) + +effects_api_gateway = EffectsAPIGateway(urban_api_handler) +effects_api_mcp_gateway = EffectsAPIGateway(urban_api_mcp_handler) + +effects_service = EffectsService(effects_api_gateway) +effects_mcp_service = EffectsService(effects_api_mcp_gateway) + +provision_service = ProvisionService(effects_api_gateway) +provision_mcp_service = ProvisionService(effects_api_mcp_gateway) diff --git a/app/dto/__init__.py b/app/dto/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/dto/provision_dto.py b/app/dto/provision_dto.py new file mode 100644 index 0000000..5fc9375 --- /dev/null +++ b/app/dto/provision_dto.py @@ -0,0 +1,13 @@ +from pydantic import BaseModel, Field + + +class ProvisionDTO(BaseModel): + + project_id: int = Field(..., examples=[72], description="Project ID") + scenario_id: int = Field(..., examples=[192], description="Scenario ID") + service_type_id: int = Field(..., examples=[22], description="Service type ID") + target_population: int | None = Field( + default=None, + examples=[200], + description="Target population for project territory", + ) diff --git a/app/effects/dto/effects_dto.py b/app/effects/dto/effects_dto.py deleted file mode 100644 index 57052f5..0000000 --- a/app/effects/dto/effects_dto.py +++ /dev/null @@ -1,19 +0,0 @@ -from typing import Optional - -from pydantic import BaseModel, Field - - -class EffectsDTO(BaseModel): - - project_id: int = Field(..., examples=[72], description="Project ID") - scenario_id: int = Field(..., examples=[192], description="Scenario ID") - service_type_id: int = Field(..., examples=[7], description="Service type ID") - year: Optional[int] = Field( - default=2024, - examples=[2024], - description="Year for data retrieval") - target_population: Optional[int] = Field( - default=None, - examples=[200], - description="Target population for project territory" - ) diff --git a/app/effects/effects_controller.py b/app/effects/effects_controller.py index 065786b..bdc7d90 100644 --- a/app/effects/effects_controller.py +++ b/app/effects/effects_controller.py @@ -2,25 +2,19 @@ from fastapi import APIRouter, Depends +from app.common.auth.service_auth import get_current_user_id +from app.dependencies import effects_service +from app.dto.provision_dto import ProvisionDTO -from .dto.effects_dto import EffectsDTO from .shemas.effects_base_schema import EffectsSchema -from .effects_service import effects_service - effects_router = APIRouter(prefix="/effects") + @effects_router.get("/evaluate_provision", response_model=EffectsSchema) async def calculate_effects( - params: Annotated[EffectsDTO, Depends(EffectsDTO)], + params: Annotated[ProvisionDTO, Depends(ProvisionDTO)], + user_id: str = Depends(get_current_user_id), ) -> EffectsSchema: - """ - Get method for retrieving effects with objectnat - Params: - - project ID: Project ID - scenario ID: Scenario ID - """ - result = await effects_service.calculate_effects(params) - return EffectsSchema(**result) + return await effects_service.calculate_effects(params, user_id) diff --git a/app/effects/effects_mcp.py b/app/effects/effects_mcp.py new file mode 100644 index 0000000..5eab780 --- /dev/null +++ b/app/effects/effects_mcp.py @@ -0,0 +1,78 @@ +import traceback + +from fastmcp import FastMCP +from loguru import logger + +from app.common.auth.service_auth import get_mcp_user_id +from app.dependencies import effects_mcp_service, service_token_verifier +from app.dto.provision_dto import ProvisionDTO + +effects_mcp = FastMCP("Object Effects MCP server", auth=service_token_verifier) + + +@effects_mcp.tool( + name="CalculateObjectEffects", + title="Get provision effects for service", + description=""" + Retrieve service provision effects by service id for scenario id. + If total population is provided, demand is restored from it. Otherwise, population is restored from living square. + + Args to select: + - scenario_id (int): Scenario ID from Urban API to calculate effects for. + - service_type_id (int): Service type ID to calculate provision effects for. + - target_population (int, optional): Total population for demand calculation. If not provided, population is restored from living square. + + + Returns effects layers with estimated pivot info for llm analyses. + Response format: + { + "before_prove_data": { + "buildings": FeatureCollection, + "services": FeatureCollection, + "links": FeatureCollection + }, + "after_prove_data": { + "buildings": FeatureCollection, + "services": FeatureCollection, + "links": FeatureCollection + }, + "effects": FeatureCollection, + "pivot": { + "sum_absolute_total": int, + "average_absolute_total": float, + "median_absolute_total": int, + "average_index_total": float, + "median_index_total": int, + "sum_absolute_within": int, + "average_absolute_within": float, + "median_absolute_within": int, + }, + "text_pivot": str + } + """, +) +async def calc_provision_effects( + scenario_id: int, service_type_id: int, target_population: int | None = None +): + + try: + user_id = get_mcp_user_id() + project_id = await effects_mcp_service.gateway.get_project_id_by_scenario( + scenario_id, user_id + ) + effects_dto = ProvisionDTO( + project_id=project_id, + scenario_id=scenario_id, + service_type_id=service_type_id, + target_population=target_population, + ) + result = await effects_mcp_service.calculate_effects( + effects_dto, user_id, for_mcp=True + ) + return result + except Exception as e: + tb = traceback.format_exc() + logger.opt(exception=True).error( + f"Error in MCP tool 'CalculateObjectEffects': {type(e).__name__}: {e}" + ) + raise Exception(f"{type(e).__name__}: {e}\n\nTraceback:\n{tb}") from e diff --git a/app/effects/effects_service.py b/app/effects/effects_service.py index de44c66..7214bc2 100644 --- a/app/effects/effects_service.py +++ b/app/effects/effects_service.py @@ -1,18 +1,25 @@ -import json import asyncio +import json import geopandas as gpd import pandas as pd from loguru import logger -from app.dependencies import http_exception -from .dto.effects_dto import EffectsDTO -from .modules import ( - effects_api_gateway, - data_restorator, +from app.common.exceptions.http_exception_wrapper import http_exception +from app.common.modules import ( + ATTRIBUTES_MAP, + BUILDINGS_DROP_COLUMNS, + EFFECTS_MAP, + SERVICE_DROP_COLUMNS, + EffectsAPIGateway, attribute_parser, - matrix_builder, objectnat_calculator + data_restorator, + matrix_builder, + objectnat_calculator, ) +from app.dto.provision_dto import ProvisionDTO + +from .shemas.effects_base_schema import EffectsSchema class EffectsService: @@ -20,9 +27,12 @@ class EffectsService: Class for handling services calculation """ + def __init__(self, gateway: EffectsAPIGateway) -> None: + self.gateway = gateway + @staticmethod async def _get_pivot( - effects: pd.DataFrame | gpd.GeoDataFrame, + effects: pd.DataFrame | gpd.GeoDataFrame, ) -> dict[str, int | float]: """ Function creates a pivot table for effects data @@ -45,32 +55,44 @@ async def _get_pivot( if effects[effects["is_project"]].empty: return result - result["median_index_scenario_project"] = int(effects[effects["is_project"]]["index_scenario_project"].median()) - result["average_index_scenario_project"] = effects[effects["is_project"]]["index_scenario_project"].mean() + result["median_index_scenario_project"] = int( + effects[effects["is_project"]]["index_scenario_project"].median() + ) + result["average_index_scenario_project"] = effects[effects["is_project"]][ + "index_scenario_project" + ].mean() result["sum_absolute_scenario_project"] = int( effects[effects["is_project"]]["absolute_scenario_project"].sum() ) result["median_absolute_scenario_project"] = int( effects[effects["is_project"]]["absolute_scenario_project"].median() ) - result["average_absolute_scenario_project"] = effects[effects["is_project"]]["absolute_scenario_project"].mean() + result["average_absolute_scenario_project"] = effects[effects["is_project"]][ + "absolute_scenario_project" + ].mean() result["median_absolute_scenario_project"] = int( effects[effects["is_project"]]["absolute_scenario_project"].median() ) - result["average_index_scenario_project"] = effects[effects["is_project"]]["index_scenario_project"].mean() - result["median_index_scenario_project"] = int(effects[effects["is_project"]]["index_scenario_project"].median()) + result["average_index_scenario_project"] = effects[effects["is_project"]][ + "index_scenario_project" + ].mean() + result["median_index_scenario_project"] = int( + effects[effects["is_project"]]["index_scenario_project"].median() + ) return result + # ToDo Add population retrievement by year # ToDo Split function # ToDo Rewrite to context ids normal handling async def calculate_effects( - self, - effects_params: EffectsDTO - ) -> dict[str, dict]: + self, effects_params: ProvisionDTO, token: str, for_mcp: bool = False + ) -> EffectsSchema: """ Calculate provision effects by project data and target scenario Args: - effects_params (EffectsDTO): Project data + effects_params (ProvisionDTO): Project data + token (str): Authorization token + for_mcp (bool): If flag enabled adds string description for llm. Default to false. Returns: gpd.GeoDataFrame: Provision effects """ @@ -78,19 +100,29 @@ async def calculate_effects( logger.info( f"Started calculating effects for {effects_params.scenario_id} and service{effects_params.service_type_id}" ) - project_data = await effects_api_gateway.get_project_data( - effects_params.project_id + project_data = await self.gateway.get_project_data( + effects_params.project_id, token + ) + project_territory = await self.gateway.get_project_territory( + effects_params.project_id, token ) - normative_data = await effects_api_gateway.get_service_normative( + service_default_capacity = await self.gateway.get_default_capacity( + service_type_id=effects_params.service_type_id + ) + normative_data = await self.gateway.get_service_normative( territory_id=project_data["territory"]["id"], + context_ids=project_data["properties"]["context"], service_type_id=effects_params.service_type_id, - year=effects_params.year, + token=token, + ) + context_population = await self.gateway.get_context_population( + territory_ids_list=project_data["properties"]["context"], token=token ) - context_population = await effects_api_gateway.get_context_population( - territory_ids_list=project_data["properties"]["context"] + context_buildings = await self.gateway.get_project_context_buildings( + scenario_id=project_data["base_scenario"]["id"], token=token ) - context_buildings = await effects_api_gateway.get_project_context_buildings( - project_id=effects_params.project_id, + context_buildings.drop( + index=context_buildings.sjoin(project_territory).index, inplace=True ) context_buildings = await attribute_parser.parse_all_from_buildings( living_buildings=context_buildings, @@ -103,25 +135,31 @@ async def calculate_effects( target_population=context_population, ) context_buildings["is_project"] = False - context_services = await effects_api_gateway.get_project_context_services( - project_id=effects_params.project_id, + context_services = await self.gateway.get_project_context_services( + scenario_id=project_data["base_scenario"]["id"], service_type_id=effects_params.service_type_id, + token=token, ) if context_services.empty: + # ToDo Revise to another code raise http_exception( status_code=404, msg="No services of {service_type_id} type found in context", _input={"service_type_id": effects_params.service_type_id}, - _detail={} + _detail={}, ) context_services = await attribute_parser.parse_all_from_services( - services=context_services, + services=context_services, service_default_capacity=service_default_capacity ) - target_scenario_population = await effects_api_gateway.get_scenario_population_data( - scenario_id=effects_params.scenario_id, + target_scenario_population = await self.gateway.get_scenario_population_data( + scenario_id=effects_params.scenario_id, token=token ) - target_scenario_buildings = await effects_api_gateway.get_scenario_buildings( - scenario_id=effects_params.scenario_id + # User-provided population overrides the scenario population restored + # from Urban API (see the CalculateObjectEffects tool contract). + if effects_params.target_population: + target_scenario_population = effects_params.target_population + target_scenario_buildings = await self.gateway.get_scenario_buildings( + scenario_id=effects_params.scenario_id, token=token ) target_scenario_buildings = await attribute_parser.parse_all_from_buildings( living_buildings=target_scenario_buildings, @@ -134,51 +172,65 @@ async def calculate_effects( target_population=target_scenario_population, ) target_scenario_buildings["is_project"] = True - target_scenario_services = await effects_api_gateway.get_scenario_services( + target_scenario_services = await self.gateway.get_scenario_services( scenario_id=effects_params.scenario_id, service_type_id=effects_params.service_type_id, + token=token, ) target_scenario_services = await attribute_parser.parse_all_from_services( services=target_scenario_services, + service_default_capacity=service_default_capacity, ) - base_scenario_buildings = await effects_api_gateway.get_scenario_buildings( - scenario_id=project_data["base_scenario"]["id"] + base_scenario_buildings = await self.gateway.get_scenario_buildings( + scenario_id=project_data["base_scenario"]["id"], token=token ) base_scenario_buildings = await attribute_parser.parse_all_from_buildings( living_buildings=base_scenario_buildings, ) + base_scenario_buildings = await asyncio.to_thread( + data_restorator.restore_demands, + buildings=base_scenario_buildings, + service_normative=normative_data["services_capacity_per_1000_normative"], + service_normative_type=normative_data["capacity_type"], + ) base_scenario_buildings["is_project"] = True - base_scenario_services = await effects_api_gateway.get_scenario_services( + base_scenario_services = await self.gateway.get_scenario_services( scenario_id=project_data["base_scenario"]["id"], service_type_id=effects_params.service_type_id, + token=token, ) base_scenario_services = await attribute_parser.parse_all_from_services( services=base_scenario_services, + service_default_capacity=service_default_capacity, ) after_buildings = await asyncio.to_thread( - pd.concat, - objs=[context_buildings, target_scenario_buildings] + pd.concat, objs=[context_buildings, target_scenario_buildings] ) after_services = await asyncio.to_thread( - pd.concat, - objs=[context_services, target_scenario_services] + pd.concat, objs=[context_services, target_scenario_services] ) - before_buildings = await asyncio.to_thread( + before_buildings = await asyncio.to_thread( pd.concat, objs=[context_buildings, base_scenario_buildings], ) before_services = await asyncio.to_thread( - pd.concat, - objs=[context_services, base_scenario_services] + pd.concat, objs=[context_services, base_scenario_services] ) after_buildings.sort_values("is_project", ascending=False, inplace=True) after_buildings.drop_duplicates("building_id", keep="first", inplace=True) after_buildings.set_index("building_id", inplace=True) after_services.set_index("service_id", inplace=True) + after_services = after_services[ + ~after_services.index.duplicated(keep="first") + ].copy() before_buildings.sort_values("is_project", ascending=False, inplace=True) before_buildings.drop_duplicates("building_id", keep="first", inplace=True) before_buildings.set_index("building_id", inplace=True) before_services.set_index("service_id", inplace=True) + before_services.drop_duplicates("geometry", inplace=True) + before_services = before_services[ + ~before_services.index.duplicated(keep="first") + ].copy() if target_scenario_buildings.empty: local_crs = context_buildings.estimate_utm_crs() else: @@ -187,6 +239,8 @@ async def calculate_effects( before_services.to_crs(local_crs, inplace=True) after_buildings.to_crs(local_crs, inplace=True) after_services.to_crs(local_crs, inplace=True) + # ToDo context - project objects relation should be revised + after_services.drop_duplicates("geometry", inplace=True) before_matrix = await asyncio.to_thread( matrix_builder.calculate_availability_matrix, buildings=before_buildings, @@ -201,17 +255,20 @@ async def calculate_effects( normative_value=normative_data["normative_value"], normative_type=normative_data["normative_type"], ) + before_services["capacity"] = before_services["capacity"].fillna( + before_services["capacity"].mean() + ) before_prove_data = await asyncio.to_thread( objectnat_calculator.evaluate_provision, buildings=before_buildings, - services=before_services, + services=before_services[~before_services.index.duplicated(keep="first")], matrix=before_matrix, service_normative=normative_data["normative_value"], ) after_prove_data = await asyncio.to_thread( objectnat_calculator.evaluate_provision, buildings=after_buildings, - services=after_services, + services=after_services[~after_services.index.duplicated(keep="first")], matrix=after_matrix, service_normative=normative_data["normative_value"], ) @@ -228,24 +285,548 @@ async def calculate_effects( result = { "before_prove_data": { "buildings": json.loads( - before_prove_data["buildings"].to_crs(4326).to_json() + before_prove_data["buildings"] + .rename( + columns={ + k: v + for k, v in ATTRIBUTES_MAP.items() + if k in before_prove_data["buildings"].columns + } + ) + .drop(columns=BUILDINGS_DROP_COLUMNS) + .to_crs(4326) + .to_json() ), "services": json.loads( - before_prove_data["services"].to_crs(4326).to_json() + before_prove_data["services"] + .rename( + columns={ + k: v + for k, v in ATTRIBUTES_MAP.items() + if k in before_prove_data["services"].columns + } + ) + .drop(columns=SERVICE_DROP_COLUMNS) + .to_crs(4326) + .to_json() + ), + "links": json.loads( + before_prove_data["links"] + .rename( + columns={ + k: v + for k, v in ATTRIBUTES_MAP.items() + if k in before_prove_data["links"].columns + } + ) + .to_crs(4326) + .to_json() ), - "links": json.loads(before_prove_data["links"].to_crs(4326).to_json()), }, "after_prove_data": { "buildings": json.loads( - after_prove_data["buildings"].to_crs(4326).to_json() + after_prove_data["buildings"] + .rename( + columns={ + k: v + for k, v in ATTRIBUTES_MAP.items() + if k in after_prove_data["buildings"].columns + } + ) + .drop(columns=BUILDINGS_DROP_COLUMNS) + .to_crs(4326) + .to_json() + ), + "services": json.loads( + after_prove_data["services"] + .rename( + columns={ + k: v + for k, v in ATTRIBUTES_MAP.items() + if k in after_prove_data["services"].columns + } + ) + .drop(columns=SERVICE_DROP_COLUMNS) + .to_crs(4326) + .to_json() + ), + "links": json.loads( + after_prove_data["links"] + .rename( + columns={ + k: v + for k, v in ATTRIBUTES_MAP.items() + if k in after_prove_data["links"].columns + } + ) + .to_crs(4326) + .to_json() ), - "services": json.loads(after_prove_data["services"].to_crs(4326).to_json()), - "links": json.loads(after_prove_data["links"].to_crs(4326).to_json()), }, - "effects": json.loads(effects.to_crs(4326).to_json()), + "effects": json.loads( + effects.rename(columns=EFFECTS_MAP).to_crs(4326).to_json() + ), "pivot": pivot, } - return result + if for_mcp: + result["text_pivot"] = await self.form_llm_context( + before_prove_data["buildings"], + after_prove_data["buildings"], + before_prove_data["services"], + after_prove_data["services"], + ) + return EffectsSchema(**result) + + @staticmethod + async def form_llm_context( + before_buildings: gpd.GeoDataFrame, + after_buildings: gpd.GeoDataFrame, + before_services: gpd.GeoDataFrame, + after_services: gpd.GeoDataFrame, + ) -> str: + """ + Function forms text repr stats from calculated provision data for llm. + Args: + before_buildings (gpd.GeoDataFrame): Buildings provision layers before. + after_buildings (gpd.GeoDataFrame): Buildings provision layers after. + before_services (gpd.GeoDataFrame): Services provision layers before. + after_services (gpd.GeoDataFrame): Services provision layers after. + Returns: + str: Text representation for formed stats in json string. + """ + + before_buildings_all = before_buildings.rename( + columns={ + k: v for k, v in ATTRIBUTES_MAP.items() if k in before_buildings.columns + } + ) + after_buildings_all = after_buildings.rename( + columns={ + k: v for k, v in ATTRIBUTES_MAP.items() if k in after_buildings.columns + } + ) + before_services_all = before_services.rename( + columns={ + k: v for k, v in ATTRIBUTES_MAP.items() if k in before_services.columns + } + ) + after_services_all = after_services.rename( + columns={ + k: v for k, v in ATTRIBUTES_MAP.items() if k in after_services.columns + } + ) + + before_buildings_context = before_buildings[ + before_buildings["is_scenario_object"] == False + ].rename( + columns={ + k: v for k, v in ATTRIBUTES_MAP.items() if k in before_buildings.columns + } + ) + after_buildings_context = after_buildings[ + after_buildings["is_scenario_object"] == False + ].rename( + columns={ + k: v for k, v in ATTRIBUTES_MAP.items() if k in after_buildings.columns + } + ) + before_services_context = before_services[ + before_services["is_scenario_object"] == False + ].rename( + columns={ + k: v for k, v in ATTRIBUTES_MAP.items() if k in before_services.columns + } + ) + after_services_context = after_services[ + after_services["is_scenario_object"] == False + ].rename( + columns={ + k: v for k, v in ATTRIBUTES_MAP.items() if k in after_services.columns + } + ) + before_buildings_project = before_buildings[ + before_buildings["is_scenario_object"] == True + ].rename( + columns={ + k: v for k, v in ATTRIBUTES_MAP.items() if k in before_buildings.columns + } + ) + after_buildings_project = after_buildings[ + after_buildings["is_scenario_object"] == True + ].rename( + columns={ + k: v for k, v in ATTRIBUTES_MAP.items() if k in after_buildings.columns + } + ) + before_services_project = before_services[ + before_services["is_scenario_object"] == True + ].rename( + columns={ + k: v for k, v in ATTRIBUTES_MAP.items() if k in before_services.columns + } + ) + after_services_project = after_services[ + after_services["is_scenario_object"] == True + ].rename( + columns={ + k: v for k, v in ATTRIBUTES_MAP.items() if k in after_services.columns + } + ) + + before_buildings.rename( + columns={ + k: v for k, v in ATTRIBUTES_MAP.items() if k in before_buildings.columns + }, + inplace=True, + ) + after_buildings.rename( + columns={ + k: v for k, v in ATTRIBUTES_MAP.items() if k in after_buildings.columns + }, + inplace=True, + ) + before_services.rename( + columns={ + k: v for k, v in ATTRIBUTES_MAP.items() if k in before_services.columns + }, + inplace=True, + ) + after_services.rename( + columns={ + k: v for k, v in ATTRIBUTES_MAP.items() if k in after_services.columns + }, + inplace=True, + ) + all_provision_before = int( + before_buildings_all[ + "Удовлетворённый спрос вне нормативной доступности (до) (чел)" + ].sum() + ) + all_provision_after = int( + after_buildings_all[ + "Удовлетворённый спрос вне нормативной доступности (после) (чел)" + ].sum() + ) + all_provision_within_before = int( + before_buildings_all[ + "Удовлетворённый спрос в нормативной доступности (до) (чел)" + ].sum() + ) + all_provision_within_after = int( + after_buildings_all[ + "Удовлетворённый спрос в нормативной доступности (после) (чел)" + ].sum() + ) + all_provision_without_before = int( + before_buildings_all[ + "Удовлетворённый спрос вне нормативной доступности (до) (чел)" + ].sum() + ) + all_provision_without_after = int( + after_buildings_all[ + "Удовлетворённый спрос вне нормативной доступности (после) (чел)" + ].sum() + ) + all_total_capacity_before = int(before_services_all["Вместимость (чел)"].sum()) + all_total_capacity_after = int(after_services_all["Вместимость (чел)"].sum()) + all_demand_before = int(before_buildings_all["Спрос (чел)"].sum()) + all_demand_after = int(after_buildings_all["Спрос (чел)"].sum()) + all_unmet_demand_before = int( + before_buildings_all["Неудовлетворённый спрос (чел)"].sum() + ) + all_unmet_demand_after = int( + after_buildings_all["Неудовлетворённый спрос (чел)"].sum() + ) + all_unmet_demand_within_before = int( + before_buildings_all[ + "Неудовлетворённый спрос в нормативной доступности (до) (чел)" + ].sum() + ) + all_unmet_demand_within_after = int( + after_buildings_all[ + "Неудовлетворённый спрос в нормативной доступности (после) (чел)" + ].sum() + ) + all_unmet_demand_without_before = int( + before_buildings_all[ + "Неудовлетворённый спрос вне нормативной доступности (до) (чел)" + ].sum() + ) + all_unmet_demand_without_after = int( + after_buildings_all[ + "Неудовлетворённый спрос вне нормативной доступности (после) (чел)" + ].sum() + ) + all_balance_before = all_total_capacity_before - all_demand_before + all_balance_after = all_total_capacity_after - all_demand_after + all_deficit_before = min(0, all_balance_before) + all_deficit_after = min(0, all_balance_after) + all_surplus_before = max(0, all_balance_before) + all_surplus_after = max(0, all_balance_after) + context_provision_before = int( + before_buildings_context[ + "Удовлетворённый спрос вне нормативной доступности (до) (чел)" + ].sum() + ) + context_provision_after = int( + after_buildings_context[ + "Удовлетворённый спрос вне нормативной доступности (после) (чел)" + ].sum() + ) + context_provision_within_before = int( + before_buildings_context[ + "Удовлетворённый спрос в нормативной доступности (до) (чел)" + ].sum() + ) + context_provision_within_after = int( + after_buildings_context[ + "Удовлетворённый спрос в нормативной доступности (после) (чел)" + ].sum() + ) + context_provision_without_before = int( + before_buildings_context[ + "Удовлетворённый спрос вне нормативной доступности (до) (чел)" + ].sum() + ) + context_provision_without_after = int( + after_buildings_context[ + "Удовлетворённый спрос вне нормативной доступности (после) (чел)" + ].sum() + ) + context_total_capacity_before = int( + before_services_context["Вместимость (чел)"].sum() + ) + context_total_capacity_after = int( + after_services_context["Вместимость (чел)"].sum() + ) + context_demand_before = int(before_buildings_context["Спрос (чел)"].sum()) + context_demand_after = int(after_buildings_context["Спрос (чел)"].sum()) + context_unmet_demand_before = int( + before_buildings_context["Неудовлетворённый спрос (чел)"].sum() + ) + context_unmet_demand_after = int( + after_buildings_context["Неудовлетворённый спрос (чел)"].sum() + ) + context_unmet_demand_within_before = int( + before_buildings_context[ + "Неудовлетворённый спрос в нормативной доступности (до) (чел)" + ].sum() + ) + context_unmet_demand_within_after = int( + after_buildings_context[ + "Неудовлетворённый спрос в нормативной доступности (после) (чел)" + ].sum() + ) + context_unmet_demand_without_before = int( + before_buildings_context[ + "Неудовлетворённый спрос вне нормативной доступности (до) (чел)" + ].sum() + ) + context_unmet_demand_without_after = int( + after_buildings_context[ + "Неудовлетворённый спрос вне нормативной доступности (после) (чел)" + ].sum() + ) + context_balance_before = context_total_capacity_before - context_demand_before + context_balance_after = context_total_capacity_after - context_demand_after + context_deficit_before = min(0, context_balance_before) + context_deficit_after = min(0, context_balance_after) + context_surplus_before = max(0, context_balance_before) + context_surplus_after = max(0, context_balance_after) + project_provision_before = int( + before_buildings_project[ + "Удовлетворённый спрос вне нормативной доступности (до) (чел)" + ].sum() + ) + project_provision_after = int( + after_buildings_project[ + "Удовлетворённый спрос вне нормативной доступности (после) (чел)" + ].sum() + ) + project_provision_within_before = int( + before_buildings_project[ + "Удовлетворённый спрос в нормативной доступности (до) (чел)" + ].sum() + ) + project_provision_within_after = int( + after_buildings_project[ + "Удовлетворённый спрос в нормативной доступности (после) (чел)" + ].sum() + ) + project_provision_without_before = int( + before_buildings_project[ + "Удовлетворённый спрос вне нормативной доступности (до) (чел)" + ].sum() + ) + project_provision_without_after = int( + after_buildings_project[ + "Удовлетворённый спрос вне нормативной доступности (после) (чел)" + ].sum() + ) + project_total_capacity_before = int( + before_services_project["Вместимость (чел)"].sum() + ) + project_total_capacity_after = int( + after_services_project["Вместимость (чел)"].sum() + ) + project_demand_before = int(before_buildings_project["Спрос (чел)"].sum()) + project_demand_after = int(after_buildings_project["Спрос (чел)"].sum()) + project_unmet_demand_before = int( + before_buildings_project["Неудовлетворённый спрос (чел)"].sum() + ) + project_unmet_demand_after = int( + after_buildings_project["Неудовлетворённый спрос (чел)"].sum() + ) + project_unmet_demand_within_before = int( + before_buildings_project[ + "Неудовлетворённый спрос в нормативной доступности (до) (чел)" + ].sum() + ) + project_unmet_demand_within_after = int( + after_buildings_project[ + "Неудовлетворённый спрос в нормативной доступности (после) (чел)" + ].sum() + ) + project_unmet_demand_without_before = int( + before_buildings_project[ + "Неудовлетворённый спрос вне нормативной доступности (до) (чел)" + ].sum() + ) + project_unmet_demand_without_after = int( + after_buildings_project[ + "Неудовлетворённый спрос вне нормативной доступности (после) (чел)" + ].sum() + ) + project_balance_before = project_total_capacity_before - project_demand_before + project_balance_after = project_total_capacity_after - project_demand_after + project_deficit_before = min(0, project_balance_before) + project_deficit_after = min(0, project_balance_after) + project_surplus_before = max(0, project_balance_before) + project_surplus_after = max(0, project_balance_after) -effects_service = EffectsService() + result = { + "all": { + "provision_before": all_provision_before, + "provision_after": all_provision_after, + "provision_delta": all_provision_after - all_provision_before, + "provision_within_before": all_provision_within_before, + "provision_within_after": all_provision_within_after, + "provision_within_delta": all_provision_within_after + - all_provision_within_before, + "provision_without_before": all_provision_without_before, + "provision_without_after": all_provision_without_after, + "provision_without_delta": all_provision_without_after + - all_provision_without_before, + "total_capacity_before": all_total_capacity_before, + "total_capacity_after": all_total_capacity_after, + "total_capacity_delta": all_total_capacity_after + - all_total_capacity_before, + "balance_before": all_balance_before, + "balance_after": all_balance_after, + "balance_delta": all_balance_after - all_balance_before, + "deficit_before": all_deficit_before, + "deficit_after": all_deficit_after, + "deficit_delta": all_deficit_after - all_deficit_before, + "surplus_before": all_surplus_before, + "surplus_after": all_surplus_after, + "surplus_delta": all_surplus_after - all_surplus_before, + "demand_before": all_demand_before, + "demand_after": all_demand_after, + "demand_delta": all_demand_after - all_demand_before, + "unmet_demand_before": all_unmet_demand_before, + "unmet_demand_after": all_unmet_demand_after, + "unmet_demand_delta": all_unmet_demand_after - all_unmet_demand_before, + "unmet_demand_within_before": all_unmet_demand_within_before, + "unmet_demand_within_after": all_unmet_demand_within_after, + "unmet_demand_within_delta": all_unmet_demand_within_after + - all_unmet_demand_within_before, + "unmet_demand_without_before": all_unmet_demand_without_before, + "unmet_demand_without_after": all_unmet_demand_without_after, + "unmet_demand_without_delta": all_unmet_demand_without_after + - all_unmet_demand_without_before, + }, + "context": { + "provision_before": context_provision_before, + "provision_after": context_provision_after, + "provision_delta": context_provision_after - context_provision_before, + "provision_within_before": context_provision_within_before, + "provision_within_after": context_provision_within_after, + "provision_within_delta": context_provision_within_after + - context_provision_within_before, + "provision_without_before": context_provision_without_before, + "provision_without_after": context_provision_without_after, + "provision_without_delta": context_provision_without_after + - context_provision_without_before, + "total_capacity_before": context_total_capacity_before, + "total_capacity_after": context_total_capacity_after, + "total_capacity_delta": context_total_capacity_after + - context_total_capacity_before, + "balance_before": context_balance_before, + "balance_after": context_balance_after, + "balance_delta": context_balance_after - context_balance_before, + "deficit_before": context_deficit_before, + "deficit_after": context_deficit_after, + "deficit_delta": context_deficit_after - context_deficit_before, + "surplus_before": context_surplus_before, + "surplus_after": context_surplus_after, + "surplus_delta": context_surplus_after - context_surplus_before, + "demand_before": context_demand_before, + "demand_after": context_demand_after, + "demand_delta": context_demand_after - context_demand_before, + "unmet_demand_before": context_unmet_demand_before, + "unmet_demand_after": context_unmet_demand_after, + "unmet_demand_delta": context_unmet_demand_after + - context_unmet_demand_before, + "unmet_demand_within_before": context_unmet_demand_within_before, + "unmet_demand_within_after": context_unmet_demand_within_after, + "unmet_demand_within_delta": context_unmet_demand_within_after + - context_unmet_demand_within_before, + "unmet_demand_without_before": context_unmet_demand_without_before, + "unmet_demand_without_after": context_unmet_demand_without_after, + "unmet_demand_without_delta": context_unmet_demand_without_after + - context_unmet_demand_without_before, + }, + "project": { + "provision_before": project_provision_before, + "provision_after": project_provision_after, + "provision_delta": project_provision_after - project_provision_before, + "provision_within_before": project_provision_within_before, + "provision_within_after": project_provision_within_after, + "provision_within_delta": project_provision_within_after + - project_provision_within_before, + "provision_without_before": project_provision_without_before, + "provision_without_after": project_provision_without_after, + "provision_without_delta": project_provision_without_after + - project_provision_without_before, + "total_capacity_before": project_total_capacity_before, + "total_capacity_after": project_total_capacity_after, + "total_capacity_delta": project_total_capacity_after + - project_total_capacity_before, + "balance_before": project_balance_before, + "balance_after": project_balance_after, + "balance_delta": project_balance_after - project_balance_before, + "deficit_before": project_deficit_before, + "deficit_after": project_deficit_after, + "deficit_delta": project_deficit_after - project_deficit_before, + "surplus_before": project_surplus_before, + "surplus_after": project_surplus_after, + "surplus_delta": project_surplus_after - project_surplus_before, + "demand_before": project_demand_before, + "demand_after": project_demand_after, + "demand_delta": project_demand_after - project_demand_before, + "unmet_demand_before": project_unmet_demand_before, + "unmet_demand_after": project_unmet_demand_after, + "unmet_demand_delta": project_unmet_demand_after + - project_unmet_demand_before, + "unmet_demand_within_before": project_unmet_demand_within_before, + "unmet_demand_within_after": project_unmet_demand_within_after, + "unmet_demand_within_delta": project_unmet_demand_within_after + - project_unmet_demand_within_before, + "unmet_demand_without_before": project_unmet_demand_without_before, + "unmet_demand_without_after": project_unmet_demand_without_after, + "unmet_demand_without_delta": project_unmet_demand_without_after + - project_unmet_demand_without_before, + }, + } + return json.dumps(result) diff --git a/app/effects/modules/__init__.py b/app/effects/modules/__init__.py deleted file mode 100644 index cb0dff8..0000000 --- a/app/effects/modules/__init__.py +++ /dev/null @@ -1,5 +0,0 @@ -from .attribute_parser import attribute_parser -from .effects_api_gateway import effects_api_gateway -from .data_restorator import data_restorator -from .matrix_builder import matrix_builder -from .objectnat_calculator import objectnat_calculator \ No newline at end of file diff --git a/app/effects/modules/effects_api_gateway.py b/app/effects/modules/effects_api_gateway.py deleted file mode 100644 index 34568ab..0000000 --- a/app/effects/modules/effects_api_gateway.py +++ /dev/null @@ -1,235 +0,0 @@ -import asyncio - -import geopandas as gpd - -from app.dependencies import urban_api_handler, http_exception - - -class EffectsAPIGateway: - - @staticmethod - async def get_service_normative( - territory_id: int, - service_type_id: int, - year: int = 2024, - ) -> dict[str, int | str]: - """ - Function retrieves normative data from urban_api - Args: - territory_id: territory id to get normative from - service_type_id: service to get normative from - year: year to get normative from - Returns: - dict[str, int | str]: normative data with normative value and normative type (Literal["time", "dist"]) - Raises: - 400, http exception id not found - """ - - response = await urban_api_handler.get( - f"/api/v1/territory/{territory_id}/normatives", - params={ - "year": year, - } - ) - for service_type in response: - if service_type["service_type"]["id"] == service_type_id: - if normative_value:=service_type["radius_availability_meters"]: - service_type["normative_value"] = normative_value - service_type["normative_type"] = "dist" - if service_type.get("services_per_1000_normative"): - service_type["capacity_type"] = "unit" - else: - service_type["capacity_type"] = "capacity" - return service_type - elif normative_value:=service_type["time_availability_minutes"]: - service_type["normative_value"] = normative_value - service_type["normative_type"] = "time" - if service_type.get("services_per_1000_normative"): - service_type["capacity_type"] = "unit" - else: - service_type["capacity_type"] = "capacity" - return service_type - else: - raise http_exception( - status_code=400, - msg="Service type normative not found", - _input={"service_type_id": service_type_id}, - _detail={ - "Available service ids": [service_type["id"] for service_type in response] - }, - ) - raise http_exception( - status_code=400, - msg="Service type normative not found", - _input={"service_type_id": service_type_id}, - _detail={ - "Available service ids": [service_type["service_type"]["id"] for service_type in response] - } - ) - - @staticmethod - async def get_project_data(project_id: int) -> dict[str, int | dict]: - """ - Function retrieves project territory data from urban_api - Args: - project_id: project id to get territory from - Returns: - dict with "geometry" field as dict with "type" and "coordinates" fields and field "base_scenario_id" - """ - - response = await urban_api_handler.get( - endpoint_url=f"/api/v1/projects/{project_id}", - ) - - return response - - @staticmethod - async def get_scenario_buildings( - scenario_id: int, - ) -> gpd.GeoDataFrame: - """ - Function retrieves scenario buildings data from urban_api - Args: - scenario_id: scenario id to get buildings from - Returns: - gpd.GeoDataFrame: buildings layer, can be empty - """ - - buildings = await urban_api_handler.get( - endpoint_url=f"/api/v1/scenarios/{scenario_id}/geometries_with_all_objects", - params={ - "physical_object_type_id": 4 - } - ) - buildings_gdf = gpd.GeoDataFrame.from_features(buildings) - if buildings_gdf.empty: - return buildings_gdf - buildings_gdf.set_crs(4326, inplace=True) - return buildings_gdf - - @staticmethod - async def get_project_context_buildings( - project_id: int, - ) -> gpd.GeoDataFrame: - """ - Function retrieves scenario context buildings data from urban_api - Args: - project_id: scenario id to get buildings from - Returns: - gpd.GeoDataFrame: buildings layer - Raises: - 404, http exception living buildings not found - """ - - context_buildings = await urban_api_handler.get( - endpoint_url=f"/api/v1/projects/{project_id}/context/geometries_with_all_objects", - params={ - "physical_object_type_id": 4, - } - ) - context_buildings_gdf = gpd.GeoDataFrame.from_features(context_buildings) - if context_buildings_gdf.empty: - return context_buildings_gdf - context_buildings_gdf.set_crs(4326, inplace=True) - return context_buildings_gdf - - @staticmethod - async def get_scenario_services( - scenario_id: int, - service_type_id: int, - ) -> gpd.GeoDataFrame: - """ - Function retrieves scenario services data from urban_api - Args: - scenario_id: scenario id to get services from - service_type_id: service to get services from - Returns: - gpd.GeoDataFrame: services layer, can be empty - """ - - services = await urban_api_handler.get( - endpoint_url=f"/api/v1/scenarios/{scenario_id}/geometries_with_all_objects", - params={ - "service_type_id": service_type_id, - } - ) - services_gdf = gpd.GeoDataFrame.from_features(services) - if services_gdf.empty: - return services_gdf - services_gdf.set_crs(4326, inplace=True) - return services_gdf - - @staticmethod - async def get_project_context_services( - project_id: int, - service_type_id: int, - ) -> gpd.GeoDataFrame: - """ - Function retrieves scenario context services data from urban_api - Args: - project_id: scenario id to get services from - service_type_id: service to get services from - Returns: - gpd.GeoDataFrame: context services layer. Can be empty - """ - - context_services = await urban_api_handler.get( - endpoint_url=f"/api/v1/projects/{project_id}/context/geometries_with_all_objects", - params={ - "service_type_id": service_type_id, - } - ) - context_services_gdf = gpd.GeoDataFrame.from_features(context_services) - if context_services_gdf.empty: - return context_services_gdf - context_services_gdf.set_crs(4326, inplace=True) - return context_services_gdf - - @staticmethod - async def get_scenario_population_data( - scenario_id: int | None - ) -> int: - """ - Function retrieves population data from urban_api - Args: - scenario_id: scenario id to get population data from - Returns: - int | none: population data layer, if < 1 returns None - """ - - population = await urban_api_handler.get( - endpoint_url=f"/api/v1/scenarios/{scenario_id}/indicators_values", - params={ - "indicators_ids": 1, - } - ) - - if (value:=population[0]["value"]) < 1: - return None - return value - - @staticmethod - async def get_context_population( - territory_ids_list: list[int], - ) -> int: - """ - Function retrieves territory population data from urban_api by territory id - Args: - territory_ids_list: list[int]: territory ids list to get population data from - Returns: - gpd.GeoDataFrame: territory population data layer - """ - - task_list = [urban_api_handler.get( - endpoint_url=f"/api/v1/territory/{territory_id}/indicator_values", - params={ - "indicator_ids": 1 - } - ) for territory_id in territory_ids_list] - - result = await asyncio.gather(*task_list) - return sum([item[0]["value"] for item in result]) - - - -effects_api_gateway = EffectsAPIGateway() diff --git a/app/effects/shemas/effects_base_schema.py b/app/effects/shemas/effects_base_schema.py index 81dedeb..2029720 100644 --- a/app/effects/shemas/effects_base_schema.py +++ b/app/effects/shemas/effects_base_schema.py @@ -1,49 +1,25 @@ -from typing import Literal, Optional, Any +from typing import Optional from pydantic import BaseModel +from app.schemas.provision_base_schema import FeatureCollectionSchema, ProvisionSchema -class GeometrySchema(BaseModel): - - type: Literal["Polygon", "MultiPolygon", "LineString", "MultiLineString", "Point", "MultiPoint"] - coordinates: list[Any] - - -class FeatureSchema(BaseModel): - - id: Optional[int | None] - type: Literal["Feature"] - geometry: GeometrySchema - properties: dict - - -class FeatureCollectionSchema(BaseModel): - - type: Literal["FeatureCollection"] - features: list[FeatureSchema] - - -class ProvisionSchema(BaseModel): - - buildings: FeatureCollectionSchema - services: FeatureCollectionSchema - links: FeatureCollectionSchema class PivotSchema(BaseModel): - sum_absolute_total: int - average_absolute_total: int | float - median_absolute_total: int - average_index_total: int | float - median_index_total: int - sum_absolute_scenario_project: Optional[int] = None - average_absolute_scenario_project: Optional[int | float] = None - median_absolute_scenario_project: Optional[int] = None - average_index_scenario_project: Optional[int | float] = None - median_index_scenario_project: Optional[int] = None - sum_absolute_within: int - average_absolute_within: int | float - median_absolute_within: int + sum_absolute_total: int + average_absolute_total: int | float + median_absolute_total: int + average_index_total: int | float + median_index_total: int + sum_absolute_scenario_project: Optional[int] = None + average_absolute_scenario_project: Optional[int | float] = None + median_absolute_scenario_project: Optional[int] = None + average_index_scenario_project: Optional[int | float] = None + median_index_scenario_project: Optional[int] = None + sum_absolute_within: int + average_absolute_within: int | float + median_absolute_within: int class EffectsSchema(BaseModel): @@ -52,3 +28,4 @@ class EffectsSchema(BaseModel): after_prove_data: ProvisionSchema effects: FeatureCollectionSchema pivot: PivotSchema + text_pivot: str | None = None diff --git a/app/main.py b/app/main.py index 4194241..1fe9e09 100644 --- a/app/main.py +++ b/app/main.py @@ -1,18 +1,64 @@ -import aiofiles +from contextlib import asynccontextmanager + from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware -from fastapi.responses import RedirectResponse +from fastapi.responses import FileResponse, RedirectResponse +from fastmcp.utilities.lifespan import combine_lifespans +from loguru import logger -from .dependencies import config +from .__version__ import APP_VERSION +from .common.middlewares.exception_handler import ExceptionHandlerMiddleware +from .common.middlewares.prometheus_handler import ObservabilityMiddleware +from .dependencies import config, http_exception, service_auth from .effects.effects_controller import effects_router +from .mcp import effects_mcp_app, provision_mcp_app +from .observability import OpenTelemetryAgent, PrometheusConfig +from .observability.metrics import setup_metrics +from .provision.provision_controller import provision_router + +log_format = "{time:YYYY-MM-DD HH:mm:ss.SSS} | {level: <8} | {message}" + +logger.add( + ".log", + format=log_format, + level="INFO", +) + +metrics = setup_metrics() + + +@asynccontextmanager +async def lifespan(app: FastAPI): + otel_agent = OpenTelemetryAgent( + prometheus_config=PrometheusConfig( + host="0.0.0.0", + port=int(config.get("PROMETHEUS_PORT")), + ), + ) + setup_metrics() + logger.info(f"Prometheus server started on {config.get('PROMETHEUS_PORT')}") + async with service_auth: + await service_auth.get_access_token() + yield + otel_agent.shutdown() + logger.info("Prometheus server was shut down") app = FastAPI( title="ObjectNat effects API", description="API for calculating effects for territory by ObjectNat library", - version=config.get("APP_VERSION"), + version=APP_VERSION, + lifespan=combine_lifespans( + lifespan, effects_mcp_app.lifespan, provision_mcp_app.lifespan + ), ) +app.include_router(effects_router) +app.include_router(provision_router) + +app.mount("/effects/mcp", effects_mcp_app) +app.mount("/provision/mcp", provision_mcp_app) + # Add CORS middleware app.add_middleware( CORSMiddleware, @@ -21,20 +67,43 @@ allow_methods=["*"], allow_headers=["*"], ) +app.add_middleware(ExceptionHandlerMiddleware, metrics=metrics) +app.add_middleware(ObservabilityMiddleware, metrics=metrics) + @app.get("/", response_model=dict[str, str]) def read_root(): - return RedirectResponse(url='/docs') + return RedirectResponse(url="/docs") + @app.get("/status") async def read_root(): return {"status": "OK"} -@app.get("/logs") -async def read_logs(): - async with aiofiles.open(config.get("LOGS_FILE")) as logs_file: - logs = await logs_file.read() - return logs[-1:-10000] +@app.get("/logs") +async def get_logs(): + """ + Get logs file from app + """ -app.include_router(effects_router) \ No newline at end of file + try: + return FileResponse( + ".log", + media_type="application/octet-stream", + filename=f"ObjectEffects.log", + ) + except FileNotFoundError as e: + raise http_exception( + status_code=404, + msg="Log file not found", + _input={"lof_file_name": ".log"}, + _detail={"error": e.__str__()}, + ) + except Exception as e: + raise http_exception( + status_code=500, + msg="Internal server error during reading logs", + _input={"log_file_name": ".log"}, + _detail={"error": e.__str__()}, + ) diff --git a/app/mcp.py b/app/mcp.py new file mode 100644 index 0000000..6c10dd7 --- /dev/null +++ b/app/mcp.py @@ -0,0 +1,10 @@ +from app.effects.effects_mcp import effects_mcp +from app.provision.provision_mcp import provision_mcp + +# Provision tools are also exposed on the effects MCP endpoint so that +# consumers (gMART agents, ChatStorage replay) reach every tool via the +# single OBJECTS_EFFECTS_MCP_SERVER URL. +effects_mcp.mount(provision_mcp) + +effects_mcp_app = effects_mcp.http_app(path="/") +provision_mcp_app = provision_mcp.http_app(path="/") diff --git a/app/observability/__init__.py b/app/observability/__init__.py new file mode 100644 index 0000000..0bfec9a --- /dev/null +++ b/app/observability/__init__.py @@ -0,0 +1,2 @@ +from .config import PrometheusConfig +from .otel_agent import OpenTelemetryAgent diff --git a/app/observability/config.py b/app/observability/config.py new file mode 100644 index 0000000..10bc273 --- /dev/null +++ b/app/observability/config.py @@ -0,0 +1,14 @@ +"""Observability config is defined here.""" + +from dataclasses import dataclass + + +@dataclass +class PrometheusConfig: + host: str + port: int + + +@dataclass +class ObservabilityConfig: + prometheus: PrometheusConfig | None = None diff --git a/app/observability/metrics.py b/app/observability/metrics.py new file mode 100644 index 0000000..86c818e --- /dev/null +++ b/app/observability/metrics.py @@ -0,0 +1,129 @@ +"""Application metrics are defined here.""" + +import threading +import time +from dataclasses import dataclass +from typing import Callable + +import psutil +from opentelemetry import metrics +from opentelemetry.metrics import CallbackOptions, Observation +from opentelemetry.sdk.metrics import Counter, Histogram, UpDownCounter + +from app.__version__ import APP_VERSION as VERSION + + +@dataclass +class HTTPMetrics: + request_processing_duration: Histogram + """Processing time histogram in seconds by `["method", "path"]`.""" + requests_started: Counter + """Total started requests counter by `["method", "path"]`.""" + requests_finished: Counter + """Total finished requests counter by `["method", "path", "status_code"]`.""" + errors: Counter + """Total errors (exceptions) counter by `["method", "path", "error_type", "status_code"]`.""" + inflight_requests: UpDownCounter + """Current number of requests handled simultaniously.""" + + +@dataclass +class Metrics: + http: HTTPMetrics + + +def setup_metrics() -> Metrics: + meter = metrics.get_meter("sirtep-api") + + _setup_callback_metrics(meter) + + return Metrics( + http=HTTPMetrics( + request_processing_duration=meter.create_histogram( + "request_processing_duration", + "sec", + "Request processing duration time in seconds", + explicit_bucket_boundaries_advisory=[ + 0.05, + 0.2, + 0.3, + 0.7, + 1.0, + 1.5, + 2.5, + 5.0, + 10.0, + 20.0, + 40.0, + 60.0, + 120.0, + ], + ), + requests_started=meter.create_counter( + "requests_started_total", "1", "Total number of started requests" + ), + requests_finished=meter.create_counter( + "request_finished_total", "1", "Total number of finished requests" + ), + errors=meter.create_counter( + "request_errors_total", + "1", + "Total number of errors (exceptions) in requests", + ), + inflight_requests=meter.create_up_down_counter( + "inflight_requests", + "1", + "Current number of requests handled simultaniously", + ), + ) + ) + + +def _setup_callback_metrics(meter: metrics.Meter) -> None: + # Create observable gauge + meter.create_observable_gauge( + name="system_resource_usage", + description="System resource utilization", + unit="1", + callbacks=[_get_system_metrics_callback()], + ) + meter.create_observable_gauge( + name="application_metrics", + description="Application-specific metrics", + unit="1", + callbacks=[_get_application_metrics_callback()], + ) + + +def _get_system_metrics_callback() -> Callable[[CallbackOptions], None]: + def system_metrics_callback( + options: CallbackOptions, + ): # pylint: disable=unused-argument + """Callback function to collect system metrics""" + + # Process CPU time, a bit more information than `process_cpu_seconds_total` + cpu_times = psutil.Process().cpu_times() + yield Observation(cpu_times.user, {"resource": "cpu", "mode": "user"}) + yield Observation(cpu_times.system, {"resource": "cpu", "mode": "system"}) + + return system_metrics_callback + + +def _get_application_metrics_callback() -> Callable[[CallbackOptions], None]: + startup_time = time.time() + + def application_metrics_callback( + options: CallbackOptions, + ): # pylint: disable=unused-argument + """Callback function to collect application-specific metrics""" + # Current timestamp + yield Observation(startup_time, {"metric": "startup_time", "version": VERSION}) + yield Observation( + time.time(), {"metric": "last_update_time", "version": VERSION} + ) + + # Active threads + active_threads = threading.active_count() + yield Observation(active_threads, {"metric": "active_threads"}) + + return application_metrics_callback diff --git a/app/observability/metrics_server.py b/app/observability/metrics_server.py new file mode 100644 index 0000000..520159a --- /dev/null +++ b/app/observability/metrics_server.py @@ -0,0 +1,22 @@ +"""Prometheus server configuration class is defined here.""" + +from threading import Thread +from wsgiref.simple_server import WSGIServer + +from prometheus_client import start_http_server + + +class PrometheusServer: # pylint: disable=too-few-public-methods + + def __init__(self, port: int = 9464, host: str = "0.0.0.0"): + self._host = host + self._port = port + self._server: WSGIServer + self._thread: Thread + + self._server, self._thread = start_http_server(self._port) + + def shutdown(self): + if self._server is not None: + self._server.shutdown() + self._server = None diff --git a/app/observability/otel_agent.py b/app/observability/otel_agent.py new file mode 100644 index 0000000..509a0f3 --- /dev/null +++ b/app/observability/otel_agent.py @@ -0,0 +1,53 @@ +"""Open Telemetry agent initialization is defined here""" + +import platform +from functools import cache + +from opentelemetry import metrics +from opentelemetry.exporter.prometheus import PrometheusMetricReader +from opentelemetry.sdk.metrics import MeterProvider +from opentelemetry.sdk.resources import ( + SERVICE_INSTANCE_ID, + SERVICE_NAME, + SERVICE_VERSION, + Resource, +) + +from app.__version__ import APP_VERSION + +from .config import PrometheusConfig +from .metrics_server import PrometheusServer + + +@cache +def get_resource() -> Resource: + return Resource.create( + attributes={ + SERVICE_NAME: "sirtep-api", + SERVICE_VERSION: APP_VERSION, + SERVICE_INSTANCE_ID: platform.node(), + } + ) + + +class OpenTelemetryAgent: # pylint: disable=too-few-public-methods + def __init__( + self, + prometheus_config: PrometheusConfig | None, + ): + self._resource = get_resource() + self._prometheus: PrometheusServer | None = None + + if prometheus_config is not None: + self._prometheus = PrometheusServer( + port=prometheus_config.port, host=prometheus_config.host + ) + + reader = PrometheusMetricReader() + provider = MeterProvider(resource=self._resource, metric_readers=[reader]) + metrics.set_meter_provider(provider) + + def shutdown(self) -> None: + """Stop metrics and tracing services if they were started.""" + if self._prometheus is not None: + self._prometheus.shutdown() diff --git a/app/provision/__init__.py b/app/provision/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/provision/provision_controller.py b/app/provision/provision_controller.py new file mode 100644 index 0000000..bc7bb47 --- /dev/null +++ b/app/provision/provision_controller.py @@ -0,0 +1,34 @@ +from typing import Annotated + +from fastapi import APIRouter, Depends + +from app.common.auth.service_auth import get_current_user_id +from app.dependencies import provision_service +from app.dto.provision_dto import ProvisionDTO +from app.schemas.provision_base_schema import ( + MultiProvisionRequestSchema, + MultiProvisionSchema, + ProvisionSchema, +) + +provision_router = APIRouter(prefix="/provision", tags=["provision"]) + + +@provision_router.get("/calc_provision", response_model=ProvisionSchema) +async def calculate_provision( + provision_dto: Annotated[ProvisionDTO, Depends(ProvisionDTO)], + user_id: str = Depends(get_current_user_id), +) -> ProvisionSchema: + + return await provision_service.calculate_provision(provision_dto, user_id) + + +@provision_router.post("/calc_provisions", response_model=MultiProvisionSchema) +async def calculate_multi_provision( + multi_provision_params: MultiProvisionRequestSchema, + user_id: str = Depends(get_current_user_id), +) -> MultiProvisionSchema: + + return await provision_service.calculate_multi_provision( + multi_provision_params, user_id + ) diff --git a/app/provision/provision_mcp.py b/app/provision/provision_mcp.py new file mode 100644 index 0000000..d63e9d7 --- /dev/null +++ b/app/provision/provision_mcp.py @@ -0,0 +1,133 @@ +import traceback + +from fastmcp import FastMCP +from loguru import logger + +from app.common.auth.service_auth import get_mcp_user_id +from app.dependencies import provision_mcp_service, service_token_verifier +from app.dto.provision_dto import ProvisionDTO +from app.schemas.provision_base_schema import ( + MultiProvisionRequestSchema, + ServiceInfoSchema, +) + +provision_mcp = FastMCP("Object Provision MCP server", auth=service_token_verifier) + + +@provision_mcp.tool( + name="CalculateServiceProvision", + title="Get service provision for scenario", + description=""" + Calculate service provision by service type id for scenario id. + Population and demand are restored from Urban API data, then provision is evaluated + with a gravity-based model within the service normative accessibility. + + Args to select: + - scenario_id (int): Scenario ID from Urban API to calculate provision for. + - service_type_id (int): Service type ID to calculate provision for. + - target_population (int, optional): Total population of the scenario territory for demand + calculation. If not provided, population is restored from Urban API data. + + Returns provision layers as GeoJSON FeatureCollections in WGS84 (EPSG:4326). + Response format: + { + "buildings": FeatureCollection, + "services": FeatureCollection, + "links": FeatureCollection + } + """, +) +async def calc_service_provision( + scenario_id: int, service_type_id: int, target_population: int | None = None +): + + try: + user_id = get_mcp_user_id() + project_id = await provision_mcp_service.gateway.get_project_id_by_scenario( + scenario_id, user_id + ) + provision_dto = ProvisionDTO( + project_id=project_id, + scenario_id=scenario_id, + service_type_id=service_type_id, + target_population=target_population, + ) + result = await provision_mcp_service.calculate_provision(provision_dto, user_id) + return result.model_dump() + except Exception as e: + tb = traceback.format_exc() + logger.opt(exception=True).error( + f"Error in MCP tool 'CalculateServiceProvision': {type(e).__name__}: {e}" + ) + raise Exception(f"{type(e).__name__}: {e}\n\nTraceback:\n{tb}") from e + + +@provision_mcp.tool( + name="CalculateServicesProvision", + title="Get provision for multiple services", + description=""" + Calculate service provision for several service types at once for scenario id. + Population and demand are restored from Urban API data, then provision is evaluated + per service type with a gravity-based model within the service normative accessibility. + + Args to select: + - scenario_id (int): Scenario ID from Urban API to calculate provision for. + - services (dict): Service type IDs to calculate, each with display name and layer flag: + {"22": {"name": "Школа", "as_layer": true}, "21": {"name": "Детский сад", "as_layer": false}} + For as_layer=true the response includes GeoJSON layers, otherwise only summary statistics. + - target_population (int, optional): Total population of the scenario territory for demand + calculation, shared by all services. If not provided, population is restored from Urban API data. + + Returns per-service results keyed by service type id. + Response format: + { + "services": { + "22": { + "name": str, + "summary": { + "services_count": int, + "total_capacity": int, + "total_demand": int, + "satisfied_demand_within": int, + "satisfied_demand_without": int, + "unsatisfied_demand": int, + "balance": int, + "deficit": int, + "surplus": int, + "average_provision_value": float, + "median_provision_value": float + }, + "layers": { + "buildings": FeatureCollection, + "services": FeatureCollection, + "links": FeatureCollection + } | null, + "error": str | null + } + } + } + """, +) +async def calc_services_provision( + scenario_id: int, + services: dict[int, ServiceInfoSchema], + target_population: int | None = None, +): + + try: + user_id = get_mcp_user_id() + multi_provision_params = MultiProvisionRequestSchema( + scenario_id=scenario_id, + services=services, + target_population=target_population, + ) + result = await provision_mcp_service.calculate_multi_provision( + multi_provision_params, user_id + ) + return result.model_dump() + except Exception as e: + tb = traceback.format_exc() + logger.opt(exception=True).error( + f"Error in MCP tool 'CalculateServicesProvision': {type(e).__name__}: {e}" + ) + raise Exception(f"{type(e).__name__}: {e}\n\nTraceback:\n{tb}") from e diff --git a/app/provision/provision_service.py b/app/provision/provision_service.py new file mode 100644 index 0000000..8e6d1fc --- /dev/null +++ b/app/provision/provision_service.py @@ -0,0 +1,326 @@ +import asyncio +import json + +import geopandas as gpd +import pandas as pd +from loguru import logger + +from app.common.exceptions.http_exception_wrapper import http_exception +from app.common.modules import ( + EffectsAPIGateway, + attribute_parser, + data_restorator, + matrix_builder, + objectnat_calculator, +) +from app.dto.provision_dto import ProvisionDTO +from app.schemas.provision_base_schema import ( + MultiProvisionRequestSchema, + MultiProvisionSchema, + ProvisionSchema, + ProvisionSummarySchema, + ServiceProvisionResultSchema, +) + +LIVING_BUILDINGS_ID = 4 + + +class ProvisionService: + + def __init__(self, gateway: EffectsAPIGateway) -> None: + self.gateway = gateway + + async def _fetch_shared_data( + self, + project_id: int, + scenario_id: int, + token: str, + ) -> dict: + """ + Fetch scenario data which does not depend on service type + Args: + project_id (int): Project ID + scenario_id (int): Target scenario ID + token (str): Authorization token + Returns: + dict: project data, context and target scenario buildings with populations + """ + + project_data = await self.gateway.get_project_data(project_id, token) + project_territory = await self.gateway.get_project_territory(project_id, token) + context_population = await self.gateway.get_context_population( + territory_ids_list=project_data["properties"]["context"], token=token + ) + context_buildings = await self.gateway.get_project_context_buildings( + scenario_id=project_data["base_scenario"]["id"], token=token + ) + context_buildings.drop( + index=context_buildings.sjoin(project_territory).index, inplace=True + ) + context_buildings = await attribute_parser.parse_all_from_buildings( + living_buildings=context_buildings, + ) + target_scenario_population = await self.gateway.get_scenario_population_data( + scenario_id=scenario_id, token=token + ) + target_scenario_buildings = await self.gateway.get_scenario_buildings( + scenario_id=scenario_id, token=token + ) + target_scenario_buildings = await attribute_parser.parse_all_from_buildings( + living_buildings=target_scenario_buildings, + ) + return { + "project_data": project_data, + "context_population": context_population, + "context_buildings": context_buildings, + "target_scenario_population": target_scenario_population, + "target_scenario_buildings": target_scenario_buildings, + } + + async def _calculate_for_service( + self, + shared_data: dict, + scenario_id: int, + service_type_id: int, + token: str, + ) -> dict[str, gpd.GeoDataFrame]: + """ + Calculate provision for one service type over prefetched scenario data + Args: + shared_data (dict): data prefetched by _fetch_shared_data + scenario_id (int): Target scenario ID + service_type_id (int): Service type ID + token (str): Authorization token + Returns: + dict[str, gpd.GeoDataFrame]: dict with fields "buildings", "services" and "links" + """ + + project_data = shared_data["project_data"] + service_default_capacity = await self.gateway.get_default_capacity( + service_type_id=service_type_id + ) + normative_data = await self.gateway.get_service_normative( + territory_id=project_data["territory"]["id"], + context_ids=project_data["properties"]["context"], + service_type_id=service_type_id, + token=token, + ) + context_buildings = await asyncio.to_thread( + data_restorator.restore_demands, + buildings=shared_data["context_buildings"].copy(), + service_normative=normative_data["services_capacity_per_1000_normative"], + service_normative_type=normative_data["capacity_type"], + target_population=shared_data["context_population"], + ) + context_buildings["is_project"] = False + context_services = await self.gateway.get_project_context_services( + scenario_id=project_data["base_scenario"]["id"], + service_type_id=service_type_id, + token=token, + ) + if context_services.empty: + # ToDo Revise to another code + raise http_exception( + status_code=404, + msg="No services of {service_type_id} type found in context", + _input={"service_type_id": service_type_id}, + _detail={}, + ) + context_services = await attribute_parser.parse_all_from_services( + services=context_services, service_default_capacity=service_default_capacity + ) + target_scenario_buildings = await asyncio.to_thread( + data_restorator.restore_demands, + buildings=shared_data["target_scenario_buildings"].copy(), + service_normative=normative_data["services_capacity_per_1000_normative"], + service_normative_type=normative_data["capacity_type"], + target_population=shared_data["target_scenario_population"], + ) + target_scenario_buildings["is_project"] = True + target_scenario_services = await self.gateway.get_scenario_services( + scenario_id=scenario_id, + service_type_id=service_type_id, + token=token, + ) + target_scenario_services = await attribute_parser.parse_all_from_services( + services=target_scenario_services, + service_default_capacity=service_default_capacity, + ) + before_buildings = await asyncio.to_thread( + pd.concat, + objs=[context_buildings, target_scenario_buildings], + ) + before_services = await asyncio.to_thread( + pd.concat, objs=[context_services, target_scenario_services] + ) + before_buildings.sort_values("is_project", ascending=False, inplace=True) + before_buildings.drop_duplicates("building_id", keep="first", inplace=True) + before_buildings.set_index("building_id", inplace=True) + before_services.set_index("service_id", inplace=True) + before_services.drop_duplicates("geometry", inplace=True) + before_services = before_services[ + ~before_services.index.duplicated(keep="first") + ].copy() + if target_scenario_buildings.empty: + local_crs = context_buildings.estimate_utm_crs() + else: + local_crs = target_scenario_buildings.estimate_utm_crs() + before_buildings.to_crs(local_crs, inplace=True) + before_services.to_crs(local_crs, inplace=True) + before_matrix = await asyncio.to_thread( + matrix_builder.calculate_availability_matrix, + buildings=before_buildings, + services=before_services, + normative_value=normative_data["normative_value"], + normative_type=normative_data["normative_type"], + ) + before_services["capacity"] = before_services["capacity"].fillna( + before_services["capacity"].mean() + ) + before_prove_data = await asyncio.to_thread( + objectnat_calculator.evaluate_provision, + buildings=before_buildings, + services=before_services[~before_services.index.duplicated(keep="first")], + matrix=before_matrix, + service_normative=normative_data["normative_value"], + ) + return before_prove_data + + @staticmethod + def _build_summary( + buildings: gpd.GeoDataFrame, + services: gpd.GeoDataFrame, + ) -> ProvisionSummarySchema: + """ + Aggregate provision results into summary statistics + Args: + buildings (gpd.GeoDataFrame): buildings layer with provision attributes + services (gpd.GeoDataFrame): services layer with load attributes + Returns: + ProvisionSummarySchema: aggregated provision statistics + """ + + provision_values = buildings["provision_value"].dropna() + total_capacity = int(services["capacity"].sum()) + total_demand = int(buildings["demand"].sum()) + balance = total_capacity - total_demand + return ProvisionSummarySchema( + services_count=int(len(services)), + total_capacity=total_capacity, + total_demand=total_demand, + satisfied_demand_within=int(buildings["supplied_demands_within"].sum()), + satisfied_demand_without=int(buildings["supplied_demands_without"].sum()), + unsatisfied_demand=int(buildings["demand_left"].sum()), + balance=balance, + deficit=max(0, -balance), + surplus=max(0, balance), + average_provision_value=( + round(float(provision_values.mean()), 3) + if not provision_values.empty + else None + ), + median_provision_value=( + round(float(provision_values.median()), 3) + if not provision_values.empty + else None + ), + ) + + async def calculate_provision( + self, provision_params: ProvisionDTO, token: str + ) -> ProvisionSchema: + """ + Calculate provision effects by project data and target scenario + Args: + provision_params (ProvisionDTO): Project data + token (str): Authorization token + Returns: + gpd.GeoDataFrame: Provision for scenario. + """ + + logger.info( + f"Started calculating effects for {provision_params.scenario_id} and service{provision_params.service_type_id}" + ) + shared_data = await self._fetch_shared_data( + project_id=provision_params.project_id, + scenario_id=provision_params.scenario_id, + token=token, + ) + if provision_params.target_population: + shared_data["target_scenario_population"] = ( + provision_params.target_population + ) + before_prove_data = await self._calculate_for_service( + shared_data=shared_data, + scenario_id=provision_params.scenario_id, + service_type_id=provision_params.service_type_id, + token=token, + ) + result = {k: json.loads(v.to_json()) for k, v in before_prove_data.items()} + logger.info( + f"Calculated PROVISION for {provision_params.scenario_id} and {provision_params.service_type_id}" + ) + return ProvisionSchema(**result) + + async def calculate_multi_provision( + self, multi_params: MultiProvisionRequestSchema, token: str + ) -> MultiProvisionSchema: + """ + Calculate provision for several service types over one scenario + Args: + multi_params (MultiProvisionRequestSchema): project, scenario and services to calculate + token (str): Authorization token + Returns: + MultiProvisionSchema: per-service summaries with optional GeoJSON layers + """ + + logger.info( + f"Started calculating multi provision for {multi_params.scenario_id} " + f"and services {list(multi_params.services)}" + ) + project_id = await self.gateway.get_project_id_by_scenario( + multi_params.scenario_id, token + ) + shared_data = await self._fetch_shared_data( + project_id=project_id, + scenario_id=multi_params.scenario_id, + token=token, + ) + if multi_params.target_population: + shared_data["target_scenario_population"] = multi_params.target_population + results = {} + for service_type_id, service_info in multi_params.services.items(): + try: + before_prove_data = await self._calculate_for_service( + shared_data=shared_data, + scenario_id=multi_params.scenario_id, + service_type_id=service_type_id, + token=token, + ) + except Exception as e: + logger.opt(exception=True).error( + f"Provision calculation failed for service type {service_type_id}: {e}" + ) + results[service_type_id] = ServiceProvisionResultSchema( + name=service_info.name, + error=f"{type(e).__name__}: {e}", + ) + continue + layers = None + if service_info.as_layer: + layers = ProvisionSchema( + **{ + k: json.loads(v.to_crs(4326).to_json()) + for k, v in before_prove_data.items() + } + ) + results[service_type_id] = ServiceProvisionResultSchema( + name=service_info.name, + summary=self._build_summary( + buildings=before_prove_data["buildings"], + services=before_prove_data["services"], + ), + layers=layers, + ) + logger.info(f"Calculated MULTI PROVISION for {multi_params.scenario_id}") + return MultiProvisionSchema(services=results) diff --git a/app/schemas/__init__.py b/app/schemas/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/schemas/provision_base_schema.py b/app/schemas/provision_base_schema.py new file mode 100644 index 0000000..b75ba0a --- /dev/null +++ b/app/schemas/provision_base_schema.py @@ -0,0 +1,98 @@ +from typing import Any, Literal, Optional + +from pydantic import BaseModel, Field + + +class GeometrySchema(BaseModel): + + type: Literal[ + "Polygon", + "MultiPolygon", + "LineString", + "MultiLineString", + "Point", + "MultiPoint", + ] + coordinates: list[Any] + + +class FeatureSchema(BaseModel): + + id: Optional[int | None] + type: Literal["Feature"] + geometry: GeometrySchema + properties: dict + + +class FeatureCollectionSchema(BaseModel): + + type: Literal["FeatureCollection"] + features: list[FeatureSchema] + + +class ProvisionSchema(BaseModel): + + buildings: FeatureCollectionSchema + services: FeatureCollectionSchema + links: FeatureCollectionSchema + + +class ServiceInfoSchema(BaseModel): + + name: str = Field(..., examples=["Школа"], description="Service display name") + as_layer: bool = Field( + default=True, + description="If true, response includes GeoJSON layers for the service", + ) + + +class MultiProvisionRequestSchema(BaseModel): + + scenario_id: int = Field(..., examples=[192], description="Scenario ID") + services: dict[int, ServiceInfoSchema] = Field( + ..., + examples=[{22: {"name": "Школа", "as_layer": True}}], + description="Service type IDs to calculate provision for", + ) + target_population: int | None = Field( + default=None, + examples=[25000], + description=( + "Total population of the scenario territory for demand calculation. " + "If not provided, population is restored from Urban API data." + ), + ) + + +class ProvisionSummarySchema(BaseModel): + + services_count: int + total_capacity: int + total_demand: int + satisfied_demand_within: int + satisfied_demand_without: int + unsatisfied_demand: int + balance: int = Field( + description="Capacity minus demand; negative means shortage of places" + ) + deficit: int = Field( + ge=0, description="Places short of demand, 0 when capacity covers demand" + ) + surplus: int = Field( + ge=0, description="Places above demand, 0 when demand exceeds capacity" + ) + average_provision_value: float | None + median_provision_value: float | None + + +class ServiceProvisionResultSchema(BaseModel): + + name: str + summary: ProvisionSummarySchema | None = None + layers: ProvisionSchema | None = None + error: str | None = None + + +class MultiProvisionSchema(BaseModel): + + services: dict[int, ServiceProvisionResultSchema] diff --git a/docker-compose.actions.yml b/docker-compose.actions.yml new file mode 100644 index 0000000..eb786bd --- /dev/null +++ b/docker-compose.actions.yml @@ -0,0 +1,15 @@ +services: + object_effects: + image: ${IMAGE} + container_name: ${CONTAINER_NAME} + ports: + - "5080:80" + - "9464:9464" + env_file: + - .env.development + environment: + SERVICE_AUTH_SERVER_URL: ${SERVICE_AUTH_SERVER_URL:?SERVICE_AUTH_SERVER_URL is required} + SERVICE_AUTH_REALM: ${SERVICE_AUTH_REALM:?SERVICE_AUTH_REALM is required} + SERVICE_AUTH_CLIENT_ID: ${SERVICE_AUTH_CLIENT_ID:?SERVICE_AUTH_CLIENT_ID is required} + SERVICE_AUTH_CLIENT_SECRET: ${SERVICE_AUTH_CLIENT_SECRET:?SERVICE_AUTH_CLIENT_SECRET is required} + restart: always diff --git a/docker-compose.yml b/docker-compose.yml index 374af27..cd9a07a 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -6,5 +6,14 @@ services: build: context: . dockerfile: ./Dockerfile + env_file: + - .env.development ports: - - 80:80 \ No newline at end of file + - "8080:80" + - "9464:9464" + networks: + - localnet + +networks: + localnet: + external: true diff --git a/pip.conf b/pip.conf new file mode 100644 index 0000000..d16d3ff --- /dev/null +++ b/pip.conf @@ -0,0 +1,4 @@ +[global] +index-url=http://10.32.11.13:3141/root/pypi/+simple/ +trusted-host=10.32.11.13 +timeout=120 diff --git a/requirements-auth.txt b/requirements-auth.txt new file mode 100644 index 0000000..cd5e3cb --- /dev/null +++ b/requirements-auth.txt @@ -0,0 +1 @@ +idu-service-auth @ https://github.com/IDUclub/idu-service-auth/archive/1b8a418d9b1ab702860eb7289a3b75244666a8d8.tar.gz diff --git a/requirements-dev.txt b/requirements-dev.txt new file mode 100644 index 0000000..3351d0b --- /dev/null +++ b/requirements-dev.txt @@ -0,0 +1 @@ +pre-commit~=4.5.1 \ No newline at end of file diff --git a/requirements.txt b/requirements.txt index e478cf4..eb2fd96 100644 Binary files a/requirements.txt and b/requirements.txt differ diff --git a/tests/test_public_diagnostics.py b/tests/test_public_diagnostics.py new file mode 100644 index 0000000..7bb78e9 --- /dev/null +++ b/tests/test_public_diagnostics.py @@ -0,0 +1,12 @@ +from fastapi.testclient import TestClient + +from app.main import app + + +def test_logs_are_public(tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + (tmp_path / ".log").write_text("diagnostic log\n", encoding="utf-8") + client = TestClient(app) + response = client.get("/logs") + assert response.status_code == 200 + assert response.text == "diagnostic log\n" diff --git a/tests/test_service_auth_transport.py b/tests/test_service_auth_transport.py new file mode 100644 index 0000000..564d174 --- /dev/null +++ b/tests/test_service_auth_transport.py @@ -0,0 +1,46 @@ +from __future__ import annotations + +from unittest.mock import patch + +import pytest +from fastmcp.exceptions import ToolError + +from app.common.api_handler.api_handler import APIHandler +from app.common.auth.service_auth import get_mcp_user_id + + +class FakeServiceAuth: + async def get_authorization_headers(self): + return {"Authorization": "Bearer service-token"} + + +@pytest.mark.asyncio +async def test_service_token_cannot_be_overridden_by_request_headers(): + handler = APIHandler("http://urban", FakeServiceAuth()) + + headers = await handler._service_headers( + {"Authorization": "Bearer caller-token", "X-User-Id": "u1"} + ) + + assert headers["Authorization"] == "Bearer service-token" + assert headers["X-User-Id"] == "u1" + + +def test_mcp_user_context_comes_from_x_user_id_not_authorization(): + with patch( + "app.common.auth.service_auth.get_http_headers", + return_value={ + "authorization": "Bearer service-token", + "x-user-id": " user-42 ", + }, + ): + assert get_mcp_user_id() == "user-42" + + +def test_mcp_user_context_is_required(): + with patch( + "app.common.auth.service_auth.get_http_headers", + return_value={"authorization": "Bearer service-token"}, + ): + with pytest.raises(ToolError, match="X-User-Id header is required"): + get_mcp_user_id() diff --git a/tests/test_urban_api_url.py b/tests/test_urban_api_url.py new file mode 100644 index 0000000..e5aa1de --- /dev/null +++ b/tests/test_urban_api_url.py @@ -0,0 +1,83 @@ +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from app.common.api_handler.api_handler import APIHandler +from app.common.api_handler.urban_api_url import normalize_urban_api_url + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "base, api_root", + [ + ("https://urban.test:8443", "https://urban.test:8443/api"), + ("https://urban.test:8443/", "https://urban.test:8443/api"), + ("https://urban.test:8443/api", "https://urban.test:8443/api"), + (" https://urban.test:8443/api/// ", "https://urban.test:8443/api"), + ( + "https://prostor-api.idu.actocgnitive.org/urban_api", + "https://prostor-api.idu.actocgnitive.org/urban_api", + ), + ( + "https://prostor-api.idu.actocgnitive.org/urban_api/", + "https://prostor-api.idu.actocgnitive.org/urban_api", + ), + ( + "https://urban.test/gateway/urban_api/", + "https://urban.test/gateway/urban_api", + ), + ], +) +@pytest.mark.parametrize("method", ["get", "post", "put", "delete"]) +@pytest.mark.parametrize( + "endpoint", + ["/api/v1/scenarios/7", "api/v1/scenarios/7", "/v1/scenarios/7", "v1/scenarios/7"], +) +async def test_requests_use_configured_api_root(base, api_root, method, endpoint): + auth = MagicMock() + auth.get_authorization_headers = AsyncMock( + return_value={"Authorization": "Bearer service"} + ) + handler = APIHandler(base, auth) + session = MagicMock() + response = MagicMock(status=200) + response.json = AsyncMock(return_value={"ok": True}) + request = getattr(session, method) + request.return_value.__aenter__.return_value = response + + assert await getattr(handler, method)( + endpoint, session=session, params={"page": 2} + ) == {"ok": True} + + assert request.call_args.kwargs["url"] == f"{api_root}/v1/scenarios/7" + assert request.call_args.kwargs["params"] == {"page": 2} + assert request.call_args.kwargs["headers"] == {"Authorization": "Bearer service"} + + +@pytest.mark.parametrize( + "base, expected", + [ + ("http://api", "http://api/api"), + ("https://urban.test/gateway/api/", "https://urban.test/gateway/api"), + ("https://urban.test/gateway/", "https://urban.test/gateway"), + ("https://urban.test/api/api/", "https://urban.test/api/api"), + ], +) +def test_normalization_preserves_authority_and_proxy_path(base, expected): + assert normalize_urban_api_url(base) == expected + assert normalize_urban_api_url(expected) == expected + + +@pytest.mark.parametrize( + "base", + [ + "", + "urban.test", + "ftp://urban.test", + "https://urban.test?x=1", + "https://urban.test#fragment", + ], +) +def test_invalid_base_url_is_rejected(base): + with pytest.raises(ValueError, match="Urban API URL"): + normalize_urban_api_url(base)