Das deutsche Python-Forum

Code: Alles auswählen

import can
import time
import os
import sys
import queue
import colorama
from colorama import Fore
from threading import Thread
import threading
import matplotlib
import matplotlib.pyplot as plt
import numpy as np


# Bus-channel aufbauen

try:
    bus = can.interface.Bus(channel='can0', bustype='socketcan_native')
except OSError:
    print('Cannot find PiCAN board.')
    exit()


threadtasks=True



Stopp_Charge		= 0x00200039
Start_Charge		= 0x00200038
Heartbeat_Request	= 0x0FE00000   #Can-Nachrichten für die Ladeplatine
Voltage_Request		= 0x00200040
Voltage_Reply		= 0x00002006


outfile = open('Platinenspannung_Datum.csv','w', newline='')

def start_charge():
    msg = can.Message(arbitration_id=Start_Charge,data=[],extended_id=True)
    bus.send(msg)
    print(Fore.BLUE + "\n" "Charging initiated" "\n")
    print(Fore.RESET)
start_charge()


def stopp_charge():
        msg = can.Message(arbitration_id=Stopp_Charge,data=[],extended_id=True)
        bus.send(msg)
        print("Charging stopped. Battery full. Stop program")
        global threadtasks
        threadtasks=False
        sys.exit(0)
        




def heartbeat_task(): # heartbeat thread
    global threadtasks
    while threadtasks:
        time.sleep(600)
        msg = can.Message(arbitration_id=Heartbeat_Request,data=[],extended_id=True)
        bus.send(msg)
        print(Fore.YELLOW + "\n" "Heartbeatrequest sent" "\n")
        print(Fore.RESET)


def can_rx_task():	# Receive thread
    global threadtasks
    while threadtasks:
        message = bus.recv()
        if message.arbitration_id == Voltage_Reply: 
            q.put(message)			# Put message into queue


def can_tx_task():	# Transmit thread
    global threadtasks
    while threadtasks:
        msg = can.Message(arbitration_id=Voltage_Request,data=[],extended_id=True)
        bus.send(msg) #Sent a Voltage request
        time.sleep(1)
						
heart = Thread(target = heartbeat_task)
heart.start()						
q = queue.Queue()
rx = Thread(target = can_rx_task)
rx.start()
tx = Thread(target = can_tx_task)
tx.start()


try:
    while True:
            while(q.empty() == True):	# Wait until there is a message
                pass
            message = q.get()

            s=''
            for i in range(message.dlc ):            
                s +=  '{0:x} '.format(message.data[i])	                		
			
            x = s.split() #Konvertierung in DEZ
            y = x[0]
            z = x[1]

            if len(y) == 1:
                y = '0' + y
            volt = int(z+y, 16)

            if volt > 41000:
                print(Fore.RED + "\n"'OVERHEATING')
                print(Fore.RESET)
                outfile.close()
                stopp_charge()
                
            else:
                print(volt, 'mV')
                print(volt, file = outfile) # Save data to file



	
except KeyboardInterrupt:
    print(Fore.RED + '\n\r keyboard interrupted\n')
    print(Fore.RESET)
    outfile.close()		# Close logger file
    stopp_charge()
		

Code: Alles auswählen

import can
import time
import queue
from colorama import Fore
from threading import Thread

STOP_CHARGE = 0x00200039
START_CHARGE = 0x00200038
HEARTBEAT_REQUEST = 0x0FE00000   #Can-Nachrichten für die Ladeplatine
VOLTAGE_REQUEST = 0x00200040
VOLTAGE_REPLY = 0x00002006

def send_message(bus, id):
    msg = can.Message(arbitration_id=id, data=[], extended_id=True)
    bus.send(msg)

def start_charge(bus):
    send_message(bus, START_CHARGE)
    print(Fore.BLUE + "\n" "Charging initiated" "\n")
    print(Fore.RESET)

def stopp_charge(bus):
    send_message(bus, STOP_CHARGE)
    print("Charging stopped. Battery full. Stop program")

