aniketSaha/NorthStar-HMI
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: aniketSaha:6c26d9d6a20116ffc81c093833985809b4ce0cb0
Branches Tags
aniketSaha:main
aniketSaha:asyncio-integration
..
compare: aniketSaha:9e5e4c2a7014bd584e9ab1d9d17b354550d34a9f
Branches Tags
aniketSaha:main
aniketSaha:asyncio-integration

No commits in common. "6c26d9d6a20116ffc81c093833985809b4ce0cb0" and "9e5e4c2a7014bd584e9ab1d9d17b354550d34a9f" have entirely different histories.
6c26d9d6a2 ... 9e5e4c2a70

1 changed files with 23 additions and 77 deletions
Show Stats Expand all files Collapse all files

100
classCAN.py
View File

@ -3,8 +3,6 @@ import canopen
import can import can
import time import time
import os import os
import logging
class CANBackend: class CANBackend:
""" """
@ -120,26 +118,20 @@ class CANBackend:
"Pump_sp": int.from_bytes(data[6:8], 'little') / 100.0, "Pump_sp": int.from_bytes(data[6:8], 'little') / 100.0,
}) })
elif cob_id == 0x2AA and len(data) >= 7: elif cob_id == 0x2AA and len(data) >= 8:
data = list(data)
self.latest_data[1].update({ self.latest_data[1].update({
"MV02": 100 * data[0] / 255, "MV02": data[0],
"MV03": 100 * data[1] / 255, "MV03": data[1],
"MV04": 100 * data[2] / 255, "MV04": data[2],
"MV05": 100 * data[3] / 255, "MV05": data[3],
"MV06": 100 * data[4] / 255, "MV06": data[4],
"MV07": 100 * data[5] / 255, "MV07": data[5],
"MV08": 100 * data[6] / 255, "MV08": data[6],
}) })
elif cob_id == 0x2AB and len(data) >= 7: elif cob_id == 0x2AB and len(data) >= 1:
self.latest_data[2].update({ self.latest_data[1]["PU1_STATE"] = data[0]
"PU2_STATE" : data[0],
"Conductivity_Feed" : int.from_bytes(data[1:3], 'little') / 100.0,
"Conductivity_Permeate": int.from_bytes(data[3:5], 'little') / 100.0,
"Conductivity_Product" : int.from_bytes(data[5:7], 'little') / 100.0,
})
# ========== PU2 COB-IDs ========== # ========== PU2 COB-IDs ==========
elif cob_id == 0x2AD and len(data) >= 8: elif cob_id == 0x2AD and len(data) >= 8:
self.latest_data[2].update({ self.latest_data[2].update({
@ -172,66 +164,20 @@ class CANBackend:
"Pump_sp": int.from_bytes(data[6:8], 'little') / 100.0, "Pump_sp": int.from_bytes(data[6:8], 'little') / 100.0,
}) })
elif cob_id == 0x2B1 and len(data) >= 7: elif cob_id == 0x2B1 and len(data) >= 8:
data = list(data)
self.latest_data[2].update({ self.latest_data[2].update({
"MV02": 100 * data[0] / 255, "MV02": data[0],
"MV03": 100 * data[1] / 255, "MV03": data[1],
"MV04": 100 * data[2] / 255, "MV04": data[2],
"MV05": 100 * data[3] / 255, "MV05": data[3],
"MV06": 100 * data[4] / 255, "MV06": data[4],
"MV07": 100 * data[5] / 255, "MV07": data[5],
"MV08": 100 * data[6] / 255, "MV08": data[6],
"MV09": data[7],
}) })
# elif cob_id == 0x1B9 and len(data) >= 6: elif cob_id == 0x2B2 and len(data) >= 1:
# self.latest_data[1].update({ self.latest_data[2]["PU2_STATE"] = data[0]
# "Conductivity_Feed": int.from_bytes(data[0:2], 'little'),
# "Conductivity_Permeate": int.from_bytes(data[2:4], 'little'),
# "Conductivity_Product": int.from_bytes(data[4:6], 'little'),
# })
# elif cob_id == 0x1BA and len(data) >= 6:
# self.latest_data[1].update({
# "Temperature_Feed": int.from_bytes(data[0:2], 'little'),
# "Temperature_Permeate": int.from_bytes(data[2:4], 'little'),
# "Temperature_Product": int.from_bytes(data[4:6], 'little'),
# })
# elif cob_id == 0x2B2 and len(data) >= 1:
# self.latest_data[2]["PU2_STATE"] = data[0]
elif cob_id == 0x2B2 and len(data) >= 7:
self.latest_data[2].update({
"PU2_STATE" : data[0],
"Conductivity_Feed" : int.from_bytes(data[1:3], 'little') / 100.0,
"Conductivity_Permeate": int.from_bytes(data[3:5], 'little') / 100.0,
"Conductivity_Product" : int.from_bytes(data[5:7], 'little') / 100.0,
})
# # ========== PU1 DRIFT CHECK ==========
# if cob_id in (0x2A6, 0x2A8): # FM1 or MV03_sp updates for PU1
# mv03_sp = self.latest_data[1].get("MV03_sp")
# qdrain = self.latest_data[1].get("FM1")
# if mv03_sp is not None and qdrain is not None:
# if mv03_sp <= 0 or qdrain <= 0:
# print(f"🔇 Skipping PU1 drift check (idle) → MV03_sp: {mv03_sp:.2f}, Qdrain: {qdrain:.2f}")
# elif detect_mv03_drift(mv03_sp, qdrain):
# print(f"⚠️ Drift detected on PU1 → MV03_sp: {mv03_sp:.2f} vs Qdrain: {qdrain:.2f}")
# else:
# print(f"✅ No drift on PU1 → MV03_sp: {mv03_sp:.2f} vs Qdrain: {qdrain:.2f}")
# # ========== PU2 DRIFT CHECK ==========
# if cob_id in (0x2AD, 0x2AF): # FM1 or MV03_sp updates for PU2
# mv03_sp = self.latest_data[2].get("MV03_sp")
# qdrain = self.latest_data[2].get("FM1")
# if mv03_sp is not None and qdrain is not None:
# if mv03_sp <= 0 or qdrain <= 0:
# print(f"🔇 Skipping PU2 drift check (idle) → MV03_sp: {mv03_sp:.2f}, Qdrain: {qdrain:.2f}")
# elif detect_mv03_drift(mv03_sp, qdrain):
# print(f"⚠️ Drift detected on PU2 → MV03_sp: {mv03_sp:.2f} vs Qdrain: {qdrain:.2f}")
# else:
# print(f"✅ No drift on PU2 → MV03_sp: {mv03_sp:.2f} vs Qdrain: {qdrain:.2f}")
except Exception as e: except Exception as e:
print(f"[TPDO PARSE ERROR] {e}") print(f"[TPDO PARSE ERROR] {e}")