Download model/tdce_model.py from iaecpsu-1/tdce-basic: direct link, hf CLI and curl.
- Browser
- Download file 22.4 kB
-
https://huggingface.co/iaecpsu-1/tdce-basic/resolve/fb514b0d05d90abb02c8c1479cbc548f5ea71d8a/model/tdce_model.py
- Command line
-
hf download hf://iaecpsu-1/tdce-basic@fb514b0d05d90abb02c8c1479cbc548f5ea71d8a/model/tdce_model.py
-
curl -L -o tdce_model.py https://huggingface.co/iaecpsu-1/tdce-basic/resolve/fb514b0d05d90abb02c8c1479cbc548f5ea71d8a/model/tdce_model.py
22.4 kB
| import numpy as np | |
| import random | |
| import pandas as pd | |
| import importlib | |
| import time | |
| import material_network as mn | |
| import time_driven_network as tdn | |
| import loss | |
| import normalize as norm | |
| import matrix_normalization as mnorm | |
| import sample_payload_adjustment as spa | |
| import weight_activation as wa | |
| importlib.reload(mn) | |
| importlib.reload(tdn) | |
| importlib.reload(loss) | |
| importlib.reload(norm) | |
| importlib.reload(mnorm) | |
| importlib.reload(spa) | |
| importlib.reload(wa) | |
| # Generated from Gemini | |
| def generate_random_numbers(): | |
| # Generate a random number between 0 and 1 (exclusive) | |
| random_val = random.random() | |
| number1 = random_val | |
| # The second number is another random value between 0 and (1 - number1) | |
| number2 = random.random() * (1 - number1) | |
| # The third number is simply 1 minus the sum of the first two | |
| number3 = 1 - number1 - number2 | |
| return np.array([number1, number2, number3]).reshape(3, 1) | |
| # Model Reference | |
| # Early Stopping https://medium.com/@juanc.olamendy/understanding-early-stopping-a-key-to-preventing-overfitting-in-machine-learning-17554fc321ff | |
| class TDCEModel: | |
| def __init__(self): | |
| self.material_element = mn.MaterialNetwork() | |
| self.employee_element = tdn.TimeDrivenNetwork() | |
| self.capital_cost_element = tdn.TimeDrivenNetwork() | |
| random_weight = np.array([1.0, 1.0, 1.0]).reshape( | |
| 3, 1 | |
| ) # generate_random_numbers() | |
| self.weights = random_weight | |
| self.bias = np.full((1, 1), 0.0) | |
| self.loss = loss.mse | |
| self.loss_prime = loss.mse_prime | |
| self.loss_percent = loss.rmspe | |
| self.gradient = np.array([[0, 0, 0]]) | |
| self.prediction_error = np.array([[0, 0, 0]]) | |
| self.weight_list = np.array([[[0], [0], [0]]]) | |
| self.epoch_error = [] | |
| self.sample_errors = [] | |
| self.material_learning_rate = 0.001 | |
| self.employee_learning_rate = 0.001 | |
| self.capital_cost_learning_rate = 0.001 | |
| self.max_data = {} | |
| self.min_data = {} | |
| self.sample_payload = [] | |
| self.use_early_stopping = False | |
| self.patience = 10 | |
| self.use_model_weight = False | |
| def inital_inside_element( | |
| self, | |
| material_layer, | |
| employee_layer, | |
| capital_cost_layer, | |
| ): | |
| self.material_element.add(material_layer) | |
| self.material_element.use(loss.mse, loss.mse_prime) | |
| self.employee_element.add(employee_layer) | |
| self.employee_element.use(loss.mse, loss.mse_prime) | |
| self.capital_cost_element.add(capital_cost_layer) | |
| self.capital_cost_element.use(loss.mse, loss.mse_prime) | |
| print("Initial Successfully") | |
| # For Setting New Network Element | |
| def set_material_element(self, material_network): | |
| self.material_element = material_network | |
| def set_employee_element(self, employee_element): | |
| self.employee_element = employee_element | |
| def set_capital_cost_element(self, capital_cost_element): | |
| self.capital_cost_element = capital_cost_element | |
| def use(self, loss, loss_prime, loss_percent): | |
| self.loss = loss | |
| self.loss_prime = loss_prime | |
| self.loss_percent = loss_percent | |
| def set_learning_rate(self, material_lr, employee_lr, cc_lr): | |
| self.material_learning_rate = material_lr | |
| self.employee_learning_rate = employee_lr | |
| self.capital_cost_learning_rate = cc_lr | |
| print("Learning Rate Set Successfully") | |
| print( | |
| f"Material LL {material_lr}, Employee LL {employee_lr}, Capital Cost LL {cc_lr}" | |
| ) | |
| def activate_early_stopping(self): | |
| self.use_early_stopping = True | |
| print("Activate Early Stopping Successfully") | |
| def deactivate_early_stopping(self): | |
| self.use_early_stopping = False | |
| print("Deactivate Early Stopping Successfully") | |
| def activete_model_weight(self): | |
| self.use_model_weight = True | |
| print("Activate Model Weight Successfully") | |
| def deactivate_model_weight(self): | |
| self.use_model_weight = False | |
| print("Deactivate Model Weight Successfully") | |
| # For Early Stopping | |
| def edit_patience_round(self, patience_round): | |
| self.patience = patience_round | |
| def fit_with_validation( | |
| self, | |
| material_cost_matrix, | |
| material_amount_matrix, | |
| employee_cost_matrix, | |
| employee_duration_matrix, | |
| employee_day_amount_matrix, | |
| capital_cost_matrix, | |
| day_amount_matrix, | |
| capital_cost_duration_matrix, | |
| result_matrix, | |
| epoch, | |
| learning_rate, | |
| validation_payload, | |
| display_round_log=False, | |
| ): | |
| results = [] | |
| epoch_errors = [] | |
| sample_errors = [] | |
| sample_payload = [] | |
| # For Early Stopping | |
| patience_counter = 0 | |
| best_validation_error = np.inf | |
| # print('Material Cost Matrix') | |
| # print(material_cost_matrix) | |
| # print('-----------------------------------') | |
| # print('Material Amount Matrix') | |
| # print(material_amount_matrix) | |
| # print('------------------------------') | |
| # print('Daily Employee Matrix') | |
| # print(daily_employee_cost_matrix) | |
| # print('------------------------------') | |
| # print('Monthly Employee Matrix') | |
| # print(monthly_employee_cost_matrix) | |
| # print('------------------------------') | |
| print(f"W: Initial Weight {self.weights}") | |
| # Duration Represent the amount of process | |
| samples = len(result_matrix) | |
| validate_result_matrix = validation_payload["validate_result_matrix"] | |
| # Some of validate payload use normalized_validate because size is not | |
| # same as train payload if they are in 3D it will use only normalized function | |
| ( | |
| normalized_material_cost, | |
| normalized_material_amount, | |
| normalized_validate_material_cost, | |
| normalized_validate_material_amount, | |
| normalized_employee_cost, | |
| normalized_employee_duration, | |
| normalized_employee_day_amount, | |
| normalized_validate_employee_cost, | |
| normalized_validate_employee_duration, | |
| normalized_validate_employee_day_amount, | |
| normalized_capital_cost, | |
| normalized_capital_cost_duration, | |
| normalized_day_amount, | |
| normalized_validate_capital_cost, | |
| normalized_validate_capital_duration, | |
| normalized_validate_day_amount, | |
| max_data, | |
| min_data, | |
| ) = mnorm.normalize_payload( | |
| material_cost_matrix=material_cost_matrix, | |
| material_amount_matrix=material_amount_matrix, | |
| employee_cost_matrix=employee_cost_matrix, | |
| employee_duration_matrix=employee_duration_matrix, | |
| employee_day_amount_matrix=employee_day_amount_matrix, | |
| capital_cost_matrix=capital_cost_matrix, | |
| day_amount_matrix=day_amount_matrix, | |
| capital_cost_duration_matrix=capital_cost_duration_matrix, | |
| validation_payload=validation_payload, | |
| display_log=False, | |
| ) | |
| self.min_data = min_data | |
| self.max_data = max_data | |
| for i in range(epoch): | |
| error = 0 | |
| sum_error = 0 | |
| sum_error_percent = 0 | |
| sum_validation_error = 0 | |
| sum_validation_error_percent = 0 | |
| start_time = time.time() | |
| validate_sample_amount = 0 | |
| for j in range(samples): | |
| # Result | |
| result_input = result_matrix[j] | |
| # Material | |
| # material_cost_input = material_cost_matrix[j] | |
| material_cost_input = normalized_material_cost[j] | |
| # material_amount_input = material_amount_matrix[j] | |
| material_amount_input = normalized_material_amount[j] | |
| # Employee / Labor | |
| employee_cost_input = normalized_employee_cost[j] | |
| employee_duration_input = normalized_employee_duration[j] | |
| employee_day_amount_input = normalized_employee_day_amount[j] | |
| # Capital Cost | |
| capital_cost_input = normalized_capital_cost[j] | |
| day_amount_input = normalized_day_amount[j] | |
| capital_cost_dur_input = normalized_capital_cost_duration[j] | |
| if j < len(normalized_validate_material_cost): | |
| # VALIDATION INPUT | |
| validate_material_cost_input = normalized_validate_material_cost[j] | |
| validate_material_amount_input = ( | |
| normalized_validate_material_amount[j] | |
| ) | |
| validate_employee_input = ( | |
| normalized_validate_employee_cost[j] | |
| ) | |
| validate_emp_dur_input = ( | |
| normalized_validate_employee_duration[j] | |
| ) | |
| validate_emp_dayamount_input = ( | |
| normalized_validate_employee_day_amount[j] | |
| ) | |
| validate_capital_cost_input = normalized_validate_capital_cost[j] | |
| validate_dayamount_input = normalized_validate_day_amount[j] | |
| validate_capital_cost_dur_input = ( | |
| normalized_validate_capital_duration[j] | |
| ) | |
| validate_result_input = validate_result_matrix[j] | |
| # Train Material ELE | |
| predicted_mc = self.material_element.predict_sample( | |
| cost_input=material_cost_input, amount_input=material_amount_input | |
| ) | |
| # Employee Cost ELE | |
| predicted_ec = self.employee_element.predict_sample( | |
| cost_input=employee_cost_input, | |
| time_input=employee_duration_input, | |
| day_amount=employee_day_amount_input | |
| ) | |
| # Capital Cost ELE | |
| predicted_cc = self.capital_cost_element.predict_sample( | |
| cost_input=capital_cost_input, | |
| time_input=capital_cost_dur_input, | |
| day_amount=day_amount_input, | |
| ) | |
| # For Validation | |
| # Predict will not update class vairable while predict_sample | |
| # which will be update their variable for train | |
| if j < len(normalized_validate_material_cost): | |
| # Test Material ELE | |
| validate_predicted_mc = self.material_element.predict( | |
| cost_input=validate_material_cost_input, | |
| amount_input=validate_material_amount_input, | |
| ) | |
| # Test Monthly Employee Cost ELE | |
| validate_predicted_ec = self.employee_element.predict( | |
| cost_input=validate_employee_input, | |
| time_input=validate_emp_dur_input, | |
| day_amount=validate_emp_dayamount_input | |
| ) | |
| # Test Capital Cost ELE | |
| validate_predicted_cc = self.capital_cost_element.predict( | |
| cost_input=validate_capital_cost_input, | |
| time_input=validate_capital_cost_dur_input, | |
| day_amount=validate_dayamount_input, | |
| ) | |
| validate_result = ( | |
| validate_predicted_mc * self.weights[0] | |
| + validate_predicted_ec * self.weights[1] | |
| + validate_predicted_cc * self.weights[2] | |
| ) + self.bias | |
| # Result Combination and Find Error | |
| result = ( | |
| predicted_mc * self.weights[0] | |
| + predicted_ec * self.weights[1] | |
| + predicted_cc * self.weights[2] | |
| ) + self.bias | |
| # if result < 0: | |
| # result = [[0]] | |
| # print( | |
| # f"predicted_mc {predicted_mc} predicted_dmc {predicted_dmc} predicted_mec{predicted_mec} predicted_cc{predicted_cc}") | |
| # Find MSE both Result and validation | |
| error = self.loss(result_input, result) | |
| # Find RMSPE both Result and Validate | |
| percent_loss = self.loss_percent(result_input, result) | |
| # Find Derivative of Loss | |
| loss_prime = self.loss_prime(result_input, result) | |
| bias_error = loss_prime | |
| if display_round_log: | |
| print( | |
| f"Epoch {i} Sample {j} : Model level weight {self.weights}") | |
| print(f"Epoch {i} Sample {j} : Bias {self.bias}") | |
| print( | |
| f"Epoch {i} Sample {j} : Result {result}, Validate Result {validate_result}" | |
| ) | |
| print( | |
| f"Epoch {i} Sample {j} : Material Result {predicted_mc} Employee {predicted_ec} Capital Cost {predicted_cc}" | |
| ) | |
| print( | |
| f"Epoch {i} Sample {j} : Actual Result {result_input}") | |
| print( | |
| f"Epoch {i} Sample {j} : Error (MSE) {error}, Error To Adjust(Loss Prime) {loss_prime} " | |
| ) | |
| print( | |
| f"Epoch {i} Sample {j} : Error Percent {percent_loss} ") | |
| print("-----------------") | |
| # Find Gradient of each weight | |
| # mse_prime dot Leaky_relu(input) | |
| # For Adjust Grdient in Model Level | |
| material_weight_error = np.dot(predicted_mc, loss_prime) | |
| ec_weight_error = np.dot(predicted_ec, loss_prime) | |
| cc_weight_error = np.dot(predicted_cc, loss_prime) | |
| # Append For log keeping | |
| sample_errors.append( | |
| { | |
| "epoch": (i - 1), | |
| "sample": j, | |
| "error": error, | |
| "error_percent": percent_loss, | |
| } | |
| ) | |
| # Sample Payload for Debugging | |
| sample_payload = spa.get_sample_payload( | |
| sample_payload=sample_payload, | |
| material_element=self.material_element, | |
| employee_element=self.employee_element, | |
| capital_cost_element=self.capital_cost_element, | |
| material_cost_input=material_cost_input, | |
| material_amount_input=material_amount_input, | |
| employee_input=employee_cost_input, | |
| emp_dur_input=employee_duration_input, | |
| employee_dayamount_input=employee_day_amount_input, | |
| capital_cost_input=capital_cost_input, | |
| day_amount_input=day_amount_input, | |
| capital_cost_dur_input=capital_cost_dur_input, | |
| predicted_mc=predicted_mc, | |
| predicted_ec=predicted_ec, | |
| predicted_cc=predicted_cc, | |
| bias=self.bias, | |
| result_input=result_input, | |
| result=result, | |
| error=error, | |
| percent_loss=percent_loss, | |
| epoch_number=i, | |
| sample_number=j, | |
| model_weights=self.weights, | |
| ) | |
| # print(f'Result {result} & Loss {loss_prime}') | |
| # print( | |
| # f'Material Gradient {material_weight_error}, DE Gradient {de_weight_error}, ME Gradient {me_weight_error}, CC Gradient {cc_weight_error}') | |
| # Back Propagation of Inside Element | |
| # mse_prime dot w | |
| sum_material_error = np.dot(loss_prime, self.weights[0]) | |
| self.material_element.back_propagate( | |
| sum_material_error, self.material_learning_rate | |
| ) | |
| sum_ec_error = np.dot(loss_prime, self.weights[1]) | |
| self.employee_element.back_propagate( | |
| sum_ec_error, self.employee_learning_rate | |
| ) | |
| sum_capital_error = np.dot(loss_prime, self.weights[2]) | |
| self.capital_cost_element.back_propagate( | |
| sum_capital_error, self.capital_cost_learning_rate | |
| ) | |
| if self.use_model_weight is True: | |
| # Update Weight | |
| overall_gradient = np.array( | |
| [ | |
| material_weight_error[0], | |
| ec_weight_error[0], | |
| cc_weight_error[0], | |
| ] | |
| ) | |
| self.weights -= learning_rate * overall_gradient | |
| self.bias -= learning_rate * bias_error | |
| result_input_value = result_input.reshape(1) | |
| result_input_value = result_input_value[0] | |
| if j < len(normalized_validate_material_cost): | |
| validation_error = self.loss( | |
| validate_result_input, validate_result) | |
| validation_error_percent = self.loss_percent( | |
| validate_result_input, validate_result | |
| ) | |
| sum_validation_error += validation_error | |
| sum_validation_error_percent += validation_error_percent | |
| validate_sample_amount += 1 | |
| sum_error += error | |
| sum_error_percent += percent_loss | |
| end_time = time.time() | |
| validation_error = sum_validation_error / validate_sample_amount | |
| print( | |
| f"{i + 1} /{epoch} Epoch Error = {sum_error / samples} ({sum_error_percent / samples} %), Validate Error = {validation_error} ({sum_validation_error_percent / validate_sample_amount}) estimate time {end_time - start_time}" | |
| ) | |
| epoch_errors.append( | |
| { | |
| "epoch": (i + 1), | |
| "error": sum_error / samples, | |
| "error_percent": sum_error_percent / samples, | |
| "validate_error": sum_validation_error / validate_sample_amount, | |
| "validate_error_percent": sum_validation_error_percent / validate_sample_amount, | |
| } | |
| ) | |
| if self.use_early_stopping: | |
| if validation_error < best_validation_error: | |
| best_validation_error = validation_error | |
| patience_counter = 0 | |
| else: | |
| patience_counter += 1 | |
| if patience_counter == self.patience: | |
| print(f"Early Stopping at Epoch {i + 1}") | |
| break | |
| sum_error = 0 | |
| sum_error_percent = 0 | |
| sum_validation_error = 0 | |
| sum_validation_error_percent = 0 | |
| self.epoch_error = epoch_errors | |
| self.sample_errors = sample_errors | |
| self.sample_payload = sample_payload | |
| epoch_error_df = pd.DataFrame(epoch_errors) | |
| last_sample_payload = sample_payload[-1] | |
| print("-------------------") | |
| print( | |
| f"Minimum Error = {epoch_error_df['error'].min()} Minimum Error percent {round(epoch_error_df['error_percent'].min(), 4)} Accuracy {100 - epoch_error_df['error_percent'].min()}%" | |
| ) | |
| print( | |
| f"Average Error = {epoch_error_df['error'].mean()} Average Error percent {round(epoch_error_df['error_percent'].mean(), 4)} Accuracy {100 - epoch_error_df['error_percent'].mean()}%" | |
| ) | |
| print( | |
| f"Maximum Error = {epoch_error_df['error'].max()} Maximum Error percent {round(epoch_error_df['error_percent'].max(), 2)} Accuracy {100 - epoch_error_df['error_percent'].max()}%" | |
| ) | |
| print( | |
| f"Minimum Validation Error = {epoch_error_df['validate_error'].min()} Minimum Error percent {round(epoch_error_df['validate_error_percent'].min(), 4)} Accuracy {100 - epoch_error_df['validate_error_percent'].min()}%" | |
| ) | |
| print( | |
| f"Average Validation Error = {epoch_error_df['validate_error'].mean()} Average Error percent {round(epoch_error_df['validate_error_percent'].mean(), 4)} Accuracy {100 - epoch_error_df['validate_error_percent'].mean()}%" | |
| ) | |
| print( | |
| f"Maximum Validation Error = {epoch_error_df['validate_error'].max()} Maximum Error percent {round(epoch_error_df['validate_error_percent'].max(), 4)} Accuracy {100 - epoch_error_df['validate_error_percent'].max()}%" | |
| ) | |
| print( | |
| f"Last Validation Error = {epoch_error_df.iloc[-1]['validate_error']} Last Validate Error percent {round(epoch_error_df.iloc[-1]['validate_error_percent'], 4)} Accuracy {100 - epoch_error_df.iloc[-1]['validate_error_percent']}%" | |
| ) | |
| print( | |
| f"Final Model - Material Weight {last_sample_payload['material_weight']} Material Bias {last_sample_payload['material_bias']}" | |
| ) | |
| print( | |
| f" - Employee Weight {last_sample_payload['employee_weight']} Monthly Employee Bias {last_sample_payload['employee_bias']}" | |
| ) | |
| print( | |
| f" - Capital Cost Weight {last_sample_payload['capital_cost_weight']} Capital Cost Bias {last_sample_payload['capital_cost_bias']}" | |
| ) | |
| print(f" - Model Bias {last_sample_payload['model_bias']}") | |
| print("-------------------") | |
| return results | |
| def get_gradient(self): | |
| return self.gradient | |
| def get_prediction_error(self): | |
| return self.prediction_error | |
| def get_weight_list(self): | |
| return self.weight_list | |
| def get_epoch_error(self): | |
| return self.epoch_error | |
| def get_sample_error(self): | |
| return self.sample_errors | |
| def get_model_element_weights(self): | |
| return ( | |
| self.material_element.get_weight_list(), | |
| self.daily_employee_element.get_weight_list(), | |
| self.monthly_employee_element.get_weight_list(), | |
| self.capital_cost_element.get_weight_list(), | |
| ) | |
| def get_sample_payload(self): | |
| return self.sample_payload | |