def heartbeat_task(bus): # heartbeat thread
    while True:
        time.sleep(600)
        send_message(bus, HEARTBEAT_REQUEST)
        print(Fore.YELLOW + "\n" "Heartbeatrequest sent" "\n")
        print(Fore.RESET)

def can_rx_task(bus, queue): # Receive thread
    while True:
        message = bus.recv()
        if message.arbitration_id == VOLTAGE_REPLY: 
            queue.put(message) # Put message into queue

def can_tx_task(bus): # Transmit thread
    while True:
        send_message(bus, VOLTAGE_REQUEST)
        time.sleep(1)

def main():
    # Bus-channel aufbauen
    try:
        bus = can.interface.Bus(channel='can0', bustype='socketcan_native')
    except OSError:
        print('Cannot find PiCAN board.')
        return
    
    msg_queue = queue.Queue()
    heart = Thread(target=heartbeat_task, args=(bus, ))
    heart.deamon = True
    heart.start()
    rx = Thread(target=can_rx_task, args=(bus, msg_queue))
    rx.deamon = True
    rx.start()
    tx = Thread(target=can_tx_task, args=(bus, ))
    tx.deamon = True
    tx.start()

    try:
        start_charge(bus)
        with open('Platinenspannung_Datum.csv','w', newline='') as outfile:
            while True:
                message = msg_queue.get()
                volt = message.data[0] + message.data[1] * 256
                print(volt, 'mV')
                print(volt, file=outfile) # Save data to file
                if volt > 41000:
                    break
            print(Fore.RED + "\n"'OVERHEATING')
            print(Fore.RESET)
    except KeyboardInterrupt:
        print(Fore.RED + '\n\r keyboard interrupted\n')
        print(Fore.RESET)
    finally:
        stopp_charge()

if __name__ == '__main__':
    main()

Code: Alles auswählen

import can
from colorama import Fore

STOP_CHARGE = 0x00200039
START_CHARGE = 0x00200038
HEARTBEAT_REQUEST = 0x0FE00000   #Can-Nachrichten für die Ladeplatine
VOLTAGE_REQUEST = 0x00200040
VOLTAGE_REPLY = 0x00002006

def send_message(bus, id, periodic=None):
    msg = can.Message(arbitration_id=id, data=[], extended_id=True)
    if periodic is None:
        bus.send(msg)
    else:
        bus.send_periodic(msg, periodic)

def start_charge(bus):
    send_message(bus, START_CHARGE)
    print(Fore.BLUE + "\n" "Charging initiated" "\n")
    print(Fore.RESET)

def stop_charge(bus):
    send_message(bus, STOP_CHARGE)
    print("Charging stopped. Battery full. Stop program")

def heartbeat(bus):
    send_message(bus, HEARTBEAT_REQUEST, periodic=600)

def can_tx(bus): # Transmit thread
    send_message(bus, VOLTAGE_REQUEST, periodic=1)

def can_rx(bus): # Receive thread
    while True:
        message = bus.recv()
        if message.arbitration_id == VOLTAGE_REPLY: 
            yield message

def main():
    # Bus-channel aufbauen
    try:
        bus = can.interface.Bus(channel='can0', bustype='socketcan_native')
    except OSError:
        print('Cannot find PiCAN board.')
        return

    try:
        start_charge(bus)
        heartbeat(bus)
        can_tx(bus)
        with open('Platinenspannung_Datum.csv', 'w', newline='') as outfile:
            for message in can_rx(bus):
                volt = message.data[0] + message.data[1] * 256
                print(volt, 'mV')
                print(volt, file=outfile) # Save data to file
                if volt > 41000:
                    break
            print(Fore.RED + "\n"'OVERHEATING')
            print(Fore.RESET)
    except KeyboardInterrupt:
        print(Fore.RED + '\n\r keyboard interrupted\n')
        print(Fore.RESET)
    finally:
        stop_charge()

if __name__ == '__main__':
    main()