CIzz22
/
rbd-app
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix calculation

Browse Source
main
Cizz22 11 months ago
parent 20d6267108
commit 89834cd4c3
1 changed files with 82 additions and 80 deletions
Show Stats Download Patch File Download Diff File

162
src/calculation_time_constrains/service.py
Unescape Escape View File

@ -54,93 +54,95 @@ def get_overhaul_cost_by_time_chart(
# return results # return results
# async def get_corrective_cost_time_chart( async def get_corrective_cost_time_chart(
# material_cost: float, service_cost: float, location_tag: str, token material_cost: float, service_cost: float, location_tag: str, token
# ) -> Tuple[np.ndarray, np.ndarray]: ) -> Tuple[np.ndarray, np.ndarray]:
# """ """
# Fetch failure data from API and calculate corrective costs, ensuring 365 days of data. Fetch failure data from API and calculate corrective costs, ensuring 365 days of data.
# Args: Args:
# material_cost: Cost of materials per failure material_cost: Cost of materials per failure
# service_cost: Cost of service per failure service_cost: Cost of service per failure
# location_tag: Location tag of the equipment location_tag: Location tag of the equipment
# token: Authorization token token: Authorization token
# Returns: Returns:
# Tuple of (corrective_costs, daily_failure_rate) Tuple of (corrective_costs, daily_failure_rate)
# """ """
# url = f"http://192.168.1.82:8000/reliability/main/number-of-failures/{location_tag}/2024-01-01/2024-12-31" url = f"http://192.168.1.82:8000/reliability/main/number-of-failures/{location_tag}/2024-01-01/2024-12-31"
# try: try:
# response = requests.get( response = requests.get(
# url, url,
# headers={ headers={
# "Content-Type": "application/json", "Content-Type": "application/json",
# "Authorization": f"Bearer {token}", "Authorization": f"Bearer {token}",
# }, },
# ) )
# data = response.json() data = response.json()
# Create a complete date range for 2024
start_date = datetime.datetime(2024, 1, 1)
date_range = [start_date + datetime.timedelta(days=x) for x in range(365)]
# # Create a complete date range for 2024 # Create a dictionary of existing data
# start_date = datetime.datetime(2024, 1, 1) data_dict = {
# date_range = [start_date + datetime.timedelta(days=x) for x in range(365)] datetime.datetime.strptime(item["date"], "%d %b %Y"): item["num_fail"]
for item in data["data"]
}
# # Create a dictionary of existing data # Fill in missing dates with nearest available value
# data_dict = { complete_data = []
# datetime.datetime.strptime(item["date"], "%d %b %Y"): item["num_fail"] last_known_value = 0 # Default value if no data is available
# for item in data["data"]
# }
# # Fill in missing dates with nearest available value for date in date_range:
# complete_data = [] if date in data_dict:
# last_known_value = 0 # Default value if no data is available if data_dict[date] is not None:
last_known_value = data_dict[date]
complete_data.append(last_known_value)
else:
complete_data.append(0)
# for date in date_range: # Convert to numpy array
# if date in data_dict: daily_failure = np.array(complete_data)
# if data_dict[date] is not None:
# last_known_value = data_dict[date]
# complete_data.append(last_known_value)
# else:
# complete_data.append(last_known_value)
# # Convert to numpy array failure_counts = np.cumsum(daily_failure)
# daily_failure = np.array(complete_data)
# # Calculate corrective costs # Calculate corrective costs
# cost_per_failure = material_cost + service_cost cost_per_failure = material_cost + service_cost
# corrective_costs = daily_failure * cost_per_failure corrective_costs = failure_counts * cost_per_failure
# return corrective_costs, daily_failure return corrective_costs, daily_failure
# except Exception as e: except Exception as e:
# print(f"Error fetching or processing data: {str(e)}") print(f"Error fetching or processing data: {str(e)}")
# raise raise
def get_corrective_cost_time_chart(material_cost: float, service_cost: float, days: int, numEquipments: int) -> Tuple[np.ndarray, np.ndarray]: # def get_corrective_cost_time_chart(material_cost: float, service_cost: float, days: int, numEquipments: int) -> Tuple[np.ndarray, np.ndarray]:
day_points = np.arange(0, days) # day_points = np.arange(0, days)
# Parameters for failure rate # # Parameters for failure rate
base_rate = 0.04 # Base failure rate per day # base_rate = 0.04 # Base failure rate per day
acceleration = 0.7 # How quickly failure rate increases # acceleration = 0.7 # How quickly failure rate increases
grace_period = 49 # Days before failures start increasing significantly # grace_period = 49 # Days before failures start increasing significantly
# Calculate daily failure rate using sigmoid function # # Calculate daily failure rate using sigmoid function
daily_failure_rate = base_rate / (1 + np.exp(-acceleration * (day_points - grace_period)/days)) # daily_failure_rate = base_rate / (1 + np.exp(-acceleration * (day_points - grace_period)/days))
# Introduce randomness in the failure rate # # Introduce randomness in the failure rate
random_noise = np.random.normal(0.0, 0.05, (numEquipments, days)) # Mean 0.0, Std Dev 0.05 # random_noise = np.random.normal(0.0, 0.05, (numEquipments, days)) # Mean 0.0, Std Dev 0.05
daily_failure_rate = daily_failure_rate + random_noise # daily_failure_rate = daily_failure_rate + random_noise
daily_failure_rate = np.clip(daily_failure_rate, 0, None) # Ensure failure rate is non-negative # daily_failure_rate = np.clip(daily_failure_rate, 0, None) # Ensure failure rate is non-negative
# Calculate cumulative failures # # Calculate cumulative failures
failure_counts = np.cumsum(daily_failure_rate) # failure_counts = np.cumsum(daily_failure_rate)
# Calculate corrective costs based on cumulative failures and combined costs # # Calculate corrective costs based on cumulative failures and combined costs
cost_per_failure = material_cost + service_cost # cost_per_failure = material_cost + service_cost
corrective_costs = failure_counts * cost_per_failure # corrective_costs = failure_counts * cost_per_failure
return corrective_costs, daily_failure_rate # return corrective_costs, daily_failure_rate
async def create_param_and_data( async def create_param_and_data(
@ -317,18 +319,18 @@ async def create_calculation_result_service(
# Calculate for each equipment # Calculate for each equipment
for eq in equipments: for eq in equipments:
# corrective_costs, daily_failures = await get_corrective_cost_time_chart( corrective_costs, daily_failures = await get_corrective_cost_time_chart(
# material_cost=eq.material_cost,
# service_cost=eq.service_cost,
# token=token,
# location_tag=eq.equipment.location_tag,
# )
corrective_costs, daily_failures = get_corrective_cost_time_chart(
material_cost=eq.material_cost, material_cost=eq.material_cost,
service_cost=eq.service_cost, service_cost=eq.service_cost,
days=days, token=token,
numEquipments=len(equipments), location_tag=eq.equipment.location_tag,
) )
# corrective_costs, daily_failures = get_corrective_cost_time_chart(
# material_cost=eq.material_cost,
# service_cost=eq.service_cost,
# days=days,
# numEquipments=len(equipments),
# )
overhaul_cost_points = get_overhaul_cost_by_time_chart( overhaul_cost_points = get_overhaul_cost_by_time_chart(
calculation_data.parameter.overhaul_cost, calculation_data.parameter.overhaul_cost,

Write Preview
Loading…
Cancel
Save