[Python 3] Cross correlation

Cross Correlation ?

In signal processing, cross-correlation is a measure of similarity of two series as a function of the displacement of one relative to the other. This is also known as a sliding dot product or sliding inner-product. It is commonly used for searching a long signal for a shorter, known feature. It has applications in pattern recognition, single particle analysis, electron tomography, averaging, cryptanalysis, and neurophysiology.

source link

https://en.wikipedia.org/wiki/Cross-correlation


## Python code
test_crosscorrelation.py

--------------------------------------------------------
import numpy as np
import matplotlib.pyplot as plt
import math

# npts = 500
x = np.linspace(0, 50, 500)

npts = len(x)
y1 = 5 * np.sin(x/2) + np.random.randn(npts)
# y2 = 5 * np.cos(x/2) + np.random.randn(npts)
y2 = 5 * np.sin(x/2) + np.random.randn(npts)

def crocor(x,y):
npts = len(x)
lags = np.arange(-npts + 1, npts)
# Cross covariance
ccov = np.correlate((y1 - y1.mean()), y2 - y2.mean(), mode='full')
# Cross correlation coefficient
ccor = ccov / (y1.std() * y2.std() * npts)
return (ccor, lags)

# ################3
# Cross correlation coefficient by user defined function
# numpy.correlate (Cross correlate) calculates the similarity
# between two vectors/signals as a function of lag
# Seems it's wrong
pcc = np.correlate( (y1-y1.mean())/np.std(y1)/len(y1) , (y2-y2.mean())/np.std(y2) , mode='full')

# Plot
ccor, lags = crocor(y1,y2)
fig, axs = plt.subplots(nrows=3)
fig.subplots_adjust(hspace=0.4)
ax = axs[0]
ax.plot(x, y1, 'b', label='y1')
ax.plot(x, y2, 'r', label='y2')
ax.set_ylim(-10, 10)
ax.legend(loc='upper right', fontsize='small', ncol=2)

ax = axs[1]
ax.plot(lags, ccor)

ax.set_ylim(-1.1, 1.1)
ax.set_ylabel('cross-correlation')
ax.set_xlabel('lag of y1 relative to y2')

ax = axs[2]
ax.plot(lags,pcc)

maxlag = lags[np.argmax(ccor)]
# print("max correlation is at lag %d" % maxlag)
# print(np.argmax(ccor))
plt.show()


# ############################
# Cross correlation coefficient according to the formula
# In signal processing ,
# numpy.correlate (Cross correlate) calculates the similarity between two vectors/signals
# as a function of lag
aa = np.cov(y1,y2, bias=True) # Without the option 'bias=True', numpy.cov calculates the sample covariance
print('Cross correlate coefficient')
print('Formula calculation :', (aa[0][1]/(y1.std() * y2.std()) ))
print('Using correlate function : ', np.max(ccor))
print('Numpy : ', np.max(pcc))

# #######################
# Mean Square Error
# mse = ((y1-y2)**2).sum() / len(y1)
# # Root Mean Square Error
# rmse = math.sqrt(mse)
# print(mse)
# RMSE by numpy
print('RMSE: ' , np.sqrt(np.mean((y1-y2)**2)) )

-------------------------------------------------------------------------------

[Python 3] Install Dlib library

1. dlib
http://dlib.net/
https://pypi.org/project/dlib/

2. cmake
https://pypi.org/project/cmake/

Before installing dlib library on Windows OS, it is required to install Visual Studio for use of C++ library and CMake.

For installing CMake, pip install cmake

[Python 3] Notepad++ setting for Python

To run python code on notepad++, we need a Menu : PlugIns > NppExec > Excecute setting

After selecting Execute... (F6) on the popup menu, set up the commands as one of below.

[Python 3] ETAS INCA interface

See the link.

https://stackoverflow.com/questions/31092720/inca-open-experiment-python

[Python 3] Pyinstaller and PyQt5 : Failed to execute script 'file name'

If we have this error 'Failed to execute script' when executing an exe file, then try below and recompile it .

It is necessary to add the path of required dlls file to the system path.

1. Install Windows SDK(or visual studio 2015), and add the path to the system 'Envrionment Variables'
Windows SDK (Korean) download

For example,

     C:\Program Files (x86)\Windows Kits\10\Redist\ucrt\DLLs\x86

2. Add the path of PyQt5 bin folder to the system 'Envrionment Variables'
For example,

       C:\Program Files\Python36\Lib\site-packages\PyQt5\Qt\bin

