feat: Implement a simulation mode for equipment prediction and EAC, and extract execution time formatting into a helper function.

main
MrWaradana 4 weeks ago
parent 498244052b
commit 090dd1ff64

@ -6,6 +6,47 @@ from src.modules.equipment.insert_actual_data import query_data
from src.modules.equipment.Prediksi import Prediksi, main as predict_run from src.modules.equipment.Prediksi import Prediksi, main as predict_run
from src.modules.equipment.Eac import Eac, main as eac_run from src.modules.equipment.Eac import Eac, main as eac_run
def format_execution_time(execution_time):
if execution_time >= 3600:
hours = int(execution_time // 3600)
minutes = int((execution_time % 3600) // 60)
seconds = execution_time % 60
return f"{hours}h {minutes}m {seconds:.2f}s."
elif execution_time >= 60:
minutes = int(execution_time // 60)
seconds = execution_time % 60
return f"{minutes}m {seconds:.2f}s."
else:
return f"{execution_time:.2f} seconds."
# Alternative calling function to just predict and calculate eac without inserting actual data
async def simulate():
start_time = time.time()
print("Starting simulation (predict + eac)...")
try:
prediction_result = await predict_run()
if prediction_result is False:
print("Prediction step failed or was skipped. Skipping EAC run.")
return
except Exception as e:
print(f"Error in predict_run: {str(e)}")
return
try:
result = eac_run()
if asyncio.iscoroutine(result):
result = await result
print("EAC run completed.")
except Exception as e:
print(f"Error in eac_run: {str(e)}")
return
end_time = time.time()
message = f"Simulation finished in {format_execution_time(end_time - start_time)}"
print(message)
return message
# Panggil fungsi # Panggil fungsi
async def main(): async def main():
start_time = time.time() start_time = time.time()
@ -29,34 +70,20 @@ async def main():
result = eac_run() result = eac_run()
if asyncio.iscoroutine(result): if asyncio.iscoroutine(result):
result = await result result = await result
if isinstance(result, (list, tuple)):
print(f"EAC run returned {len(result)} items.")
else:
print("EAC run completed.") print("EAC run completed.")
except Exception as e: except Exception as e:
print(f"Error in eac_run: {str(e)}") print(f"Error in eac_run: {str(e)}")
return return
end_time = time.time() end_time = time.time()
execution_time = end_time - start_time message = f"Script calculation finished in {format_execution_time(end_time - start_time)}"
# format execution time into h/m/s as needed
if execution_time >= 3600:
hours = int(execution_time // 3600)
minutes = int((execution_time % 3600) // 60)
seconds = execution_time % 60
message = f"Script calculation finished in {hours}h {minutes}m {seconds:.2f}s."
elif execution_time >= 60:
minutes = int(execution_time // 60)
seconds = execution_time % 60
message = f"Script calculation finished in {minutes}m {seconds:.2f}s."
else:
message = f"Script calculation finished in {execution_time:.2f} seconds."
print(message) print(message)
return message return message
if __name__ == "__main__": if __name__ == "__main__":
asyncio.run( import sys
main() # Use 'simulate' argument to run without query_data
) if len(sys.argv) > 1 and sys.argv[1] == "simulate":
asyncio.run(simulate())
else:
asyncio.run(main())

Loading…
Cancel
Save