# V1.0 Maintain Ah on restart # @reboot sleep 20 && sudo python /var/www/html/BatteryMon_3.py & # 24.6.23 Voltage measurement for Controller 2 # DEBUG SETTINGS #************************************************************************************** DEBUG_INFO = True # General information prints (state changes, charge start/stop) DEBUG_DETAIL = False # Detailed debug prints (measurements, timers, technical data) def debug_info(*args, **kwargs): """Print general information if DEBUG_INFO is enabled""" if DEBUG_INFO: print(*args, **kwargs) def debug_detail(*args, **kwargs): """Print detailed debug information if DEBUG_DETAIL is enabled""" if DEBUG_DETAIL: print(*args, **kwargs) # IMPORT DEFINITIONS #************************************************************************************** import platform # for platform.system() check from pathlib import Path import sys import os # for CSV file existence check import time sys.path.append(r"C:\Users\kauko\env\Lib\site-packages") import serial import datetime # for time management from datetime import timedelta # for time management import matplotlib # Styles matplotlib.use('Agg') import pylab # for graph saving import matplotlib.pyplot as plt # graphs library import matplotlib.dates as mdates # formatting the graph plt.style.use('ggplot') # graph style # Removed drawnow import as it's not needed for non-interactive plotting import atexit from csv import writer # for CSV file control import csv import logging import traceback import signal # FILE DEFINITIONS #************************************************************************************** if platform.system() == "Windows": base_path = Path("C:\Apache24\htdocs\Data") # Vaihda halutuksi else: base_path = Path("/var/www/html/Data") Controlfile = base_path / 'Control.csv' # CSV file for control data StatusFile = base_path / 'Status.csv' # CSV file for status data DataFile = base_path / 'Tiedosto.csv' # CSV file for log data GraphFile = base_path / 'GraphFile.csv' # CSV file for graph data Graph = base_path / 'Graph6' # CSV file for graph data # HW MEASUREMENT RELATED DEFINITIONS #************************************************************************************** # Voltage measurement calibration factor VoltageFactor = 0.0189 #Proto1 12V corresponds 2,99V in ADC and as 635 binary with Vref 4.5V #VoltageFactor = 0.0195 #Proto2 # DC virran mittaus 50 (proto1) or 20 (proto2) is the multiplier in Arduino IdcFactor = 0.715/20 # = 0,03575 Controller V2 #IdcFactor = 0.749/50 # = 0,01498 Controller V1 Current = 0.0 # Current value Voltage = 11.5 # Voltage value Energy2 = 0.0 # Energy value Energy = 0.0 # Energy value EnergyIn = 0.0 EnergyOut = 0.0 InvTrigger = 0.0 VMin = 10.5 AhValue = 300 VMin = 10 IOffset = 0 ICal = 0 # CHARGER DEFINITIONS #************************************************************************************** AhStart = 0.0 VStart = 0.0 AhStop = 0.0 VStop = 0.0 ChargeTime = timedelta() ChargeTimeTot = timedelta() GenOn = 1 # Generator sensing input 1=no AC input ChargerOffFilter = 140 # how many seconds charger relay can be off until charge is stopped ChargeAttempts = 0 # Nr of allowed charge attempts chargeStartTime = time.time() ChargeTimeTotStr = "00:00:00" ChargeSource = "None" # Track what initiated the charge: "Manual", "Generator", "AutoStart", "None" ChargeSourceTimer = None # Timer to reset ChargeSource to "None" after 140 seconds # STATE DEFINITIONS #************************************************************************************** State = "PowerOff" Alarm = "None" PrevState = "PowerOff" x2 = datetime.datetime.now() # Time stamp for graph, refreshed by message from Arduino shortTimerOn = 0 sec5Timer = 60 # time for 5 sec sampling until goes back to 120 sec graphTimer = 600 # period for stroring graph data # ARDUINO INTERFACE #************************************************************************************** # Detect the operating system and define the serial USB port accordingly, baud must be same as in Arduino def get_serial_port(): if platform.system() == "Windows": return "COM6" # Example for Windows elif platform.system() == "Linux": return "/dev/ttyACM0" # Raspberry Pi, Arduino Every, Nano is "/dev/ttyUSB0" else: raise EnvironmentError("Unsupported platform") port = get_serial_port() baud = 9600 # GRAPHS DEFINITIONS #************************************************************************************** def OwnGraph1(): plt.figure(1) plt.clf() # Clear the entire figure before redrawing fig = plt.figure(1) fig.set_figwidth(5) #Picture sizing fig.set_figheight(8) #SoC plt.subplot(311) plt.ylim(0, AhValue) plt.title("Varaustila %1.1f Ah" %Energy2, fontweight='bold', fontsize=12) plt.plot(x, values3, color='mediumseagreen', linewidth=2) plt.grid(linewidth=1) plt.axhspan(60, 300, facecolor="green", alpha=0.1) plt.axhspan(30, 60, facecolor="yellow", alpha=0.1) plt.axhspan(0, 30, facecolor="red", alpha=0.1) #plt.suptitle("xxxxx", fontsize=12) #Current plt.subplot(312) plt.title("Virta %1.1f A, (+lataus, -purku)" %Current, fontweight='bold', fontsize=12) if ((max(values2) < 10) & (min(values2) > -10)): plt.ylim(-10, 10) plt.plot(x, values2, color='lightcoral', linewidth=2) #Voltage plt.subplot(313) plt.title("Jannite %1.1f V, GenOn %1.0f" % (Voltage, GenOn), fontweight='bold', fontsize=12) plt.ylim(10, 16) plt.plot(x, values, color='steelblue', linewidth=2) plt.grid(linewidth=1) plt.axhspan(14.8, 16, facecolor="red", alpha=0.1) plt.axhspan(VMin, 14.8, facecolor="green", alpha=0.1) plt.axhspan(10, VMin, facecolor="red", alpha=0.1) fig.tight_layout() plt.gcf().autofmt_xdate() #plt.show() # FUNCTIONS #************************************************************************************** def setup_logging(): """Setup logging to capture script termination reasons""" log_file = '/var/www/html/BatteryMon_errors.log' logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler(log_file), logging.StreamHandler() # Also log to console ] ) logging.info("BatteryMon script started") def log_termination_reason(reason, exc_info=None): """Log the reason for script termination""" logging.error(f"Script terminating: {reason}") if exc_info: logging.error(f"Exception details: {traceback.format_exc()}") def signal_handler(signum, frame): """Handle system signals that might terminate the script""" signal_names = { signal.SIGTERM: "SIGTERM (termination request)", signal.SIGINT: "SIGINT (Ctrl+C)", signal.SIGKILL: "SIGKILL (force kill)", signal.SIGHUP: "SIGHUP (hangup)" } reason = signal_names.get(signum, f"Unknown signal {signum}") log_termination_reason(f"Received signal: {reason}") updateGraphFile() # Save data before exit sys.exit(0) def cleanup_on_exit(): """Cleanup function to run on normal exit""" log_termination_reason("Normal script exit") updateGraphFile() # Save data before exit def readArduino (Energy): global EnergyIn global EnergyOut global Energy2 # This is new energy value, Energy is the old value given via function call global Time0 global Voltage global VoltageFactor global Current global x2 global GenOn try: data = serialArduino.readline()[:-2] #the last bit gets rid of the new-line chars b = data koe = b.split(b" ")[0] if koe == b"Graph1": Vdc = b.split(b" ")[1] Idc = b.split(b" ")[2] GenOn = int(b.split(b" ")[3]) #GenOn = 0 #print ("Idendified Arduino values:",Vdc,Idc,GenOn) Voltage1 = int(Vdc) Voltage = round(Voltage1*VoltageFactor,2) Current1 = int(Idc) Current = round(IOffset+ICal*Current1*IdcFactor,3) # remove 1 bit noise if (Current1*Current1) == 1: Current = 0 Time = (datetime.datetime.now()-Time0).total_seconds() Energy2 = round(Energy+Current*Time/3600,2) #Ensure that energy is not negative or above battery capacity if Energy2 < 0.05: Energy2 = 0.1 if Energy2 > (AhValue - 2): Energy2 = (AhValue - 2) Time0 = datetime.datetime.now() EnergyChange = round(Current*Time/3600,3) if EnergyChange > 0: EnergyIn = round(EnergyIn+EnergyChange,3) else: EnergyOut = round(EnergyOut-EnergyChange,3) # print ("Idendified Arduino value:",koe,Voltage,Current,Energy2) time_str = str(datetime.datetime.now().replace(microsecond=0)) x2 = datetime.datetime.strptime(time_str, '%Y-%m-%d %H:%M:%S') updateStatusfile(Voltage, Current, Energy2, EnergyIn, EnergyOut, x2) updateGraph2(Voltage, Current, Energy2, x2) updateDataFile(Voltage, Current, Energy2, x2) return Energy2 except serial.SerialException as e: logging.error(f"Serial communication error: {e}") log_termination_reason(f"Serial error: {e}", exc_info=True) raise except ValueError as e: logging.error(f"Data parsing error: {e}, raw data: {data}") return Energy # Return previous energy value on parse error except Exception as e: logging.error(f"Unexpected error in readArduino: {e}") log_termination_reason(f"Unexpected error in readArduino: {e}", exc_info=True) raise def updateGraph2(Voltage, Current, Energy, x2): global values, values2, values3 if 0 <= Voltage <= 10240000: values.append(Voltage) values.pop(0) values2.append(Current) values2.pop(0) values3.append(Energy) values3.pop(0) x.append(x2) x.pop(0) OwnGraph1() plt.savefig(Graph, bbox_inches="tight") return Voltage, Current, Energy else: print("Invalid! Voltage out of range") return None def sendToArduino( Label, Data ): #Add message header/terminator as defined in Arduino SW Task = "<" + Label + " " + Data + ">" print ("Sent command 1:",Task) serialArduino.write(Task.encode()) return; def initControlData (): global ControlData0 with open(Controlfile, 'r') as f: reader = csv.reader(f) ControlData0 = list(reader) lines = len(ControlData0) #print ("Control data lines from file ",lines) for ind in range(0,lines): ControlData0[ind][1] = "xx" #Read data from CSV file cell, compare to previous and send delta to Arduino def readControlfile(): global ControlData0, Energy2, VMin, AhValue, IOffset, ICal, VStart, AhStart, VStop, AhStop, InvTrigger, ChargeTime, ChargeTimeTot with open(Controlfile, 'r') as f: reader = csv.reader(f) ControlData = list(reader) if ControlData0 != ControlData: #compare first full arrays, if delta compare line by line for ind, control in enumerate(ControlData): #Then compare line by luine if ControlData0[ind] != control: Label, Data = control[0], control[1] if Label == "Energy" and float(Data) >= 2: #Update energy only if it is bigger that 2 Energy2 = float(Data) updateControlData0("Energy",1) #set energy calibration to NA value updateControlfile() if Label == "AhValue": AhValue = float(Data) # CURRENT MEASUREMENT CALIBRATION if Label == "IOffset": IOffset = float(Data) if Label == "ICal": ICal = float(Data) # BATTERY PROTECTION PARAMETERS if Label == "VMin": # voltage to stop discharge VMin = float(Data) # GENERATOR COUNTER RESET if Label == "Command" and Data == "ResetGen1": # voltage to stop discharge ChargeTime = timedelta(seconds=0) # Reset to 0 print("ChargeTime reset to: ", ChargeTime) updateControlData0("Command",0) #set energy calibration to NA value updateControlfile() if Label == "Command" and Data == "ResetGen2": # voltage to stop discharge ChargeTimeTot = timedelta(seconds=0) # Reset to 0 print("ChargeTimeTot reset to: ", ChargeTimeTot) updateControlData0("Command",0) #set energy calibration to NA value updateControlfile() # CHARGER AUTOSTART PARAMETERS if Label == "VStart": # voltage to start charger VStart = float(Data) if Label == "AhStart": # Ah to start charger AhStart = float(Data) if Label == "VStop": # Voltage to stop charger VStop = float(Data) if Label == "AhStop": # Ah to stop charger AhStop = float(Data) if Label == "InvTrigger": InvTrigger = float(Data) elif Label == "Run" and Data == "Stop": updateGraphFile() print("Stopped by Run stop command") sys.exit() sendToArduino(Label, Data) ControlData0 = ControlData return "NA" return "NA" #STORES THE DATA ABOUT ENERGY, CURRENT AND VOLTAGE #************************************************************************************** def createDataFile(): if os.path.isfile(DataFile): print("Data file exists") else: print("Create new Data file") with open(DataFile, "w", newline="") as f: writer = csv.writer(f) writer.writerow(['Date', 'Ah', 'I', 'V']) writer.writerow([0, 0, 0]) def updateDataFile (v,c,ah,x2): with open(DataFile, "a", newline='', encoding='utf-8') as f: writer = csv.writer(f) writer.writerow([x2,ah,c,v]) #STORES THE DATA SHOWN IN "TILASTOT" VIEW #************************************************************************************** def format_timedelta(td): total_seconds = int(td.total_seconds()) hours, remainder = divmod(total_seconds, 3600) minutes, seconds = divmod(remainder, 60) return f"{hours:02}:{minutes:02}:{seconds:02}" def updateStatusfile(v, c, e, eIn, eOut, timestamp): global ChargeTimeTot, ChargeTime, ChargeSource with open(StatusFile, "w", newline='', encoding='utf-8') as f: writer = csv.writer(f) #print("XXXXXX Update Status") #print("Chargetime2 ", ChargeTime) writer.writerows([ ["Voltage/V", v], ["Current/A", c], ["Energy/Ah", e], ["EnergyIn", eIn], ["EnergyOut", eOut], ["Timestamp", timestamp], ["ChargeTime", format_timedelta(ChargeTime)], ["ChargeTimeTot", format_timedelta(ChargeTimeTot)], ["State", State], ["Alarm", Alarm], ["ChargeSource", ChargeSource] ]) # At SW start, this reads the last status data from the file def readStatusfile(): global EnergyIn, EnergyOut, Energy2, ChargeTime, ChargeTimeTot data = {} with open(StatusFile, 'r') as f: reader = csv.reader(f) for row in reader: if len(row) == 2: label, value = row data[label] = value try: Energy2 = float(data.get("Energy/Ah", 0.0)) EnergyIn = float(data.get("EnergyIn", 0.0)) EnergyOut = float(data.get("EnergyOut", 0.0)) ChargeTime = parse_timedelta(data.get("ChargeTime", "00:00:00")) ChargeTimeTot = parse_timedelta(data.get("ChargeTimeTot", "00:00:00")) except ValueError as e: print("Error parsing values:", e) def parse_timedelta(time_str): try: time_parts = list(map(int, time_str.split(':'))) if len(time_parts) == 3: return timedelta(hours=time_parts[0], minutes=time_parts[1], seconds=time_parts[2]) elif len(time_parts) == 4: return timedelta(hours=time_parts[0], minutes=time_parts[1], seconds=time_parts[2], milliseconds=time_parts[3]) except ValueError: print("Invalid time format:", time_str) return timedelta() def updateControlData0(Labelx, Datax): global ControlData0 for control in ControlData0: if control[0] == Labelx: control[1] = Datax print("Python changes control data:", Labelx, Datax) break return def get_control_value(label): for control in ControlData0: if control[0] == label: return float(control[1]) return None #This function writes ControlData0 back into file.This can be used to change controls from Python! def updateControlfile(): global ControlData0 with open(Controlfile, "w", newline='', encoding='utf-8') as f: writer = csv.writer(f) writer.writerows(ControlData0) #print("Controlfile updated") def updateGraphFile (): global values, values2, values3, x with open(GraphFile, "w", newline='', encoding='utf-8') as f: writer = csv.writer(f) for ind in range(0,XLength): writer.writerow([x[ind],values[ind],values2[ind],values3[ind]]) def readGraphFile (): # File opened with w for adding data global values, values2, values3, x with open(GraphFile, "r", newline='', encoding='utf-8') as f: reader = csv.reader(f) GraphData = list(reader) # Check that oldest time stamp is less than 28 hours and data is then relevant Time1 = datetime.datetime.strptime(GraphData[1][0], '%Y-%m-%d %H:%M:%S') Timeperiod = ((datetime.datetime.now() - Time1)/3600).seconds if Timeperiod > 28: return for ind in range(0,XLength): time = GraphData[ind][0] x[ind] = datetime.datetime.strptime(time, '%Y-%m-%d %H:%M:%S') values[ind] = float(GraphData[ind][1]) values2[ind] = float(GraphData[ind][2]) values3[ind] = float(GraphData[ind][3]) #print ("Max value", min(values3)) # Checks current and voltage and changes state accordingly #************************************************************************************** def startCharge (source="None"): global State, shortTimerOn, chargeStartTime, ChargeAttempts, GenTimer, ChargeSource, ChargeSourceTimer State = "Charge" ChargeSource = source # Store what initiated the charge ChargeSourceTimer = None # Reset the cooldown timer when charging starts GenTimer = time.time() # Timer to measure how long ChargeAttempts -= 1 debug_info(f"Charge started by: {source}, attempts remaining: {ChargeAttempts}") chargeStartTime = time.time() readControlfile() updateControlData0("Load",1) updateControlData0("Charge",1) updateControlData0("GraphT",5) updateControlfile() def stopCharge (source="Unknown"): global State, ChargeAttempts, ChargeSource, ChargeSourceTimer State = "PowerOn" # Don't change ChargeSource immediately - let the timer handle it # This prevents immediate restart of generator detection ChargeSourceTimer = time.time() # Start the cooldown timer debug_info(f"Charge stopped by: {source}, ChargeSource cooldown started") readControlfile() updateControlData0("Charge",0) updateControlData0("GraphT",5) updateControlfile() #if ChargeAttempts >= 1: #time.sleep(10) #startCharge() return def StateMachine(): global State, Alarm, Voltage, Current, Load, Charge, ChargeAttempts, ChargerOffFilter, GenTimer, GenStopTimer, GenOn # tarvitaan jos muutetaan PrevState = State #print(f"Condition{Energy2} {AhStart} {Voltage} {VStart}") if State == "PowerOff": if get_control_value("Load") == 1 and Alarm != "VMin": State = "PowerOn" elif Alarm == "VMin": State = "PowerOff" elif State == "PowerOn": if get_control_value("Load") == 0: State = "PowerOff" elif get_control_value("Charge") == 1: #Charge command by user ChargeAttempts = 3 # Allow 3 charge attempts startCharge("Manual") elif GenOn == 0 and ChargeSource == "None": # Charger has been started without user command and detected by Gen sense relay ChargeAttempts = 1 # Only one attempt startCharge("Generator") elif get_control_value("AutoStart") == 1 and (Energy2 < AhStart or Voltage <= VStart): debug_info(f"AutoStart condition met: Energy={Energy2}/{AhStart}Ah, Voltage={Voltage}/{VStart}V") ChargeAttempts = 3 startCharge("AutoStart") elif Alarm == "VMin": State = "PowerOff" elif State == "Charge": if get_control_value("AutoStart") == 1 and (Energy2 > AhStop or Voltage > VStop): debug_info(f"AutoStart stop condition met: Energy={Energy2}/{AhStop}Ah, Voltage={Voltage}/{VStop}V") stopCharge("Autostart") elif get_control_value("AutoStart") == 0 and get_control_value("Charge") == 0: stopCharge("Manual") # Counter to remove Charge state if GenOn is not active in ChargerOffFilter sec if GenOn == 1: #print("ChargeTimer1") #print("ChargeOff filtering", GenOn, (time.time() - GenTimer)) if(time.time() - GenTimer) > ChargerOffFilter: print("No charger input1") GenTimer = time.time() stopCharge("None") #updateControlData0("Charge", 0) #updateControlfile() elif GenOn == 0: GenTimer = time.time() #print("Reset ChargeOff filter", GenOn) if PrevState != State: debug_info(f"→ Siirtyi tilaan {State}") PrevState = State debug_detail(f"GenOn status: {GenOn}") # Monitor voltage and set Alarm as needed #************************************************************************************** def MonitorChargeSource(): """Monitor and reset ChargeSource after cooldown period""" global ChargeSource, ChargeSourceTimer if ChargeSourceTimer is not None and ChargeSource != "None": if (time.time() - ChargeSourceTimer) >= 140: # 140 seconds cooldown previous_source = ChargeSource ChargeSource = "None" ChargeSourceTimer = None debug_info(f"ChargeSource cooldown expired: {previous_source} -> None") def MonitorVoltage(): global State, Alarm, Voltage, VMin, VMinTimer, VMinTimer2 if Alarm != "VMin" and Voltage < VMin: if VMinTimer is None: VMinTimer = time.time() # Start the timer elif time.time() - VMinTimer >= 20: # Check if 20 seconds have passed Alarm = "VMin" print(f"Voltage {Voltage}V is below {VMin}V for 20 seconds. Alarm --> {Alarm}.") updateControlData0("Load", 0) # Go to PowerOff state updateControlfile() elif Alarm == "VMin" and Voltage >= VMin: if VMinTimer2 is None: VMinTimer2 = time.time() # Start the timer elif time.time() - VMinTimer2 >= 140:# Check if 140 seconds have passed Alarm = "None" print(f"Voltage {Voltage}V is {VMin}V for 120 seconds. Alarm --> {Alarm}.") updateControlData0("Load", 1) # Go to PowerOff state updateControlfile() else: VMinTimer = None # Reset the timer if voltage is above MinVoltage VMinTimer2 = None # Reset the timer if voltage is above MinVoltage # MAIN #************************************************************************************** if __name__ == '__main__': serialArduino = serial.Serial(port, baud, timeout=1) time.sleep(1) serialArduino.flushInput() serialArduino.flushOutput() # CONTROL DATA TABLE INITIALIZATION #************************************************************************************** XLength = 600 values = [0] * XLength values2 = [0] * XLength values3 = [0] * XLength #x = [datetime.datetime.now() + datetime.timedelta(seconds=i - XLength) for i in range(XLength)] x = [datetime.datetime.now().replace(microsecond=0) + datetime.timedelta(seconds=i - XLength) for i in range(XLength)] #plt.ion() # Tell matplotlib you want interactive mode to plot live data initControlData() # TIME VARIABLE DEFINITIONS #************************************************************************************** ChargeTime = datetime.timedelta(hours=0, minutes=0, seconds=0) Time0 = datetime.datetime.now() startTime = time.time() # Timer for 5 sec interval setting, until switch back to 120 sec shortTimerOn = 0 # Flag for 5 sec interval setting graphTime = time.time() # Timer for storing graph data # INITIALIZE TABLES #************************************************************************************** createDataFile() readStatusfile() readGraphFile() readControlfile() updateControlData0("GraphT", 5) updateControlData0("Charge", 0) updateControlfile() VMinTimer = None VMinTimer2 = None while True: readControlfile() readArduino(Energy2) MonitorVoltage() # Call the voltage monitoring function MonitorChargeSource() # Monitor ChargeSource cooldown timer StateMachine() # Counter to remove 5 sec interval setting after sec5Timer if shortTimerOn == 0 and float(ControlData0[0][1]) == 5: shortTimerOn = 1 startTime = time.time() elif shortTimerOn == 1 and (time.time() - startTime) > sec5Timer: shortTimerOn = 0 updateControlData0("GraphT", 120) updateControlfile() # Counter to store graph data after graphTimer if (time.time() - graphTime) > graphTimer: updateGraphFile() graphTime = time.time() # Update cumulative charge time if state is Charge if State == "Charge": elapsed_time = time.time() - chargeStartTime ChargeTime += timedelta(seconds=elapsed_time) ChargeTimeTot += timedelta(seconds=elapsed_time) chargeStartTime = time.time() # Reset the start time for the next interval else: chargeStartTime = time.time() # Reset the start time for the next interval # print(ChargeTime)