Spaces:
Build error
Build error
File size: 3,579 Bytes
c40c447 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 |
"""
Mapper para casos de uso de Forecasting.
Convierte entre API schemas (Pydantic) y DTOs de aplicación.
"""
from typing import Dict, Any
from app.schemas.requests.forecast import (
ForecastUnivariateRequest,
ForecastMultiSeriesRequest,
SeriesData
)
from app.schemas.responses.forecast import (
ForecastUnivariateResponse,
ForecastMultiSeriesResponse
)
from app.application.dtos.forecast_dtos import (
ForecastInputDTO,
ForecastOutputDTO,
MultiForecastInputDTO,
MultiForecastOutputDTO,
SeriesInputDTO
)
class ForecastMapper:
"""
Mapper para convertir entre API schemas y DTOs de forecasting.
"""
@staticmethod
def to_univariate_input_dto(
request: ForecastUnivariateRequest
) -> ForecastInputDTO:
"""
Convierte API request a DTO de entrada.
Args:
request: Request de la API
Returns:
ForecastInputDTO: DTO para el caso de uso
"""
return ForecastInputDTO(
values=request.series.values,
prediction_length=request.prediction_length,
quantile_levels=request.quantile_levels,
timestamps=request.series.timestamps,
series_id=getattr(request.series, 'series_id', 'series_0'),
freq=request.freq
)
@staticmethod
def from_univariate_output_dto(
dto: ForecastOutputDTO
) -> ForecastUnivariateResponse:
"""
Convierte DTO de salida a API response.
Args:
dto: DTO del caso de uso
Returns:
ForecastUnivariateResponse: Response para la API
"""
return ForecastUnivariateResponse(
timestamps=dto.timestamps,
median=dto.median,
quantiles=dto.quantiles,
series_id=dto.series_id,
metadata=dto.metadata
)
@staticmethod
def to_multi_series_input_dto(
request: ForecastMultiSeriesRequest
) -> MultiForecastInputDTO:
"""
Convierte API request multi-series a DTO de entrada.
Args:
request: Request de la API
Returns:
MultiForecastInputDTO: DTO para el caso de uso
"""
series_list = []
for series_data in request.series_list:
series_dto = SeriesInputDTO(
series_id=series_data.series_id,
values=series_data.values,
timestamps=series_data.timestamps
)
series_list.append(series_dto)
return MultiForecastInputDTO(
series_list=series_list,
prediction_length=request.prediction_length,
quantile_levels=request.quantile_levels,
freq=request.freq
)
@staticmethod
def from_multi_series_output_dto(
dto: MultiForecastOutputDTO
) -> ForecastMultiSeriesResponse:
"""
Convierte DTO de salida multi-series a API response.
Args:
dto: DTO del caso de uso
Returns:
ForecastMultiSeriesResponse: Response para la API
"""
# Convertir cada resultado individual
results = [
ForecastMapper.from_univariate_output_dto(result)
for result in dto.results
]
return ForecastMultiSeriesResponse(
results=results,
total_series=dto.total_series,
successful=dto.successful,
failed=dto.failed
)
|