sajad-dev/Schemas
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

UPDATE

Browse Source
This commit is contained in:
Mohammad Sajad Pourajam
2026-04-27 16:40:50 +03:30
parent 78acb5510d
commit 993066a19f
11 changed files with 801 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files
__init__.py
+38 -20
View File
@@ -5,35 +5,53 @@ from .crop_simulation_growth_status import CONTRACT as CROP_SIMULATION_GROWTH_ST
from .crop_simulation_harvest_prediction import CONTRACT as CROP_SIMULATION_HARVEST_PREDICTION_CONTRACT
from .crop_simulation_yield_prediction import CONTRACT as CROP_SIMULATION_YIELD_PREDICTION_CONTRACT
from .economy_overview import CONTRACT as ECONOMY_OVERVIEW_CONTRACT
from .farm_alerts import CONTRACTS as FARM_ALERTS_CONTRACTS
from .farm_data_upsert import CONTRACT as FARM_DATA_UPSERT_CONTRACT
from .farm_detail import CONTRACT as FARM_DETAIL_CONTRACT
from .farm_parameter import CONTRACT as FARM_PARAMETER_CONTRACT
from .fertilization_recommend import CONTRACT as FERTILIZATION_RECOMMEND_CONTRACT
from .irrigation_list import CONTRACT as IRRIGATION_LIST_CONTRACT
from .irrigation_methods import CONTRACTS as IRRIGATION_METHOD_CONTRACTS
from .irrigation_recommend import CONTRACT as IRRIGATION_RECOMMEND_CONTRACT
from .irrigation_water_stress import CONTRACT as IRRIGATION_WATER_STRESS_CONTRACT
from .pest_disease import CONTRACTS as PEST_DISEASE_CONTRACTS
from .plant import CONTRACTS as PLANT_CONTRACTS
from .rag_chat import CONTRACT as RAG_CHAT_CONTRACT
from .soil_data import CONTRACTS as SOIL_DATA_CONTRACTS
from .soile_anomaly_detection import CONTRACT as SOILE_ANOMALY_DETECTION_CONTRACT
from .soile_health_summary import CONTRACT as SOILE_HEALTH_SUMMARY_CONTRACT
from .soile_moisture_heatmap import CONTRACT as SOILE_MOISTURE_HEATMAP_CONTRACT
from .weather_farm_card import CONTRACT as WEATHER_FARM_CARD_CONTRACT
from .weather_water_need_prediction import CONTRACT as WEATHER_WATER_NEED_PREDICTION_CONTRACT
ALL_ROUTE_CONTRACTS: list[RouteContract] = [
RAG_CHAT_CONTRACT,
*FARM_ALERTS_CONTRACTS,
*SOIL_DATA_CONTRACTS,
SOILE_MOISTURE_HEATMAP_CONTRACT,
SOILE_HEALTH_SUMMARY_CONTRACT,
SOILE_ANOMALY_DETECTION_CONTRACT,
FARM_DATA_UPSERT_CONTRACT,
FARM_DETAIL_CONTRACT,
FARM_PARAMETER_CONTRACT,
WEATHER_FARM_CARD_CONTRACT,
WEATHER_WATER_NEED_PREDICTION_CONTRACT,
ECONOMY_OVERVIEW_CONTRACT,
*PLANT_CONTRACTS,
*PEST_DISEASE_CONTRACTS,
*IRRIGATION_METHOD_CONTRACTS,
IRRIGATION_RECOMMEND_CONTRACT,
IRRIGATION_WATER_STRESS_CONTRACT,
FERTILIZATION_RECOMMEND_CONTRACT,
CROP_SIMULATION_GROWTH_CONTRACT,
CROP_SIMULATION_GROWTH_STATUS_CONTRACT,
CROP_SIMULATION_CURRENT_FARM_CHART_CONTRACT,
CROP_SIMULATION_HARVEST_PREDICTION_CONTRACT,
CROP_SIMULATION_YIELD_PREDICTION_CONTRACT,
]
ROUTE_CONTRACTS: dict[str, RouteContract] = {
contract.path: contract
for contract in [
RAG_CHAT_CONTRACT,
SOILE_MOISTURE_HEATMAP_CONTRACT,
SOILE_HEALTH_SUMMARY_CONTRACT,
SOILE_ANOMALY_DETECTION_CONTRACT,
FARM_DATA_UPSERT_CONTRACT,
WEATHER_WATER_NEED_PREDICTION_CONTRACT,
ECONOMY_OVERVIEW_CONTRACT,
IRRIGATION_LIST_CONTRACT,
IRRIGATION_RECOMMEND_CONTRACT,
FERTILIZATION_RECOMMEND_CONTRACT,
CROP_SIMULATION_GROWTH_CONTRACT,
CROP_SIMULATION_GROWTH_STATUS_CONTRACT,
CROP_SIMULATION_CURRENT_FARM_CHART_CONTRACT,
CROP_SIMULATION_HARVEST_PREDICTION_CONTRACT,
CROP_SIMULATION_YIELD_PREDICTION_CONTRACT,
]
f'{contract.method} {contract.path}': contract
for contract in ALL_ROUTE_CONTRACTS
}
__all__ = ['ROUTE_CONTRACTS', 'RouteContract']
__all__ = ['ALL_ROUTE_CONTRACTS', 'ROUTE_CONTRACTS', 'RouteContract']
farm_alerts.py
+90
View File
@@ -0,0 +1,90 @@
from __future__ import annotations
from typing import Literal
from uuid import UUID
from pydantic import Field
from .common import ApiEnvelope, JsonObject, JsonValue, RouteContract, SchemaModel
class FarmAlertsRequest(SchemaModel):
farm_uuid: UUID
sensor_uuid: UUID | None = None
query: str | None = None
class FarmAlertNotificationSchema(SchemaModel):
id: int | None = None
farm_uuid: str | None = None
endpoint: str | None = None
level: Literal['danger', 'warning', 'info'] | str
title: str
message: str
suggested_action: str | None = None
source_alert_id: str | None = None
source_metric_type: str | None = None
payload: JsonObject = Field(default_factory=dict)
created_at: str | None = None
updated_at: str | None = None
class FarmAlertsTimelineItem(SchemaModel):
timestamp: str | None = None
level: Literal['danger', 'warning', 'info'] | str
title: str
description: str | None = None
source_alert_id: str | None = None
source_metric_type: str | None = None
class FarmAlertsTrackerResponseData(SchemaModel):
farm_uuid: str
service_id: str | None = None
knowledge_base: str | None = None
tone_file: str | None = None
tracker: JsonObject = Field(default_factory=dict)
headline: str
overview: str | None = None
status_level: Literal['danger', 'warning', 'info'] | str
notifications: list[FarmAlertNotificationSchema] = Field(default_factory=list)
raw_llm_response: str | None = None
structured_context: JsonObject = Field(default_factory=dict)
class FarmAlertsTimelineResponseData(SchemaModel):
farm_uuid: str
service_id: str | None = None
knowledge_base: str | None = None
tone_file: str | None = None
tracker: JsonObject = Field(default_factory=dict)
headline: str
overview: str | None = None
timeline: list[FarmAlertsTimelineItem] = Field(default_factory=list)
notifications: list[FarmAlertNotificationSchema] = Field(default_factory=list)
raw_llm_response: str | None = None
structured_context: JsonObject = Field(default_factory=dict)
class FarmAlertsTrackerResponse(ApiEnvelope[FarmAlertsTrackerResponseData]):
pass
class FarmAlertsTimelineResponse(ApiEnvelope[FarmAlertsTimelineResponseData]):
pass
CONTRACTS = [
RouteContract(
method='POST',
path='/api/farm-alerts/tracker/',
request_model=FarmAlertsRequest.__name__,
response_model=FarmAlertsTrackerResponse.__name__,
),
RouteContract(
method='POST',
path='/api/farm-alerts/timeline/',
request_model=FarmAlertsRequest.__name__,
response_model=FarmAlertsTimelineResponse.__name__,
),
]
farm_detail.py
+113
View File
@@ -0,0 +1,113 @@
from __future__ import annotations
from pydantic import Field
from .common import ApiEnvelope, JsonObject, RouteContract, SchemaModel
HTTP_METHOD = 'GET'
ROUTE_PATH = '/api/farm-data/<farm_uuid>/detail/'
class FarmDetailRequest(SchemaModel):
farm_uuid: str
class FarmCenterLocationSchema(SchemaModel):
id: int
lat: float
lon: float
farm_boundary: JsonObject = Field(default_factory=dict)
class WeatherForecastDetailSchema(SchemaModel):
id: int
forecast_date: str | None = None
temperature_min: float | None = None
temperature_max: float | None = None
temperature_mean: float | None = None
precipitation: float | None = None
precipitation_probability: float | None = None
humidity_mean: float | None = None
wind_speed_max: float | None = None
et0: float | None = None
weather_code: int | None = None
class FarmSoilDepthSchema(SchemaModel):
depth_label: str
bdod: float | None = None
cec: float | None = None
cfvo: float | None = None
clay: float | None = None
nitrogen: float | None = None
ocd: float | None = None
ocs: float | None = None
phh2o: float | None = None
sand: float | None = None
silt: float | None = None
soc: float | None = None
wv0010: float | None = None
wv0033: float | None = None
wv1500: float | None = None
class FarmSoilPayloadSchema(SchemaModel):
resolved_metrics: JsonObject = Field(default_factory=dict)
metric_sources: JsonObject = Field(default_factory=dict)
depths: list[FarmSoilDepthSchema] = Field(default_factory=list)
class FarmPlantSchema(SchemaModel):
id: int
name: str
light: str | None = None
watering: str | None = None
soil: str | None = None
temperature: str | None = None
growth_stage: str | None = None
planting_season: str | None = None
harvest_time: str | None = None
spacing: str | None = None
fertilizer: str | None = None
created_at: str | None = None
updated_at: str | None = None
class FarmIrrigationMethodSchema(SchemaModel):
id: int
name: str
category: str | None = None
description: str | None = None
water_efficiency_percent: float | None = None
water_pressure_required: str | None = None
flow_rate: str | None = None
coverage_area: str | None = None
soil_type: str | None = None
climate_suitability: str | None = None
created_at: str | None = None
updated_at: str | None = None
class FarmDetailResponseData(SchemaModel):
center_location: FarmCenterLocationSchema
weather: WeatherForecastDetailSchema | None = None
sensor_payload: JsonObject = Field(default_factory=dict)
soil: FarmSoilPayloadSchema
plant_ids: list[int] = Field(default_factory=list)
plants: list[FarmPlantSchema] = Field(default_factory=list)
irrigation_method_id: int | None = None
irrigation_method: FarmIrrigationMethodSchema | None = None
created_at: str | None = None
updated_at: str | None = None
class FarmDetailResponse(ApiEnvelope[FarmDetailResponseData]):
pass
CONTRACT = RouteContract(
method=HTTP_METHOD,
path=ROUTE_PATH,
request_model=FarmDetailRequest.__name__,
response_model=FarmDetailResponse.__name__,
)
farm_parameter.py
+41
View File
@@ -0,0 +1,41 @@
from __future__ import annotations
from pydantic import Field
from .common import ApiEnvelope, JsonObject, RouteContract, SchemaModel
HTTP_METHOD = 'POST'
ROUTE_PATH = '/api/farm-data/parameters/'
class FarmParameterRequest(SchemaModel):
sensor_key: str = 'sensor-7-1'
code: str
name_fa: str
unit: str | None = ''
data_type: str | None = 'float'
metadata: JsonObject = Field(default_factory=dict)
class FarmParameterResponseData(SchemaModel):
id: int
sensor_key: str
code: str
name_fa: str
unit: str | None = None
data_type: str | None = None
metadata: JsonObject = Field(default_factory=dict)
created_at: str | None = None
action: str
class FarmParameterResponse(ApiEnvelope[FarmParameterResponseData]):
pass
CONTRACT = RouteContract(
method=HTTP_METHOD,
path=ROUTE_PATH,
request_model=FarmParameterRequest.__name__,
response_model=FarmParameterResponse.__name__,
)
irrigation_list.py
+3 -3
View File
@@ -1,8 +1,8 @@
from __future__ import annotations
from pydantic import RootModel
from pydantic import Field
from .common import EmptyRequest, RouteContract, SchemaModel
from .common import ApiEnvelope, EmptyRequest, RouteContract, SchemaModel
HTTP_METHOD = 'GET'
ROUTE_PATH = '/api/irrigation/'
@@ -27,7 +27,7 @@ class IrrigationMethodSchema(SchemaModel):
updated_at: str | None = None
class IrrigationListResponse(RootModel[list[IrrigationMethodSchema]]):
class IrrigationListResponse(ApiEnvelope[list[IrrigationMethodSchema]]):
pass
irrigation_methods.py
+91
View File
@@ -0,0 +1,91 @@
from __future__ import annotations
from pydantic import Field
from .common import ApiEnvelope, EmptyRequest, JsonObject, JsonValue, RouteContract, SchemaModel
class IrrigationMethodPayload(SchemaModel):
name: str
category: str | None = None
description: str | None = None
water_efficiency_percent: float | None = None
water_pressure_required: str | None = None
flow_rate: str | None = None
coverage_area: str | None = None
soil_type: str | None = None
climate_suitability: str | None = None
class IrrigationMethodPartialPayload(SchemaModel):
name: str | None = None
category: str | None = None
description: str | None = None
water_efficiency_percent: float | None = None
water_pressure_required: str | None = None
flow_rate: str | None = None
coverage_area: str | None = None
soil_type: str | None = None
climate_suitability: str | None = None
class IrrigationMethodSchema(IrrigationMethodPayload):
id: int
created_at: str | None = None
updated_at: str | None = None
class IrrigationMethodDetailRequest(SchemaModel):
pk: int
class IrrigationMethodListResponse(ApiEnvelope[list[IrrigationMethodSchema]]):
pass
class IrrigationMethodDetailResponse(ApiEnvelope[IrrigationMethodSchema]):
pass
class IrrigationMethodDeleteResponse(ApiEnvelope[JsonValue | None]):
pass
CONTRACTS = [
RouteContract(
method='GET',
path='/api/irrigation/',
request_model=EmptyRequest.__name__,
response_model=IrrigationMethodListResponse.__name__,
),
RouteContract(
method='POST',
path='/api/irrigation/',
request_model=IrrigationMethodPayload.__name__,
response_model=IrrigationMethodDetailResponse.__name__,
),
RouteContract(
method='GET',
path='/api/irrigation/<pk>/',
request_model=IrrigationMethodDetailRequest.__name__,
response_model=IrrigationMethodDetailResponse.__name__,
),
RouteContract(
method='PUT',
path='/api/irrigation/<pk>/',
request_model=IrrigationMethodPayload.__name__,
response_model=IrrigationMethodDetailResponse.__name__,
),
RouteContract(
method='PATCH',
path='/api/irrigation/<pk>/',
request_model=IrrigationMethodPartialPayload.__name__,
response_model=IrrigationMethodDetailResponse.__name__,
),
RouteContract(
method='DELETE',
path='/api/irrigation/<pk>/',
request_model=IrrigationMethodDetailRequest.__name__,
response_model=IrrigationMethodDeleteResponse.__name__,
),
]
irrigation_water_stress.py
+35
View File
@@ -0,0 +1,35 @@
from __future__ import annotations
from typing import Literal
from uuid import UUID
from pydantic import Field
from .common import ApiEnvelope, JsonObject, RouteContract, SchemaModel
HTTP_METHOD = 'POST'
ROUTE_PATH = '/api/irrigation/water-stress/'
class IrrigationWaterStressRequest(SchemaModel):
farm_uuid: UUID
sensor_uuid: UUID | None = None
class IrrigationWaterStressResponseData(SchemaModel):
farm_uuid: str
waterStressIndex: int | float
level: Literal['low', 'medium', 'high'] | str
sourceMetric: JsonObject = Field(default_factory=dict)
class IrrigationWaterStressResponse(ApiEnvelope[IrrigationWaterStressResponseData]):
pass
CONTRACT = RouteContract(
method=HTTP_METHOD,
path=ROUTE_PATH,
request_model=IrrigationWaterStressRequest.__name__,
response_model=IrrigationWaterStressResponse.__name__,
)
pest_disease.py
+118
View File
@@ -0,0 +1,118 @@
from __future__ import annotations
from typing import Literal
from uuid import UUID
from pydantic import Field
from .common import ApiEnvelope, JsonObject, JsonValue, RouteContract, SchemaModel
class PestDiseaseDetectionRequest(SchemaModel):
farm_uuid: UUID
sensor_uuid: UUID | None = None
plant_name: str | None = None
query: str | None = None
image_urls: list[str] = Field(default_factory=list)
image: str | None = None
images: list[str] = Field(default_factory=list)
class PestDiseaseDetectionResponseData(SchemaModel):
has_issue: bool
category: Literal['no_issue', 'pest', 'disease', 'nutrient_stress', 'abiotic_stress', 'unknown'] | str
confidence: float | None = None
severity: Literal['low', 'medium', 'high'] | str
summary: str
detected_signs: list[str] = Field(default_factory=list)
possible_causes: list[str] = Field(default_factory=list)
immediate_actions: list[str] = Field(default_factory=list)
reasoning: list[str] = Field(default_factory=list)
farm_uuid: str | None = None
knowledge_base: str | None = None
tone_file: str | None = None
raw_response: str | None = None
class PestDiseaseRiskRequest(SchemaModel):
farm_uuid: UUID
sensor_uuid: UUID | None = None
plant_name: str | None = None
growth_stage: str | None = None
query: str | None = None
class RiskBlock(SchemaModel):
score: float | None = None
level: Literal['low', 'medium', 'high'] | str | None = None
likely_conditions: list[str] = Field(default_factory=list)
reasoning: list[str] = Field(default_factory=list)
statsLabel: str | None = None
class PestDiseaseRiskResponseData(SchemaModel):
summary: str
forecast_window: str | None = None
overall_risk: Literal['low', 'medium', 'high'] | str
disease_risk: RiskBlock = Field(default_factory=RiskBlock)
pest_risk: RiskBlock = Field(default_factory=RiskBlock)
key_drivers: list[str] = Field(default_factory=list)
recommended_actions: list[str] = Field(default_factory=list)
farm_context: JsonObject = Field(default_factory=dict)
farm_uuid: str | None = None
knowledge_base: str | None = None
tone_file: str | None = None
raw_response: str | None = None
class PestDiseaseRiskSummaryRequest(SchemaModel):
farm_uuid: UUID
sensor_uuid: UUID | None = None
class PestDiseaseRiskSummaryDrivers(SchemaModel):
keyDrivers: list[str] = Field(default_factory=list)
summary: str | None = None
forecastWindow: str | None = None
source: str | None = None
class PestDiseaseRiskSummaryResponseData(SchemaModel):
farm_uuid: str
diseaseRisk: RiskBlock = Field(default_factory=RiskBlock)
pestRisk: RiskBlock = Field(default_factory=RiskBlock)
drivers: PestDiseaseRiskSummaryDrivers = Field(default_factory=PestDiseaseRiskSummaryDrivers)
class PestDiseaseDetectionResponse(ApiEnvelope[PestDiseaseDetectionResponseData]):
pass
class PestDiseaseRiskResponse(ApiEnvelope[PestDiseaseRiskResponseData]):
pass
class PestDiseaseRiskSummaryResponse(ApiEnvelope[PestDiseaseRiskSummaryResponseData]):
pass
CONTRACTS = [
RouteContract(
method='POST',
path='/api/pest-disease/detect/',
request_model=PestDiseaseDetectionRequest.__name__,
response_model=PestDiseaseDetectionResponse.__name__,
),
RouteContract(
method='POST',
path='/api/pest-disease/risk/',
request_model=PestDiseaseRiskRequest.__name__,
response_model=PestDiseaseRiskResponse.__name__,
),
RouteContract(
method='POST',
path='/api/pest-disease/risk-summary/',
request_model=PestDiseaseRiskSummaryRequest.__name__,
response_model=PestDiseaseRiskSummaryResponse.__name__,
),
]
plant.py
+107
View File
@@ -0,0 +1,107 @@
from __future__ import annotations
from pydantic import Field
from .common import ApiEnvelope, EmptyRequest, JsonObject, JsonValue, RouteContract, SchemaModel
class PlantPayload(SchemaModel):
name: str
light: str | None = None
watering: str | None = None
soil: str | None = None
temperature: str | None = None
growth_stage: str | None = None
planting_season: str | None = None
harvest_time: str | None = None
spacing: str | None = None
fertilizer: str | None = None
class PlantPartialPayload(SchemaModel):
name: str | None = None
light: str | None = None
watering: str | None = None
soil: str | None = None
temperature: str | None = None
growth_stage: str | None = None
planting_season: str | None = None
harvest_time: str | None = None
spacing: str | None = None
fertilizer: str | None = None
class PlantRecord(PlantPayload):
id: int
created_at: str | None = None
updated_at: str | None = None
class PlantListResponse(ApiEnvelope[list[PlantRecord]]):
pass
class PlantDetailResponse(ApiEnvelope[PlantRecord]):
pass
class PlantDeleteResponse(ApiEnvelope[JsonValue | None]):
pass
class PlantDetailRequest(SchemaModel):
pk: int
class PlantFetchInfoRequest(SchemaModel):
name: str
class PlantFetchInfoResponse(ApiEnvelope[JsonObject]):
pass
CONTRACTS = [
RouteContract(
method='GET',
path='/api/plants/',
request_model=EmptyRequest.__name__,
response_model=PlantListResponse.__name__,
),
RouteContract(
method='POST',
path='/api/plants/',
request_model=PlantPayload.__name__,
response_model=PlantDetailResponse.__name__,
),
RouteContract(
method='GET',
path='/api/plants/<pk>/',
request_model=PlantDetailRequest.__name__,
response_model=PlantDetailResponse.__name__,
),
RouteContract(
method='PUT',
path='/api/plants/<pk>/',
request_model=PlantPayload.__name__,
response_model=PlantDetailResponse.__name__,
),
RouteContract(
method='PATCH',
path='/api/plants/<pk>/',
request_model=PlantPartialPayload.__name__,
response_model=PlantDetailResponse.__name__,
),
RouteContract(
method='DELETE',
path='/api/plants/<pk>/',
request_model=PlantDetailRequest.__name__,
response_model=PlantDeleteResponse.__name__,
),
RouteContract(
method='POST',
path='/api/plants/fetch-info/',
request_model=PlantFetchInfoRequest.__name__,
response_model=PlantFetchInfoResponse.__name__,
),
]
soil_data.py
+124
View File
@@ -0,0 +1,124 @@
from __future__ import annotations
from uuid import UUID
from pydantic import Field
from .common import ApiEnvelope, JsonObject, JsonValue, RouteContract, SchemaModel
class SoilDataCoordinatesRequest(SchemaModel):
lat: float
lon: float
class SoilDepthDataSchema(SchemaModel):
depth_label: str
bdod: float | None = None
cec: float | None = None
cfvo: float | None = None
clay: float | None = None
nitrogen: float | None = None
ocd: float | None = None
ocs: float | None = None
phh2o: float | None = None
sand: float | None = None
silt: float | None = None
soc: float | None = None
wv0010: float | None = None
wv0033: float | None = None
wv1500: float | None = None
class SoilLocationPayload(SchemaModel):
source: str
id: int
lon: float
lat: float
depths: list[SoilDepthDataSchema] = Field(default_factory=list)
class SoilTaskQueuedResponseData(SchemaModel):
source: str = 'task'
task_id: str
lon: float
lat: float
status_url: str | None = None
class SoilTaskStatusResponseData(SchemaModel):
task_id: str
status: str
message: str | None = None
progress: JsonObject = Field(default_factory=dict)
result: JsonValue | None = None
error: str | None = None
class NdviHealthRequest(SchemaModel):
farm_uuid: UUID
class NdviHealthDataItem(SchemaModel):
title: str
value: JsonValue | None = None
color: str
icon: str
class NdviHealthResponseData(SchemaModel):
ndviIndex: float | None = None
mean_ndvi: float | None = None
ndvi_map: JsonObject = Field(default_factory=dict)
vegetation_health_class: str | None = None
observation_date: str | None = None
satellite_source: str | None = None
healthData: list[NdviHealthDataItem] = Field(default_factory=list)
class SoilDataResponse(ApiEnvelope[SoilLocationPayload]):
pass
class SoilTaskQueuedResponse(ApiEnvelope[SoilTaskQueuedResponseData]):
pass
class SoilTaskStatusResponse(ApiEnvelope[SoilTaskStatusResponseData]):
pass
class NdviHealthResponse(ApiEnvelope[NdviHealthResponseData]):
pass
CONTRACTS = [
RouteContract(
method='GET',
path='/api/soil-data/',
request_model=SoilDataCoordinatesRequest.__name__,
response_model=SoilDataResponse.__name__,
),
RouteContract(
method='POST',
path='/api/soil-data/',
request_model=SoilDataCoordinatesRequest.__name__,
response_model=SoilDataResponse.__name__,
),
RouteContract(
method='GET',
path='/api/soil-data/tasks/<task_id>/status/',
request_model='SoilTaskStatusRequest',
response_model=SoilTaskStatusResponse.__name__,
),
RouteContract(
method='POST',
path='/api/soil-data/ndvi-health/',
request_model=NdviHealthRequest.__name__,
response_model=NdviHealthResponse.__name__,
),
]
class SoilTaskStatusRequest(SchemaModel):
task_id: str
weather_farm_card.py
+41
View File
@@ -0,0 +1,41 @@
from __future__ import annotations
from uuid import UUID
from pydantic import Field
from .common import ApiEnvelope, RouteContract, SchemaModel
HTTP_METHOD = 'POST'
ROUTE_PATH = '/api/weather/farm-card/'
class WeatherFarmCardRequest(SchemaModel):
farm_uuid: UUID
class WeatherChartData(SchemaModel):
labels: list[str] = Field(default_factory=list)
series: list[list[float]] = Field(default_factory=list)
class WeatherFarmCardResponseData(SchemaModel):
condition: str
temperature: float | int
unit: str
humidity: float | int
windSpeed: float | int
windUnit: str
chartData: WeatherChartData = Field(default_factory=WeatherChartData)
class WeatherFarmCardResponse(ApiEnvelope[WeatherFarmCardResponseData]):
pass
CONTRACT = RouteContract(
method=HTTP_METHOD,
path=ROUTE_PATH,
request_model=WeatherFarmCardRequest.__name__,
response_model=WeatherFarmCardResponse.__name__,
)