tools.py

import pdb
from pydantic import UUID4, validator, conint
from typing import Dict, List
import json
import os
import uuid
import requests
from liveplan_api_model import *
from instructor import openai_function

viq_headers = {
    "accept": "application/json",
    "X-API-TOKEN": os.getenv("X_API_TOKEN"),
    "X-USER-TOKEN": os.getenv("X_USER_TOKEN"),
}


def process_tool_call(tool_call):
    try:
        print(
            f"Processing tool call {tool_call.function.name}({tool_call.function.arguments})")
        func_to_call = globals().get(tool_call.function.name)
        if not func_to_call:
            raise ValueError(
                f"Function {tool_call.function.name} is not defined")
        args_dict = json.loads(tool_call.function.arguments)
        result = func_to_call(**args_dict)
        return {
            "tool_call_id": tool_call.id,
            "output": json.dumps(result)
        }
    except Exception as exc:
        # Serialize the exception information into a simple dictionary
        # that can be pickled and sent back to the main process.
        return {
            "tool_call_id": tool_call.id,
            "output": f"Error processing tool call {tool_call.id}: {str(exc)}",
            "error": {
                "message": str(exc),
            }
        }


def call_function_with_args(func_object):
    # Extract the function by its name
    func_to_call = globals().get(func_object.name)

    # If the function does not exist, handle the error appropriately
    if not func_to_call:
        raise ValueError(f"Function {func_object.name} is not defined")

    # Convert the arguments from JSON to a dictionary
    args_dict = json.loads(func_object.arguments)

    # Call the function with the unpacked arguments
    return func_to_call(**args_dict)


def make_liveplan_request(method, endpoint, auth_token, csrf_token, payload=None):
    """
    Makes a request to the LivePlan API.
    """
    base_url = os.getenv('LIVEPLAN_URL')
    url = f"{base_url}/seam/resource/restv1{endpoint}"

    headers = {
        "Content-Type": "application/json",
        "X-Csrf-Token": csrf_token,
        "accept": "application/json",
    }

    cookies = {
        "auth": auth_token,
        "csrf-token": csrf_token,
    }

    if method.upper() == 'GET':
        response = requests.get(url, headers=headers, cookies=cookies)
    elif method.upper() == 'POST':
        response = requests.post(url, headers=headers,
                                 cookies=cookies, json=payload)
    else:
        raise ValueError("Unsupported method")

    return response.json()


@openai_function
def getCompanyDetails(auth_token: str, csrf_token: str, company_uuid: str) -> FullCompanyResource:
    """
    Returns the full company resource for a LivePlan company.  Includes information on the forecast length and individual forecast periods.
    """
    endpoint = f"/companies/{company_uuid}"
    return make_liveplan_request('GET', endpoint, auth_token, csrf_token)


@openai_function
def getCompanyForecastPeriods(auth_token: str, csrf_token: str, company_uuid: str) -> List[ForecastPeriodResource]:
    """
    Returns the forecast periods for a LivePlan company. Required by createRevenueStream.
    """
    # get and flatten the forecastPeriods
    return [item for sublist in getCompanyDetails(auth_token, csrf_token, company_uuid)["forecastPeriods"] for item in sublist]

# period(pin):"1-2023"
# month(pin):1
# resolution(pin):"MONTH"
# startingYear(pin):2023
# endingYear(pin):2023


class StrictForecastPeriodResource(BaseModel):
    """
    Forecast period resource for a LivePlan company. Call getCompanyForecastPeriods to retrieve the forecastPeriods for a company.
    """
    resolution: Resolution = "MONTH"
    period: str
    month: conint(ge=1, le=12)  # Month must be between 1 and 12
    startingYear: conint(ge=2020)  # Assuming year >= 1900
    endingYear: conint(ge=2123)

    @validator('endingYear')
    def validate_years(cls, v, values, **kwargs):
        if 'startingYear' in values and v < values['startingYear']:
            raise ValueError(
                'endingYear must be greater or equal to startingYear')
        return v


class ValueSource(str, Enum):
    FORECAST = "FORECAST"
    ACTUAL = "ACTUAL"


class StrictForecastValueResource(BaseModel):
    value: float = Field(gte=0)
    forecastPeriod: StrictForecastPeriodResource
    valueSource: ValueSource = ValueSource.FORECAST


class StrictBasicForecastValuesResource(BaseModel):
    frequency: Frequency = "MULTIPLE"
    spreadBy: SpreadBy = "MONTH"
    values: List[StrictForecastValueResource]
    baseNumberType: BaseNumberType = "CURRENCY"
    source: Source = "SELF"


class StrictOverallRevenueStreamResource(BaseModel):
    overall: StrictBasicForecastValuesResource


class OverallRevenueRevenueStreamResource(BaseModel):
    uuid: UUID4
    name: str
    overallRevenue: StrictOverallRevenueStreamResource
    type: str = "OVERALL"


