| import os | |
| from pydantic import BaseModel, Field, validator, ValidationError | |
| import gradio as gr | |
| from openai import OpenAI | |
| from typing import List, Dict, Any, Optional, Literal, Union | |
| # Chatbot model | |
| os.environ["OPENAI_API_KEY"] = os.getenv("OPENAI_API_KEY") | |
| client = OpenAI() | |
| # Experimentation Notes | |
| # maxLength is not permitted, which we have for some fields | |
| # every object requires a 'type' field | |
| # pattern is not permitted, instead we would have to create homemade validators | |
| # doesn't support optional fields, but could use a trick with None if necessary | |
| # | |
| # | |
| # | |
| #class ConventionData(BaseModel): | |
| # type: Literal["taxonomy_term--convention"] = Field(..., description="Constant value indicating the type.") | |
| # id: Any = Field(..., description="ID of the convention.") | |
| #class CategoryData(BaseModel): | |
| # type: Literal["taxonomy_term--log_category"] = Field(..., description="Constant value indicating the type.") | |
| # id: Any = Field(..., description="ID of the log category.") | |
| #class NoteValue(BaseModel): | |
| # value: str = Field(..., title="Text") | |
| # format: Optional[str] = Field(None, title="Text format") | |
| #class IDTag(BaseModel): | |
| # id: str = Field(..., title="ID of the tag") | |
| # type: str = Field(..., title="Type of the tag") | |
| # location: str = Field(..., title="Location of the tag") | |
| #class InventoryItem(BaseModel): | |
| # measure: str = Field(..., title="Measure of the inventory") | |
| # value: str = Field(..., title="Value of the inventory") | |
| # units: str = Field(..., title="Units of the inventory") | |
| #class IntrinsicGeometry(BaseModel): | |
| # value: str = Field(..., title="Geometry") | |
| # geo_type: str = Field(..., title="Geometry Type") | |
| # lat: float = Field(..., title="Centroid Latitude") | |
| # lon: float = Field(..., title="Centroid Longitude") | |
| # left: float = Field(..., title="Left Bounding") | |
| # top: float = Field(..., title="Top Bounding") | |
| # right: float = Field(..., title="Right Bounding") | |
| # bottom: float = Field(..., title="Bottom Bounding") | |
| # geohash: str = Field(..., title="Geohash") | |
| # latlon: str = Field(..., title="LatLong Pair") | |
| #class PlantTypeSpecies(BaseModel): | |
| # type: Literal["taxonomy_term--plant_type"] = Field(..., description="Constant value indicating the type of taxonomy term.") | |
| # id: Dict[str, Any] = Field(..., description="ID of the plant type species") | |
| #class PlantTypeVariety(BaseModel): | |
| # type: Literal["taxonomy_term--plant_type"] = Field(..., description="Constant value indicating the type of taxonomy term.") | |
| # id: Dict[str, Any] = Field(..., description="ID of the plant type variety") | |
| #class Season(BaseModel): | |
| # type: Literal["taxonomy_term--season"] = Field(..., description="Constant value indicating the type of taxonomy term.") | |
| # id: Dict[str, Any] = Field(..., description="ID of the season") | |
| #class PlantRelationships(BaseModel): | |
| # convention: List['ConventionData'] = Field(..., description="Array of convention objects.") | |
| # plant_type: List[Union[PlantTypeSpecies, PlantTypeVariety]] = Field(..., description="Array of plant type objects") | |
| # season: List[Season] = Field(..., description="Array of season objects") | |
| #class PlantAttributes(BaseModel): | |
| # name: str = Field(..., title="Name", max_length=255, description="The name of the asset.") | |
| # status: Literal["active"] = Field(..., description="Constant value indicating status.") | |
| # archived: Optional[str] = Field(None, title="Timestamp", description="The time the asset was archived.") | |
| # data: Optional[str] = Field(None, title="Data") | |
| # notes: NoteValue = Field(..., title="Notes") | |
| # flag: List[str] = Field(..., title="Flags") | |
| # id_tag: List[IDTag] = Field(..., title="ID tags") | |
| # inventory: List[InventoryItem] = Field(..., title="Current inventory") | |
| # geometry: Optional[Dict] = Field(None) | |
| # intrinsic_geometry: Optional[IntrinsicGeometry] = Field(None, title="Intrinsic geometry") | |
| # is_location: bool = Field(..., title="Is location") | |
| # is_fixed: bool = Field(..., title="Is fixed") | |
| # surveystack_id: Optional[str] = Field(None, title="Surveystack ID") | |
| class PlantAttributes(BaseModel): | |
| name: str = Field(..., title="Name", description="The name of the asset.") | |
| status: Literal["active"] = Field(..., description="Constant value indicating status.") | |
| notes: str = Field(..., title="Notes") | |
| class PlantingAsset(BaseModel): | |
| id: str = Field(..., description="A valid UUID") | |
| type: Literal["asset--plant"] = Field(..., description="Constant value indicating type of asset.") | |
| attributes: PlantAttributes | |
| class LogAttributes(BaseModel): | |
| name: str = Field(..., description="Name of the log activity.") | |
| timestamp: str = Field(..., description="Timestamp of the event being logged.") | |
| status: Literal["done"] = Field(..., description="Constant value indicating status.") | |
| notes: Optional[str] = Field(..., title="Notes") | |
| class LogRelationships(BaseModel): | |
| asset: PlantingAsset | |
| class LogActivity(BaseModel): | |
| id: str = Field(..., description="A valid UUID") | |
| attributes: LogAttributes | |
| relationships: LogRelationships | |
| class FlamingConvention(BaseModel): | |
| pass | |
| def generate_json(specification): | |
| """ | |
| Function to prompt OpenAI API to generate structured JSON output. | |
| """ | |
| try: | |
| # Call OpenAI API to generate structured output based on prompt | |
| response = client.beta.chat.completions.parse( | |
| model="gpt-4o-2024-08-06", # Use GPT model that supports structured output | |
| messages=[ | |
| {"role": "system", "content": "Extract the farm management information."}, | |
| {"role": "user", "content": specification} | |
| ], | |
| response_format=LogActivity, | |
| ) | |
| generated_json = response.choices[0].message.parsed | |
| print(generated_json) # debugging | |
| pretty_json = generated_json.json() | |
| if 'error' in response: | |
| raise ValueError(f"API error: {response['error']['message']}") | |
| return pretty_json | |
| except ValidationError as e: | |
| return {"error": str(e)} | |
| except Exception as e: | |
| return {"error": "Failed to generate valid JSON. " + str(e)} | |
| def validate_json(json_schema): | |
| # This method attempts to validate the created json from the prompt. | |
| # It may be the case that we don't actually need this using chatgpt magic | |
| pass | |
| def process_specifications(data): | |
| # This method just drives the process | |
| resulting_schema = generate_json(data) | |
| return resulting_schema | |
| demo = gr.Interface( | |
| fn=process_specifications, | |
| inputs=[gr.Textbox(label="Necessary Values: Plant Asset ID, Plant asset name, Plant Notes, Activity Name, Activity ID, Activity Timestamp, Activity Notes")], | |
| outputs=[gr.Textbox(label="JSON Data Output")], | |
| title="JSON Schema Crafting Experiment", | |
| description="Input your specification, receive the schema and payload!", | |
| allow_flagging="never") | |
| if __name__ == "__main__": | |
| demo.launch() |