functions/data_extractor/emanufac_tdabc_extractor_class.py

import pandas as pd
import requests
import numpy as np
import pickle


class EManufacTDABCExtractor:
    def __init__(self, api_url, api_key, profile_id, costed_place):
        self.api_url = api_url
        self.api_key = api_key
        self.process_df = pd.DataFrame()
        self.material_usage_df = pd.DataFrame()
        self.original_material_usage_df = pd.DataFrame()
        self.original_employee_usage_df = pd.DataFrame()
        self.original_process_df = pd.DataFrame()
        self.capital_cost_df = pd.DataFrame()
        self.employee_usage_df = pd.DataFrame()
        self.original_capital_cost_df = pd.DataFrame()
        self.profile_id = profile_id  # Profile For TDABC
        self.original_procedure_profile_id = ""  # Profile For Factory Current  Method
        self.profile_element_name_for_material = "ต้นทุนวัตถุดิบ"
        self.costed_place = costed_place
        self.running_no_start = ""
        self.running_no_end = ""

    def set_running_no_margin(self, start, end):
        self.running_no_start = start
        self.running_no_end = end
        print("Setting Successfully")

    def fetch_material_usage(self, start_date, end_date, limit, page=1):
        url = f"{self.api_url}/cost-estimation/on-type"
        headers = {
            "Accept": "application/json",
            "Authorization": f"Bearer {self.api_key}",
        }
        querystring = {
            "startDate": start_date,
            "endDate": end_date,
            "size": limit,
            "page": page,
            "profile": self.profile_id,
            "elementType": "MATERIAL",
            "placeRestricNotify": "true",
            "correctPlaceOnly": "true",
            "extractOriginalLot": "true",
            "place": self.costed_place,
            "specifyProfileElement": self.profile_element_name_for_material,
            "runningNoStart": self.running_no_start,
            "runningNoEnd": self.running_no_end,
        }
        response = requests.get(url, headers=headers, params=querystring)
        try:
            material_data = response.json()["rows"]
        except Exception as e:
            print("Error in fetch material", e)
            material_data = []

        self.material_usage_df = pd.DataFrame(material_data)
        self.original_material_usage_df = pd.DataFrame(material_data)

        with open("material.pickle", "wb") as handle:
            pickle.dump(self.material_usage_df, handle)

        return (self.material_usage_df,)

    def fetch_employee_usage(self, start_date, end_date, limit, page=1):
        url = f"{self.api_url}/cost-estimation/on-type"
        headers = {
            "Accept": "application/json",
            "Authorization": f"Bearer {self.api_key}",
        }
        querystring = {
            "startDate": start_date,
            "endDate": end_date,
            "size": limit,
            "page": page,
            "profile": self.profile_id,
            "elementType": "LABOR",
            "placeRestricNotify": "true",
            "correctPlaceOnly": "true",
            "place": self.costed_place,
            "merged": "true",
            "runningNoStart": self.running_no_start,
            "runningNoEnd": self.running_no_end,
        }
        response = requests.get(url, headers=headers, params=querystring)
        try:
            employee_data = response.json()["rows"]
        except Exception as e:
            print("Error in fetch employee", e)
            employee_data = []

        self.employee_usage_df = pd.DataFrame(employee_data)
        self.original_employee_usage_df = pd.DataFrame(employee_data)

        with open("employee.pickle", "wb") as handle:
            pickle.dump(self.employee_usage_df, handle)

        return (self.employee_usage_df,)

    def fetch_capital_cost_usage(self, start_date, end_date, limit, page=1):
        url = f"{self.api_url}/cost-estimation/on-type"
        headers = {
            "Accept": "application/json",
            "Authorization": f"Bearer {self.api_key}",
        }
        querystring = {
            "startDate": start_date,
            "endDate": end_date,
            "size": limit,
            "page": page,
            "profile": self.profile_id,
            "elementType": "CAPITAL_COST",
            "placeRestricNotify": "true",
            "correctPlaceOnly": "true",
            "place": self.costed_place,
            "merged": "true",
            "splitCostDriver": "true",
            "runningNoStart": self.running_no_start,
            "runningNoEnd": self.running_no_end,
        }
        response = requests.get(url, headers=headers, params=querystring)
        try:
            capital_cost_data = response.json()["rows"]
        except Exception as e:
            print("Error in fetch capital cost", e)
            capital_cost_data = []

        self.capital_cost_df = pd.DataFrame(capital_cost_data)
        self.original_capital_cost_df = pd.DataFrame(capital_cost_data)

        with open("capital.pickle", "wb") as handle:
            pickle.dump(self.capital_cost_df, handle)

        return (self.capital_cost_df,)

    def change_profile_element_for_material(self, new_element_name):
        self.profile_element_name_for_material = new_element_name
        print("Success Changing")

    # Profile For Current Factory Cost Estimation Method
    def change_profile_for_original_procedure(self, profile_id):
        self.original_procedure_profile_id = profile_id
        print("Success Changing")

    # Get Data From Current Factory Cost Estimation Method as a Reference or Result
    # Of our new Profile using TDABC
    def fetch_process_data(self, start_date, end_date, limit, page=1):
        url = f"{self.api_url}/cost-estimation"
        headers = {
            "Accept": "application/json",
            "Authorization": f"Bearer {self.api_key}",
        }
        querystring = {
            "startDate": start_date,
            "endDate": end_date,
            "size": limit,
            "page": page,
            "profile": self.original_procedure_profile_id,
            "hideResultList": "true",
            "placeRestricNotify": "true",
            "correctPlaceOnly": "true",
            "runningNoStart": self.running_no_start,
            "runningNoEnd": self.running_no_end,
            "place": self.costed_place,
        }
        response = requests.get(url, headers=headers, params=querystring)
        process_data = response.json()["rows"]

        self.process_df = pd.DataFrame(process_data)
        self.original_process_df = pd.DataFrame(process_data)

        with open("process.pickle", "wb") as handle:
            pickle.dump(self.process_df, handle)

        return (self.process_df,)

    def load_from_pickle(self):

        try:
            with open("material.pickle", "rb") as handle:
                material_usage_df = pickle.load(handle)
                self.material_usage_df = material_usage_df
                self.original_material_usage_df = material_usage_df
        except:
            print("Error loading material Pickle")

        try:
            with open("employee.pickle", "rb") as handle:
                employee_usage_df = pickle.load(handle)
                self.employee_usage_df = employee_usage_df
                self.original_employee_usage_df = employee_usage_df
        except:
            print("Error loading employee Pickle")

        try:
            with open("capital.pickle", "rb") as handle:
                capital_cost_df = pickle.load(handle)
                self.capital_cost_df = capital_cost_df
                self.original_capital_cost_df = capital_cost_df
        except:
            print("Error loading capital Pickle")

        try:
            with open("process.pickle", "rb") as handle:
                process_df = pickle.load(handle)
                self.process_df = process_df
                self.original_process_df = process_df
        except:
            print("Error loading Process Pickle")

        # self.capital_cost_df = capital_cost_df
        print("Loaded from pickle Successfully")

    def get_process_list(self):
        return self.process_df

    def get_material_usage(self):
        return self.material_usage_df

    def get_employee_usage(self):
        return self.employee_usage_df

    def get_capital_cost(self):
        return self.capital_cost_df

    def load_material_usage(self, material_usage_df):
        self.material_usage_df = material_usage_df

    def load_employee_usage(self, employee_usage_df):
        self.employee_usage_df = employee_usage_df

    def load_capital_cost(self, capital_cost_df):
        self.capital_cost_df = capital_cost_df

    def load_process(self, process_df):
        self.process_df = process_df

    def load_original_process(self, process_df):
        self.original_process_df = process_df

    def load_original_material_usage(self, material_usage_df):
        self.original_material_usage_df = material_usage_df

    def load_original_employee_usage(self, employee_usage_df):
        self.original_employee_usage_df = employee_usage_df

    def load_original_capital_cost(self, capital_cost_df):
        self.original_capital_cost_df = capital_cost_df

    def adjust_material_usage(self):
        new_material_usage_df = pd.DataFrame()  # self.material_usage_df.copy()
        new_material_usage_df["_id"] = self.original_material_usage_df["material_id"]
        new_material_usage_df["process_id"] = self.original_material_usage_df[
            "process_id"
        ]
        new_material_usage_df["name"] = self.original_material_usage_df["material_name"]
        new_material_usage_df["amount"] = self.original_material_usage_df[
            "used_quantity"
        ]
        new_material_usage_df["unit_cost"] = self.original_material_usage_df[
            "unit_cost"
        ]
        # TODO:  Update in EManufac Code to pick the purchase date instead
        new_material_usage_df["date"] = self.original_material_usage_df["used_date"]
        self.material_usage_df = new_material_usage_df

    def adjust_employee_usage(self):
        new_employee_usage_df = pd.DataFrame()
        new_employee_usage_df["_id"] = self.original_employee_usage_df[
            "artifact_employee_id"
        ]
        new_employee_usage_df["employee_id"] = self.original_employee_usage_df[
            "artifact_employee_id"
        ]
        new_employee_usage_df["process_id"] = self.original_employee_usage_df[
            "process_id"
        ]
        new_employee_usage_df["employee_name"] = self.original_employee_usage_df[
            "artifact_employee_name"
        ]
        new_employee_usage_df["amount"] = self.original_employee_usage_df[
            "average_labor_amount"
        ]
        new_employee_usage_df["date"] = self.original_employee_usage_df["receipt_date"]

        # If more than 1 employee (unit cost is same) we group to one, and sum the duration
        new_employee_usage_df["duration"] = (
            self.original_employee_usage_df["artifact_minute_use"]
            * self.original_employee_usage_df["average_labor_amount"]
        )

        # new_employee_usage_df["type"]
        new_employee_usage_df["type"] = "daily"
        new_employee_usage_df['day_amount'] = 1
        try:
            new_employee_usage_df["cost"] = self.original_employee_usage_df[
                "average_daily_labor_cost"
            ]
        except:
            new_employee_usage_df["cost"] = 0

        new_employee_usage_df = new_employee_usage_df.dropna(subset=["cost"])

        self.employee_usage_df = new_employee_usage_df

    def adjust_capital_cost(self):
        new_capital_cost_df = pd.DataFrame()
        try:
            new_capital_cost_df["_id"] = self.original_capital_cost_df[
                "artifact_cost_title"
            ]
            new_capital_cost_df["process_id"] = self.original_capital_cost_df[
                "process_id"
            ]
            new_capital_cost_df["name"] = self.original_capital_cost_df[
                "artifact_cost_title"
            ]
            new_capital_cost_df["cost"] = self.original_capital_cost_df[
                "artifact_capital_cost"
            ]
            new_capital_cost_df["day_amount"] = self.original_capital_cost_df[
                "average_day_amount"
            ]
            new_capital_cost_df["hour_amount"] = self.original_capital_cost_df[
                "average_hour_amount"
            ]
            new_capital_cost_df["unit_cost"] = self.original_capital_cost_df[
                "artifact_unit_cost"
            ]
            new_capital_cost_df["duration"] = self.original_capital_cost_df[
                "artifact_used_time"
            ]
            new_capital_cost_df["date"] = self.original_capital_cost_df["receipt_date"]

            # new_capital_cost_df["machine_hour"] = new_capital_cost_df["machineHour"]
            # new_capital_cost_df["life_time"] = new_capital_cost_df["lifeTime"]
            # new_capital_cost_df["day_per_month"] = new_capital_cost_df["dayPerMonth"]

            zero_cost = new_capital_cost_df[new_capital_cost_df["cost"] == 0]
            # Remove None machine usage
            new_capital_cost_df = new_capital_cost_df.drop(zero_cost.index)
            self.capital_cost_df = new_capital_cost_df
        except Exception as e:
            print("Error in adjust capital cost", e)

    def adjust_process_df(self):
        temp_process_df = self.original_process_df.copy()
        temp_process_df = temp_process_df[temp_process_df["cost"] > 0]

        self.process_df = temp_process_df

    def save_material_csv(self, destination_folder_path="generated"):
        self.material_usage_df.to_csv(
            f"{destination_folder_path}/generated_material_usage.csv"
        )
        self.original_material_usage_df.to_csv(
            f"{destination_folder_path}/original_material_usage.csv"
        )

    def save_process_csv(self, destination_folder_path="generated"):
        self.process_df.to_csv(
            f"{destination_folder_path}/generated_process_data.csv")
        self.original_process_df.to_csv(
            f"{destination_folder_path}/original_process_data.csv"
        )

    def save_employee_csv(self, destination_folder_path="generated"):
        self.employee_usage_df.to_csv(
            f"{destination_folder_path}/generated_employee_usage.csv"
        )
        self.original_employee_usage_df.to_csv(
            f"{destination_folder_path}/original_employee_usage.csv"
        )

    def save_capital_csv(self, destination_folder_path="generated"):
        self.capital_cost_df.to_csv(
            f"{destination_folder_path}/generated_captial_cost.csv"
        )
        self.original_capital_cost_df.to_csv(
            f"{destination_folder_path}/original_captial_cost.csv"
        )

    def save_csv(self, destination_folder_path="generated"):
        self.save_process_csv(destination_folder_path)
        self.save_material_csv(destination_folder_path)
        self.save_employee_csv(destination_folder_path)
        self.save_capital_csv(destination_folder_path)