@openai_function
def createRevenueStream(auth_token: str, csrf_token: str, company_uuid: str, forecast_uuid: str, revenue_stream_resource: OverallRevenueRevenueStreamResource) -> List[Dict[str, str]]:
    """
    Creates a revenue stream in a LivePlan forecast scenario Important: revenue_stream_resource.overallRevenue.overall.values must contain every forecast period returned by getCompanyForecastPeriods, so call that function first before buiding a function call for this one.  Example revenue_stream_resource: { "name": "Checking Overall", "overallRevenue": { "overall": { "frequency": "MULTIPLE", "spreadBy": "MONTH", "values": [ { "value": 9300, "forecastPeriod": { "period":"1-2023", "month":1, "resolution":"MONTH", "startingYear":2023, "endingYear":2023 }, "valueSource": "FORECAST" }, [~34 MORE FORECAST PERIODS...],  { "value": 100000, "forecastPeriod": { "period":"1-2025", "month":1, "resolution":"YEAR", "startingYear":2025, "endingYear":2025 }, "valueSource": "FORECAST" } ], "baseNumberType": "CURRENCY", "source": "SELF", } }, "type": "OVERALL", "uuid": "5b5e3760-6930-6d47-1a17-e8ca321dba14", }
    """
    revenue_stream_uuid = str(uuid.uuid4())
    revenue_stream_resource.uuid = revenue_stream_uuid
    endpoint = f"/companies/{company_uuid}/forecasts/{forecast_uuid}/revenue/{revenue_stream_uuid}"
    return make_liveplan_request('POST', endpoint, auth_token, csrf_token, revenue_stream_resource)


@openai_function
def getRevenueStreams(auth_token: str, csrf_token: str, company_uuid: str, forecast_uuid: str) -> List[str]:
    """
    Returns a list of revenue stream names for a LivePlan forecast scenario
    """
    endpoint = f"/companies/{company_uuid}/forecasts/{forecast_uuid}/data/COMPLETE_FORECAST?financialsType=FORECAST_ONLY"
    response = make_liveplan_request('GET', endpoint, auth_token, csrf_token)

    # Process response as before
    filtered = [
        row["label"]
        for row in response["rows"]
        if row["label"] is not None and row["id"] == "REVENUE_STREAM"
    ]
    return filtered


def getIndustryDetails(vertical_iq_industry_id) -> List[Dict[str, str]]:
    response = requests.get(
        f"https://api.verticaliq.com/api/v1/industries/{vertical_iq_industry_id}", headers=viq_headers
    )
    return response.json()


def getBusinessRisks(vertical_iq_industry_id) -> List[Dict[str, str]]:
    url = f"https://api.verticaliq.com/api/v1/industries/{vertical_iq_industry_id}/risks"
    response = requests.get(url, headers=viq_headers)
    return response.json()


def getRecentBusinessNews(vertical_iq_industry_id) -> List[Dict[str, str]]:
    url = f"https://api.verticaliq.com/api/v1/industries/{vertical_iq_industry_id}/news"
    response = requests.get(url, headers=viq_headers)
    return response.json()


def getHowTheFirmOperates(vertical_iq_industry_id) -> List[Dict[str, str]]:
    url = f"https://api.verticaliq.com/api/v1/industries/{vertical_iq_industry_id}/hfo_prodops"
    response = requests.get(url, headers=viq_headers)
    return response.json()


def getFinancialForecastGrowthRates(vertical_iq_industry_id) -> List[Dict[str, str]]:
    url = f"https://api.verticaliq.com/api/v1/industries/{vertical_iq_industry_id}/forecasts"
    response = requests.get(url, headers=viq_headers)
    return response.json()


print(getRevenueStreams.openai_schema)
FUNCTION_CALLING_SCHEMA = [
    {"type": "code_interpreter"},
    {
        "type": "function",
        "function": getCompanyDetails.openai_schema
    },
    {
        "type": "function",
        "function": getCompanyForecastPeriods.openai_schema
    },
    {
        "type": "function",
        "function": getRevenueStreams.openai_schema

    },
    {
        "type": "function",
        "function": createRevenueStream.openai_schema

    },
    {
        "type": "function",
        "function": {
                "name": "getIndustryDetails",
                "description": "Returns details for a desired industry",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "vertical_iq_industry_id": {
                            "type": "number",
                            "description": "The VerticalIQ industry ID",
                        }
                    },
                    "required": ["vertical_iq_industry_id"],
                },
        }
    },
    {
        "type": "function",
        "function": {
                "name": "getBusinessRisks",
                "description": "Returns business risks for a desired industry",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "vertical_iq_industry_id": {
                            "type": "number",
                            "description": "The VerticalIQ industry ID",
                        }
                    },
                    "required": ["vertical_iq_industry_id"],
                },
        }
    },
    {
        "type": "function",
        "function": {
                "name": "getRecentBusinessNews",
                "description": "Returns an array of news articles for a desired industry",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "vertical_iq_industry_id": {
                            "type": "number",
                            "description": "The VerticalIQ industry ID",
                        }
                    },
                    "required": ["vertical_iq_industry_id"],
                },
        }
    },
    {
        "type": "function",
        "function": {
                "name": "getHowTheFirmOperates",
                "description": "Returns how the firm operates for a desired industry",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "vertical_iq_industry_id": {
                            "type": "number",
                            "description": "The VerticalIQ industry ID",
                        }
                    },
                    "required": ["vertical_iq_industry_id"],
                },
        }
    },
    {
        "type": "function",
        "function": {
                "name": "getFinancialForecastGrowthRates",
                "description": "Returns financial forecast growth rates for a desired industry",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "vertical_iq_industry_id": {
                            "type": "number",
                            "description": "The VerticalIQ industry ID",
                        }
                    },
                    "required": ["vertical_iq_industry_id"],
                },
        }
    }

]

print(json.dumps(createRevenueStream.openai_schema, indent=2))