or directly modify xxxxx.spec file which was created after compiling, and re-compile with xxxxx.spec file.


After this, compile python code with pyinstaller, and put ui file( GUI code created by PyQt5) in the same folder as exe file. 

: pyinstaller -F --noconsole 'filename.py'

[Python 3] IPG CarMaker Automation w/ Vector CANoe and Powersupply for HIL

# Main function

• Automate IPG CarMaker execution
• Switch on/Off powersupply output
• Start/Stop data logging of CANoe 
• Open/Close EyeQClient

1. GUI by PyQt5
automation.ui and TestAutomation_pyqt.py

2. GUI by tkinter
TestAutomation_sw.py

[Python 3] CANoe via COM API

Example code


# coding: utf-8
#"""API for setup/usage of Canoe COM Client interface.
#"""
# --------------------------------------------------------------------------
# Standard library imports
import os
import sys
import subprocess
# import win32com.client
import time
import threading
from win32com.client import *
from win32com.client.connect import *



# Vector Canoe Class
class CANoe:
    def __init__(self):
        self.application = None
        # check if there is any instance of CANoe process
        # output = subprocess.check_output('tasklist', shell=True)
        # if CANoe process is still available, kill the process
        # if "CANoe32.exe" in str(output):
        #     os.system("taskkill /im CANoe32.exe /f 2>nul >nul")

        # re-dispatch object for CANoe Application
        self.application = win32com.client.DispatchEx("CANoe.Application")
        self.ver = self.application.Version
        print('Loaded CANoe version ',
            self.ver.major, '.',
            self.ver.minor, '.',
            self.ver.Build, '...')#, sep,''

        self.Measurement = self.application.Measurement.Running
        print(self.Measurement)


    def open_simulation(self, cfgname):
        # open CANoe simulation
        if (self.application != None):
            # check for valid file and it is *.cfg file
            if os.path.isfile(cfgname) and (os.path.splitext(cfgname)[1] == ".cfg"):
                self.application.Open(cfgname)
            else:
                raise RuntimeError("Can't find CANoe cfg file")
        else:
            raise RuntimeError("CANoe Application is missing,unable to open simulation")

    def close_simulation(self):
        # close CANoe simulation
        if (self.application != None):
            self.stop_Measurement()
            self.application.Quit()

        # make sure the CANoe is close properly, otherwise enforce taskkill
        output = subprocess.check_output('tasklist', shell=True)

        if "CANoe32.exe" in str(output):
            os.system("taskkill /im CANoe32.exe /f 2>nul >nul")

        self.application = None

    def start_Measurement(self):
        retry = 0
        retry_counter = 5
        # try to establish measurement within 20s timeout
        while not self.application.Measurement.Running and (retry < retry_counter):
            self.application.Measurement.Start()
            time.sleep(1)
            retry += 1
        if (retry == retry_counter):
            raise RuntimeWarning("CANoe start measuremet failed, Please Check Connection!")

    def stop_Measurement(self):
        if self.application.Measurement.Running:
            self.application.Measurement.Stop()
        else:
            pass

    def get_EnvVar(self, var):

        if (self.application != None):
            result = self.application.Environment.GetVariable(var)
            return result.Value
        else:
            raise RuntimeError("CANoe is not open,unable to GetVariable")

    def set_EnvVar(self, var, value):
        result = None

        if (self.application != None):
            # set the environment varible
            result = self.application.Environment.GetVariable(var)
            result.Value = value

            checker = self.get_EnvVar(var)
            # check the environment varible is set properly?
            while (checker != value):
                checker = self.get_EnvVar(var)
        else:
            raise RuntimeError("CANoe is not open,unable to SetVariable")

    def get_SigVal(self, channel_num, msg_name, sig_name, bus_type="CAN"):
        # """
        # @summary Get the value of a raw CAN signal on the CAN simulation bus
        # @param channel_num - Integer value to indicate from which channel we will read the signal, usually start from 1,
                             # Check with CANoe can channel setup.
        # @param msg_name - String value that indicate the message name to which the signal belong. Check DBC setup.
        # @param sig_name - String value of the signal to be read
        # @param bus_type - String value of the bus type - e.g. "CAN", "LIN" and etc.
        # @return The CAN signal value in floating point value.
                # Even if the signal is of integer type, we will still return by
                # floating point value.
        # @exception None
        # """
        if (self.application != None):
            result = self.application.GetBus(bus_type).GetSignal(channel_num, msg_name, sig_name)
            return result.Value
        else:
            raise RuntimeError("CANoe is not open,unable to GetVariable")

    def get_SysVar(self, ns_name, sysvar_name):

        if (self.application != None):
            systemCAN = self.application.System.Namespaces
            sys_namespace = systemCAN(ns_name)
            sys_value = sys_namespace.Variables(sysvar_name)
            return sys_value.Value
        else:
            raise RuntimeError("CANoe is not open,unable to GetVariable")

    def set_SysVar(self, ns_name, sysvar_name, var):

        if (self.application != None):
            systemCAN = self.application.System.Namespaces
            sys_namespace = systemCAN(ns_name)
            sys_value = sys_namespace.Variables(sysvar_name)
            sys_value.Value = var
            # print(sys_value)
            # result = sys_value(sys_name)
            #
            # result = var
        else:
            raise RuntimeError("CANoe is not open,unable to GetVariable")

    def get_all_SysVar(self, ns_name):

        if (self.application != None):
            sysvars=[]
            systemCAN = self.application.System.Namespaces
            sys_namespace = systemCAN(ns_name)
            sys_value = sys_namespace.Variables
            for sys in sys_value:
                sysvars.append(sys.Name)
                sysvars.append(sys.Value)
            return sysvars
        else:
            raise RuntimeError("CANoe is not open,unable to GetVariable")

