Source code for examples.provencia.schemas

from typing import Final

import numpy as np
import pandas as pd
import pandera.pandas as pa
from pandera.typing.pandas import Index, Series

from sara.oar import OARSchema




class ProvenciaETLSchema(pa.DataFrameModel):
    """Describe Schema for Provencia ETL dataframe."""

    delivered: Series[pd.Int32Dtype] = pa.Field(
        metadata={
            "description": "Quantity of product that has been delivered "
            "in shop in the morning"
        }
    )
    purchase_price: Series[pd.Int32Dtype] = pa.Field(
        metadata={"description": "Price of the product in the shop in €/100"}
    )
    ordered: Series[pd.Int32Dtype] = pa.Field(
        metadata={"description": "Quantity of product ordered in shop in the morning"}
    )
    total_paid: Series[pd.Float64Dtype] = pa.Field(
        metadata={"description": "How much money has been paid for the order"}
    )
    sold: Series[pd.Int32Dtype] = pa.Field(
        metadata={"description": "The quantity of product sold in the day"}
    )
    gross: Series[pd.Int32Dtype] = pa.Field(
        metadata={
            "description": "How much money has been received "
            "for the sales of the day in €/100"
        }
    )
    stock_evening: Series[pd.Int32Dtype] = pa.Field(
        metadata={
            "description": "Estimated stock of product in shop " "at the end of the day"
        }
    )

    store_id: Index[pd.StringDtype] = pa.Field(
        metadata={"description": "The store id is 2 uppercase letters"}
    )
    product_code: Index[pd.Int64Dtype] = pa.Field(
        metadata={"description": "The product code is an positive integer"}
    )
    date: Index[np.datetime64] = pa.Field(
        metadata={"description": "The date of the day"}
    )

    class Config:
        metadata: Final[dict] = {
            "process": (
                "0. In store `store_id`, for product `product_code`, at day `date`.\n"
                "1. In the morning, `delivered` products has been received.\n"
                "   They have been ordered yesterday.\n"
                "2. Manager then acknowledge the new price `purchase_price`.\n"
                "3. Manager then order `ordered` products.\
                They will be delivered tomorrow.\n"
                "4. Manager pays `total_paid` for its order.\n"
                "6. In the evening, it has been sold `sold` products during the day."
                "7. These sales bring `gross` money."
                "8. There are `stock_evening` products left in stock"
            )
        }





class ProvenciaEnvSchema(OARSchema):
    """Pandera OAR Schema for provencia supply chain environment."""

    sold_yd: Series[pd.Int32Dtype] = pa.Field(
        alias=("obs0", "sold_yd"),
        metadata={"description": "Quantity of product sold in shop during yesterday"},
    )
    gross_yd: Series[pd.Int32Dtype] = pa.Field(
        alias=("obs0", "gross_yd"),
        metadata={
            "description": "Income in €/100 for selling product in shop during yesterday"
        },
    )
    stock_after_delivery: Series[pd.Int32Dtype] = pa.Field(
        alias=("obs0", "stock_after_delivery"),
        metadata={
            "description": "Estimated stock of product in shop right after delivery"
        },
    )
    delivered: Series[pd.Int32Dtype] = pa.Field(
        alias=("obs0", "delivered"),
        metadata={
            "description": (
                "Quantity of product that has been delivered in shop " "in the morning"
            )
        },
    )
    purchase_price: Series[pd.Int32Dtype] = pa.Field(
        alias=("obs0", "purchase_price"),
        metadata={"description": "Unitary (PI or KG) buying price of product"},
    )
    ordered: Series[pd.Int32Dtype] = pa.Field(
        alias=("act0", "ordered"),
        metadata={"description": "Quantity of product ordered in shop in the morning"},
    )
    profit: Series[pd.Float64Dtype] = pa.Field(
        alias=("rew1", "profit"),
        metadata={
            "description": (
                "The profit is sales at day (+lag) in shop " "minus the ordering cost"
            )
        },
    )

    store_id: Index[pd.StringDtype] = pa.Field(
        metadata={"description": "The store id is 2 uppercase letters"}
    )
    product_code: Index[pd.Int64Dtype] = pa.Field(
        metadata={"description": "The product code is an positive integer"}
    )
    date: Index[np.datetime64] = pa.Field(
        metadata={"description": "The date of the day"}
    )

    class Config:
        metadata: Final[dict] = {
            "description": (
                "Créé en 1963 à Annecy, le groupe Provencia est aujourd hui un "
                "acteur majeur de la grande distribution en Rhône Alpes. "
                "Cet environnement modélise la commande de produits pour le rayon "
                "boucherie."
            ),
            "process": (
                "0. In store `store_id`, for product `product_code`, at day `date`. "
                "1. Agent observes 'obs0' and selects action `act0`. "
                "2. Its action immediately yields reward `rew1`."
            ),
        }