cbonesana

"""Execute as a normal script, or subdivided using uvicorn:

Together:
    python streams.py

GENERATOR:
    uvicorn idk:app_gen --host localhost --port 8000
PROXY:
    uvicorn idk:app_proxy --host localhost --port 7000
CLIENT:

cbonesana

"""Launch this script directly or launche the three components separately using uvicorn."""
from typing import AsyncGenerator, Any

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from fastapi.requests import Request
from httpx import AsyncClient
from pydantic import BaseModel

import asyncio

cbonesana

# %%
from openai import AsyncOpenAI
from openai.types.chat import ChatCompletion
from dotenv import load_dotenv
from tqdm.asyncio import tqdm as atqdm

import asyncio
import os
import nest_asyncio
import yaml

cbonesana

  version: 1
  disable_existing_loggers: true
  formatters:
    standard_formatter:
      format: "%(asctime)s %(levelname)7s %(name)s: %(message)s"
      datefmt: "%Y-%m-%d %H:%M:%S"
  handlers:
    console:
      formatter: standard_formatter

cbonesana

from numpy.typing import NDArray
from pydantic import BaseModel, BeforeValidator, ConfigDict, PlainSerializer
from typing_extensions import Annotated

import json
import numpy as np


# define serializers and validator
def nd_array_custom_before_validator(x: list[float]) -> NDArray[np.float32]:

cbonesana

FROM python:3.12.3

RUN pip install \
    pandas \
    scikit-learn \
    fastapi \
    aiofiles

cbonesana

from typing import Any
from pydantic import BaseModel, root_validator
import json

# this dictionary will collect all classes constructors
class_registry: dict[str, Any] = dict()


# base class
class Creature(BaseModel):

cbonesana

from sqlalchemy import create_engine

from sqlalchemy import select, insert, func

from tables import Base, Record

from time import time

import ray
import uuid

cbonesana

from multiprocessing import Queue, Process

import random
import time
import signal


stop_sentinel = "STOP"

cbonesana

from pydantic import BaseModel
from enum import Enum

import json


class Season(str, Enum):
    SPRING = 'spring'
    SUMMER = 'summer'
    AUTUMN = 'autumn'