50 lines
1.5 KiB
Python
50 lines
1.5 KiB
Python
import mysql.connector
|
|
import json
|
|
from datetime import datetime
|
|
|
|
DB_HOST = "localhost"
|
|
DB_PORT = 3306
|
|
DB_USER = "ctnt_root"
|
|
DB_PASSWORD = "Umsang6595!!"
|
|
DB_NAME = "carbon_cutting_mc_db"
|
|
|
|
def main():
|
|
try:
|
|
conn = mysql.connector.connect(
|
|
host=DB_HOST,
|
|
port=DB_PORT,
|
|
user=DB_USER,
|
|
password=DB_PASSWORD,
|
|
database=DB_NAME
|
|
)
|
|
cur = conn.cursor(dictionary=True)
|
|
cur.execute("SELECT m_read_raw, device_state, timestamp FROM machine_status WHERE equipment_id = 8")
|
|
row = cur.fetchone()
|
|
|
|
print("\n================ [DB REAL-TIME STATUS] ================")
|
|
if row:
|
|
print(f"Device State : {row['device_state']}")
|
|
print(f"Last Update : {row['timestamp']}")
|
|
|
|
m_read_raw_str = row['m_read_raw']
|
|
print(f"Raw String : {m_read_raw_str}")
|
|
|
|
if m_read_raw_str:
|
|
m_read = json.loads(m_read_raw_str)
|
|
active_bits = [int(k) for k, v in m_read.items() if int(v) == 1]
|
|
active_bits.sort()
|
|
print(f"Active Bits (val=1): {active_bits}")
|
|
else:
|
|
print("Active Bits : No Data")
|
|
else:
|
|
print("No status row found for equipment_id = 8")
|
|
print("========================================================\n")
|
|
|
|
cur.close()
|
|
conn.close()
|
|
except Exception as e:
|
|
print(f"DB Connection or Query failed: {e}")
|
|
|
|
if __name__ == "__main__":
|
|
main()
|