Grafana
Mungkin setiap IT atau para praktisi penyuka monitoring sudah mengetahui grafana ini, grafana aplikasi atau software untuk monitoring, disini saya ada 2 percobaan yaitu mikrotik router dan API Sensor yang dimana untuk api sensor ini saya mencoba menggunakan ESP32 mikrokontroller untuk mengirim api ke grafana.
Pada ESP32 terdapat sensor sepert stop kontak / iginition, rfid one wire(ignition), sensor suhu(thermistor) wire, lalu ada eye sensor suhu & humidity(Wireless), button untuk harsh breaking(rem dadakan), lalu potesio untuk speednya, lcd tft(opsional), buzzer speed limiter.
Pengujian
Pada pengujian ini bentuk code pada esp32 json nya seperti berikut
import urllib3
from flask import Flask
from flask_socketio import SocketIO
import requests
import time
import threading
import mysql.connector
from geopy.distance import geodesic
import numpy as np
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score, classification_report, confusion_matrix
import pandas as pd
app = Flask(__name__)
app.config['SECRET_KEY'] = 'secret!'
socketio = SocketIO(app, cors_allowed_origins="*")
urllib3.disable_warnings()
lat = []
lng = []
# Konfigurasi
FUEL_TANK_CAPACITY = 300 # Kapasitas tangki bensin dalam liter
url = 'urljsonapi'
db_config = {
'host': 'localhost',
'user': 'root',
'password': '',
'database': 'vehicle_data'
}
# Fungsi Predictive Maintenance
def train_predictive_maintenance_model():
# Buat dataset simulasi
np.random.seed(42)
n_samples = 1000
temperature = np.random.normal(90, 10, n_samples) # Suhu mesin
vibration = np.random.normal(5, 2, n_samples) # Getaran mesin
oil_pressure = np.random.normal(50, 5, n_samples) # Tekanan oli
# Status mesin: 0 = normal, 1 = gagal
failure = np.where(
(temperature > 100) | (vibration > 8) | (oil_pressure < 40),
1, # Gagal
0 # Normal
)
# Gabungkan data ke dalam DataFrame
data = pd.DataFrame({
'temperature': temperature,
'vibration': vibration,
'oil_pressure': oil_pressure,
'failure': failure
})
# Pisahkan fitur dan target
X = data[['temperature', 'vibration', 'oil_pressure']]
y = data['failure']
# Bagi data menjadi data latih dan data uji
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)
# Latih model Random Forest
model = RandomForestClassifier(n_estimators=100, random_state=42)
model.fit(X_train, y_train)
return model
# Load model predictive maintenance
pm_model = train_predictive_maintenance_model()
def predict_failure(vehicle_data):
# Ambil data sensor
temperature = vehicle_data.get('engine_oil_temp', 90) # Default 90°C jika data tidak ada
vibration = vehicle_data.get('vibration', 5) # Default 5 jika data tidak ada
oil_pressure = vehicle_data.get('engine_oil_pressure', 50) # Default 50 jika data tidak ada
# Buat input untuk model
input_data = np.array([[temperature, vibration, oil_pressure]])
# Prediksi status mesin
prediction = pm_model.predict(input_data)
print(f"Prediksi: {prediction} %")
return prediction[0] # 0 = normal, 1 = gagal
# Fungsi lainnya (get_current_location_name, predict_fuel_efficiency, dll.) tetap sama
# ...
def get_current_location_name(lat, lng):
try:
# Validasi koordinat
if not isinstance(lat, (int, float)) or not isinstance(lng, (int, float)):
return 'Koordinat tidak valid'
if lat < -90 or lat > 90 or lng < -180 or lng > 180:
return 'Koordinat di luar jangkauan'
# Tambahkan header User-Agent sesuai persyaratan Nominatim
headers = {
'User-Agent': 'YourApp/1.0 (contact@yourdomain.com)'
}
# Tambahkan delay untuk menghindari rate limiting
time.sleep(1)
response = requests.get(
f"https://nominatim.openstreetmap.org/reverse?lat={lat}&lon={lng}&format=json",
headers=headers
)
# Cek status code
if response.status_code != 200:
print(f"Error Nominatim: Status code {response.status_code}")
return 'Lokasi tidak diketahui'
# Cek content type
if 'application/json' not in response.headers.get('Content-Type', ''):
print(f"Invalid response content: {response.text[:100]}")
return 'Lokasi tidak diketahui'
data = response.json()
# Ekstrak nama lokasi
address = data.get('address', {})
components = []
if address.get('village'):
components.append(address['village'])
if address.get('city'):
components.append(address['city'])
if address.get('state'):
components.append(address['state'])
return ', '.join(components) if components else 'Lokasi tidak diketahui'
except json.JSONDecodeError:
print(f"Gagal parse JSON. Response: {response.text[:200]}")
return 'Lokasi tidak diketahui'
except Exception as e:
print(f"Error reverse geocoding: {str(e)}")
return 'Lokasi tidak diketahui'
def predict_fuel_efficiency(vehicle_data):
# Base efficiency untuk truk berat
base_eff = 2.8 # km/L
# Ambil nilai sensor dengan default 0 dan handle None
rpm = vehicle_data.get('engine_rpm') or 0
speed = vehicle_data.get('speed') or 0
engine_temp = vehicle_data.get('engine_oil_temp') or 90 # Default 90°C jika data tidak ada
# 1. Validasi tipe data dan handle None
try:
rpm = float(rpm)
speed = float(speed)
engine_temp = float(engine_temp)
except (TypeError, ValueError):
return 0.0
# 2. Logika prediksi
# Engine off condition
if rpm <= 0 or speed < 1:
return 0.0
# Pengaruh RPM
if rpm > 2100:
base_eff *= 0.80
elif rpm > 1800:
base_eff *= 0.85
elif rpm < 1100:
base_eff *= 0.90
# Pengaruh kecepatan
if speed > 90:
base_eff *= 0.75
elif speed > 75:
base_eff *= 0.85
elif speed < 40:
base_eff *= 0.90
# Pengaruh temperatur mesin
if engine_temp > 105:
base_eff *= 0.85
elif engine_temp < 75:
base_eff *= 0.90
return max(round(base_eff, 2), 0.0)
def generate_fuel_message(range_km):
if range_km > 300:
return f"🟢 Bahan bakar cukup untuk sejauh {range_km:.1f} km"
elif range_km > 150:
return f"🟡 Bahan bakar cukup untuk sejauh {range_km:.1f} km"
else:
return f"🔴 Hanya cukup sejauh {range_km:.1f} km - Isi ulang Bensin segera"
def generate_service_message(odometer_km):
try:
odometer_km = float(odometer_km)
if odometer_km >= 15000:
return "🔴 Servis darurat diperlukan! Sudah lebih dari 15,000 km"
elif odometer_km >= 5000:
return "🟡 Waktunya servis rutin (melebihi 5,000 km)"
else:
next_service = 5000 - odometer_km
return f"🟢 Servis berikutnya dalam {next_service:.0f} km"
except:
return "⚪ Data servis tidak tersedia"
def fetch_and_process_data():
try:
response = requests.get(url, verify=False)
response.raise_for_status()
data = response.json()
except Exception as e:
print(f"Error fetching data: {e}")
return []
filtered_devices = []
for group in data:
for device in group['items']:
device_info = {
'device_id': device.get('id'),
'license_plate': device.get('name', '').split('\t')[0],
'speed': device.get('speed', 0),
'odometer': device.get('odometer', 0),
'lat': device.get('lat'),
'lng': device.get('lng'),
'total_distance': device.get('total_distance', 0),
'fuel_tank_percent': None,
'engine_rpm': None,
'engine_oil_temp': None,
'engine_oil_pressure': None,
'fuel_consumption': None,
'current_gear': None,
'battery_voltage': None,
'coolant_temp': None
}
for sensor in device.get('sensors', []):
sensor_type = sensor.get('type')
sensor_name = sensor.get('name', '')
sensor_value = sensor.get('val')
try:
if sensor_type == 'fuel_tank' and 'Fuel Level' in sensor_name:
device_info['fuel_tank_percent'] = float(sensor_value)
elif sensor_type == 'odometer' and 'High Resolution Total Vehicle Distance' in sensor_name:
device_info['odometer'] = float(sensor_value)
elif sensor_type == 'tachometer' and 'Engine RPM' in sensor_name:
device_info['engine_rpm'] = float(sensor_value)
elif sensor_type == 'temperature' and 'Engine Oil Temperature' in sensor_name:
device_info['engine_oil_temp'] = float(sensor_value)
elif sensor_type == 'temperature' and 'f. Engine Oil Pressure' in sensor_name:
device_info['engine_oil_pressure'] = float(sensor_value)
elif sensor_type == 'temperature' and 'Actual Fuel Consumption' in sensor_name:
device_info['fuel_consumption'] = float(sensor_value)
elif sensor_type == 'temperature' and 'Engine Coolant Temperature' in sensor_name:
device_info['coolant_temp'] = float(sensor_value)
elif sensor_type == 'battery' and 'Battery Voltage' in sensor_name:
device_info['battery_voltage'] = float(sensor_value)
elif sensor_type == 'fuel_consumption' and 'Fuel Consumption' in sensor_name:
device_info['fuel_consumption'] = float(sensor_value)
except (ValueError, TypeError) as e:
print(f"Error processing sensor data: {e}")
# Prediksi kegagalan mesin
device_info['failure_prediction'] = predict_failure(device_info)
if device_info['fuel_tank_percent']:
try:
fuel_liters = (device_info['fuel_tank_percent'] / 100) * FUEL_TANK_CAPACITY
efficiency = predict_fuel_efficiency(device_info)
predicted_range = fuel_liters * efficiency
location_name = get_current_location_name(
device_info['lat'],
device_info['lng']
)
device_info.update({
'predicted_range': round(predicted_range, 2),
'fuel_efficiency': efficiency,
'location_name': location_name,
'fuel_message': generate_fuel_message(predicted_range),
'service_message': generate_service_message(device_info.get('odometer', 0))
})
except Exception as e:
print(f"Error calculating fuel prediction: {e}")
filtered_devices.append(device_info)
return filtered_devices
# Fungsi lainnya (create_table, data_update_loop, dll.) tetap sama
# ...
def create_table():
try:
conn = mysql.connector.connect(**db_config)
cursor = conn.cursor()
cursor.execute('''
CREATE TABLE IF NOT EXISTS vehicles (
id INT AUTO_INCREMENT PRIMARY KEY,
device_id VARCHAR(255),
license_plate VARCHAR(20),
timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
lat DECIMAL(10,6),
lng DECIMAL(10,6),
speed DECIMAL(6,2),
engine_rpm INT,
fuel_tank_percent DECIMAL(5,2),
engine_oil_temp DECIMAL(5,2),
battery_voltage DECIMAL(5,2),
predicted_range DECIMAL(10,2),
fuel_efficiency DECIMAL(5,2),
location_name TEXT,
fuel_message TEXT,
service_message TEXT
)
''')
conn.commit()
except mysql.connector.Error as err:
print(f"Error creating table: {err}")
def data_update_loop():
while True:
try:
vehicles = fetch_and_process_data()
for vehicle in vehicles:
try:
conn = mysql.connector.connect(**db_config)
cursor = conn.cursor()
cursor.execute('''
INSERT INTO vehicles (
device_id, license_plate, lat, lng, speed, odometer, fuel_consumption,
engine_rpm, coolant_temp, fuel_tank_percent, engine_oil_temp, engine_oil_pressure,
battery_voltage, predicted_range, fuel_efficiency,
location_name, fuel_message, service_message
) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
''', (
vehicle['device_id'],
vehicle['license_plate'],
vehicle['lat'],
vehicle['lng'],
vehicle['speed'],
vehicle.get('odometer'),
vehicle.get('fuel_consumption'),
vehicle.get('engine_rpm'),
vehicle.get('coolant_temp'),
vehicle.get('fuel_tank_percent'),
vehicle.get('engine_oil_temp'),
vehicle.get('engine_oil_pressure'),
vehicle.get('battery_voltage'),
vehicle.get('predicted_range'),
vehicle.get('fuel_efficiency'),
vehicle.get('location_name'),
vehicle.get('fuel_message'),
vehicle.get('service_message')
))
conn.commit()
except mysql.connector.Error as err:
print(f"Database error: {err}")
finally:
if conn.is_connected():
cursor.close()
conn.close()
time.sleep(30)
except Exception as e:
print(f"Error in update loop: {e}")
time.sleep(30)
if __name__ == '__main__':
create_table()
threading.Thread(target=data_update_loop, daemon=True).start()
socketio.run(app, debug=True, port=5000)
setelah code itu dijalankan maka pada cmd akan mendapatkan data / getdata api dari platform ataupun dari esp32 di parsing menjadi API. lalu setiap data API tersebut akan di filter dan store ke mysql.
dengan hasil grafana sebagai berikut
Selanjutnya data tersebut dilakukan riset terkait AI nya untuk memberikan notifikasinya.