#On Event Thread
class Event_Job(threading.Thread):

    def __init__(self, name,var,event):#*args, **kwargs
        super(Event_Job, self).__init__()#*args, **kwargs
        self.__flag = threading.Event()     #
        self.__flag.set()       # 设置为True
        self.__running = threading.Event()      #
        self.__running.set()      #

        self.name = name
        self.var = var
        self.event = event

    def run(self):
        pythoncom.CoInitialize()
        self.app = DispatchEx('CANoe.Application')
        self.systemCAN = self.app.System.Namespaces
        self.sys_namespace = self.systemCAN(self.name)
        self.sys_value = self.sys_namespace.Variables
        self.result = self.sys_value(self.var)
        WithEvents(self.result, self.event)

        while self.__running.isSet():
            self.__flag.wait()      #
            # self.func(self.txt)
            # print time.time()
            # pythoncom.CoInitialize()
            pythoncom.PumpWaitingMessages()
            time.sleep(0.01)
            # pythoncom.CoUninitialize()

    def pause(self):
        self.__flag.clear()     #

    def resume(self):
        self.__flag.set()    #

    def stop(self):
        self.__flag.set()       #
        self.__running.clear()        #
        pythoncom.CoUninitialize()

class MFL_volplus_Events(object):
    def __init__(self):
        pass

    #"""Handler for CANoe var events"""
    def OnChange(self,value):
        # self.Changed = True
        print("< MFL_volplus_Events var change>")
        # print(self.Name)
        print(value)

class MFL_volminus_Events(object):
    def __init__(self):
        pass

    #"""Handler for CANoe var events"""
    def OnChange(self,value):
        # self.Changed = True
        print("< MFL_volminus_Events var change>")
        # print(self.Name)
        print(value)


if __name__ == '__main__':
os.chdir("C:\Program Files\Vector CANoe 10.0\Exec64")
os.popen("canoe64 -regserver").readlines()

app = CANoe()
time.sleep(5)
app.start_Measurement()


# #Regserver COM API
# os.chdir("C:\Program Files (x86)\Vector CANoe 9.0\Exec32")
# os.popen("canoe32 -regserver").readlines()
# #or
# os.chdir("C:\Program Files\Vector CANoe 10.0\Exec64")
# os.popen("canoe64 -regserver").readlines()

# app = CANoe()
# time.sleep(5111)
# app.start_Measurement()
# time.sleep(5)
# app.stop_Measurement()
# time.sleep(1111)
# varnames = app.get_all_SysVar("mfl")
#
# time.sleep(1)
# vol_plus = Event_Job("mfl","vol_plus",MFL_volplus_Events)
# vol_plus.start()
#
# vol_minus = Event_Job("mfl","vol_minus",MFL_volminus_Events)
# vol_minus.start()
#
# time.sleep(31)
# app.set_SysVar("mfl","vol_plus",1)
#
# print(app.get_SysVar("mfl","vol_plus"))
#
# vol_plus.stop()
# vol_minus.stop()


Python code download