CIzz22
/
be_monitoring
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix reliabilty target oh

Browse Source
main
Cizz22 11 months ago
parent 553c80d394
commit a62f873774
4 changed files with 158 additions and 63 deletions
Show Stats Download Patch File Download Diff File

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

@ -139,60 +139,149 @@ async def get_corrective_cost_time_chart(
) -> Tuple[np.ndarray, np.ndarray]:
days_difference = (end_date - start_date).days
url = f"http://192.168.1.82:8000/reliability/main/number-of-failures/{location_tag}/{start_date.strftime('%Y-%m-%d')}/{end_date.strftime('%Y-%m-%d')}"
today = datetime.datetime.now().replace(hour=0, minute=0, second=0, microsecond=0)
url_prediction = f"http://192.168.1.82:8000/reliability/main/number-of-failures/{location_tag}/{start_date.strftime('%Y-%m-%d')}/{end_date.strftime('%Y-%m-%d')}"
url_history = f"http://192.168.1.82:8000/reliability/main/failures/{location_tag}/{start_date.strftime('%Y-%m-%d')}/{end_date.strftime('%Y-%m-%d')}"
# Initialize monthly data dictionary
monthly_data = {}
# Get historical data (start_date to today)
if start_date <= today:
try:
response = requests.get(
url,
url_history,
headers={
"Content-Type": "application/json",
"Authorization": f"Bearer {token}",
},
)
data = response.json()
latest_num = data["data"][-1]["num_fail"]
history_data = response.json()
# Process historical data - accumulate failures by month
history_dict = {}
for item in history_data["data"]:
date = datetime.datetime.strptime(item["date"], "%d %b %Y")
month_key = datetime.datetime(date.year, date.month, 1)
# Initialize if first occurrence of this month
if month_key not in history_dict:
history_dict[month_key] = 0
# Accumulate failures for this month
if item["num_fail"] is not None:
history_dict[month_key] += item["num_fail"]
# Update monthly_data with historical data
monthly_data.update(history_dict)
except Exception as e:
print(f"Error fetching historical data: {e}")
latest_num = 1
# Get prediction data (today+1 to end_date)
if end_date > today:
try:
response = requests.get(
url_prediction,
headers={
"Content-Type": "application/json",
"Authorization": f"Bearer {token}",
},
)
prediction_data = response.json()
# Use the last prediction value for future months
latest_num = prediction_data["data"][-1]["num_fail"] if prediction_data["data"] else 1
if not latest_num:
latest_num = 1
# Create a complete date range for 2025
date_range = [start_date + datetime.timedelta(days=x) for x in range(days_difference)]
# Create prediction dictionary
prediction_dict = {}
for item in prediction_data["data"]:
date = datetime.datetime.strptime(item["date"], "%d %b %Y")
month_key = datetime.datetime(date.year, date.month, 1)
prediction_dict[month_key] = item["num_fail"]
# Create a dictionary of existing data
data_dict = {
datetime.datetime.strptime(item["date"], "%d %b %Y"): item["num_fail"]
for item in data["data"]
}
# Update monthly_data with prediction data
for key in prediction_dict:
if key not in monthly_data: # Don't overwrite historical data
monthly_data[key] = prediction_dict[key]
except Exception as e:
print(f"Error fetching prediction data: {e}")
monthly_data = {}
current_date = start_date.replace(day=1)
# Create a complete date range covering all months from start to end
current_date = datetime.datetime(start_date.year, start_date.month, 1)
while current_date <= end_date:
monthly_data[current_date] = float('inf') # Start with infinity to find minimum
if current_date not in monthly_data:
monthly_data[current_date] = 0
# Move to next month
if current_date.month == 12:
current_date = datetime.datetime(current_date.year + 1, 1, 1)
else:
current_date = datetime.datetime(current_date.year, current_date.month + 1, 1)
# Get the minimum value for each month
for date in data_dict.keys():
month_key = datetime.datetime(date.year, date.month, 1)
if month_key in monthly_data and data_dict[date] is not None:
# Update only if the value is lower (to get the minimum value)
monthly_data[month_key] = min(monthly_data[month_key], data_dict[date])
# Convert to list maintaining chronological order
complete_data = []
for month in sorted(monthly_data.keys()):
# Replace any remaining infinity values with 0 or another appropriate default
if monthly_data[month] == float('inf'):
monthly_data[month] = 0
complete_data.append(monthly_data[month])
# Convert to numpy array
monthly_failure = np.array(complete_data)
# try:
# response = requests.get(
# url,
# headers={
# "Content-Type": "application/json",
# "Authorization": f"Bearer {token}",
# },
# )
# data = response.json()
# latest_num = data["data"][-1]["num_fail"]
# if not latest_num:
# latest_num = 1
# # Create a complete date range for 2025
# date_range = [start_date + datetime.timedelta(days=x) for x in range(days_difference)]
# # Create a dictionary of existing data
# data_dict = {
# datetime.datetime.strptime(item["date"], "%d %b %Y"): item["num_fail"]
# for item in data["data"]
# }
# monthly_data = {}
# current_date = start_date.replace(day=1)
# while current_date <= end_date:
# monthly_data[current_date] = float('inf') # Start with infinity to find minimum
# # Move to next month
# if current_date.month == 12:
# current_date = datetime.datetime(current_date.year + 1, 1, 1)
# else:
# current_date = datetime.datetime(current_date.year, current_date.month + 1, 1)
# # Get the minimum value for each month
# for date in data_dict.keys():
# month_key = datetime.datetime(date.year, date.month, 1)
# if month_key in monthly_data and data_dict[date] is not None:
# # Update only if the value is lower (to get the minimum value)
# monthly_data[month_key] = min(monthly_data[month_key], data_dict[date])
# # Convert to list maintaining chronological order
# complete_data = []
# for month in sorted(monthly_data.keys()):
# # Replace any remaining infinity values with 0 or another appropriate default
# if monthly_data[month] == float('inf'):
# monthly_data[month] = 0
# complete_data.append(monthly_data[month])
# # Convert to numpy array
# monthly_failure = np.array(complete_data)
# Calculate corrective costs
cost_per_failure = (material_cost + service_cost) / latest_num
if cost_per_failure == 0:
@ -201,12 +290,11 @@ async def get_corrective_cost_time_chart(
corrective_costs = monthly_failure * cost_per_failure
return corrective_costs, monthly_failure
except Exception as e:
print(f"Error fetching or processing data: {str(e)}")
raise
# except Exception as e:
# print(f"Error fetching or processing data: {str(e)}")
# raise
def get_overhaul_cost_by_time_chart(
overhaul_cost: float, months_num: int, numEquipments: int, decay_base: float = 1.01

2
src/overhaul_activity/model.py
Unescape Escape View File

@ -27,5 +27,5 @@ class OverhaulActivity(Base, DefaultMixin):
overhaul_scope = relationship(
"OverhaulScope",
lazy="raise",
lazy="joined",
)

6
src/overhaul_activity/schema.py
Unescape Escape View File

@ -21,6 +21,11 @@ class OverhaulActivityUpdate(OverhaulActivityBase):
material_cost: Optional[float] = Field(0)
service_cost: Optional[float] = Field(0)
class OverhaulScope(DefultBase):
type: str
start_date: datetime
end_date: datetime
duration_oh: int
class OverhaulActivityRead(OverhaulActivityBase):
id: UUID
@ -29,6 +34,7 @@ class OverhaulActivityRead(OverhaulActivityBase):
assetnum: str = Field(..., description="Assetnum is required")
status: str
equipment: MasterEquipmentRead
overhaul_scope: OverhaulScope
class OverhaulActivityPagination(Pagination):

1
src/overhaul_activity/service.py
Unescape Escape View File

@ -69,6 +69,7 @@ async def get_all_by_session_id(*, db_session: DbSession, overhaul_session_id):
Select(OverhaulActivity)
.where(OverhaulActivity.overhaul_scope_id == overhaul_session_id)
.options(joinedload(OverhaulActivity.equipment))
.options(joinedload(OverhaulActivity.overhaul_scope))
)
results = await db_session.execute(query)

Write Preview
Loading…
Cancel
Save