file SpvPython.py 

This file is the python interface to SPV, it enable schedule simulator event, doing Wait([event,time]), declare signal, generator and more. this is the main file that enable using python as a verification environment.

​

from ctypes import *
from enum import Enum
import sys
import threading
import time
import datetime
import SPV 

class SPSig():
   def __init__(self, SigFullPath):
       self.m_Name = SigFullPath;
       self.SigId  = SPV.InitSig(SigFullPath);

    def  Set(self,value):
       SPV.Set2Sig(int(value),self.SigId);

    def    Val64(self):
       val = SPV.SigVal64(self.SigId);
       return val;

    def    Val32(self):
       val = SPV.SigVal32(self.SigId);
       return val;

    def    Name(self):        
       return  self.m_Name;

#need to implement
class    SpvGen():
   def __init__(self, GenGuiName):
       self.m_Name = GenGuiName;
       self.GenId    = SPV.GetGuiGen(GenGuiName);

    def  Gen(self):
       val = SPV.Gen(self.GenId);
       return genVal;

​

class SPEvType(Enum):
   AtPos = 0
   AtNeg = 1
   AtChange = 2
   AtEqual = 3    

    def  Uint(self):
       return self.value

    def  __str__(self):
       return 'Event'.format(self.value)

​

​

#Short key for any using.
AtPos         = SPEvType.AtPos
AtNeg         = SPEvType.AtNeg
AtChange    = SPEvType.AtChange
AtEqual        = SPEvType.AtEqual
       

class  SPSigEvent():
   def __init__(self, sig : SPSig,ev : SPEvType):
       self.m_Sig        = sig;
       self.m_EvType    = 1;#ev;  SHMUEL need to fix why self.m_EvType = ev is not good.
       self.m_IsTime    = 0;
       self.m_SpvEvId    = SPV.InitSigEv(self.m_Sig.SigId,self.m_EvType);#SpvEvId comming from C code (of SPV).
   

#work around of problem with print (Bacause SPV stole any print to ActiveOut class), use PR that print output to stderr
#which work well
pyLog = open('Python.log', 'w');
def   PR(s):
   print(str(s),file=sys.stderr);
   pyLog.write(str(s) + "\n");    
   pyLog.flush();
   return
   
###############        Python-Spv    Scheduler
#these are the field that user should supply for any thread that used as SPV TCM 
#Wait(Event) and any other timming ability that comming from SPV.
class    ThreadParam():
       def __init__(self,DefFunction,name: str,id: int):
           self.ThreadFunction =  DefFunction;
           self.m_ThreadName      =  name;
           self.ThreadId        =  id;

#this class pass as is to SPV for any wait command. SPV create AtomicProcess CB  StartAProc for always.
#when the event occure, SPV mark     m_IsEvOccure = 1. the Scheduler function SPSchedulerCenter::Run
#run the event that m_IsEvOccure = 1, if there is again Wait we SPV check if the CB is for the same event.
#if the answer is True, it is just wait to next event occuration, else it kill the prev event and make new one.
#class OneCEvent(Structure):
#    _fields_ = [
#        ("m_SigInd",    c_uint),#the pointer of the signal that the event (Pos,Neg<change) wait for it.       
#        ("m_EventType", c_uint),#Pos,Neg,change
#        ("m_IsTime",    c_uint),#if is time it is Wait (time)
#        ("m_Delay",        c_uint),#the time in case m_IsTime = True
#        ("m_IsEvOccure",c_uint),#fill by SPV when the event occure.
#        ("m_IsAsPrev",    c_uint),#Python fill this, if it is as prev we don't create CB, just use the prev CB.
#        ("m_LastAProc", c_ulong)#the pointer of the last process control, if it is new Event we kill the old one, and create new one.
#        ]

#when Python (SPV) create thread it create one instance of     OneCEventInfo
#it old one CEvent 
class    OneCEventInfo():    
   def __init__(self,SpvTimeEvId: int,SpvEvId: int,ThreadBornIn : int,name: str):
       #self.m_CEvent     = CEvent;
       self.m_TSpvEvId = SPV.InitSigEv(999,999);#Create time C event -- 999 will mark that it is time event 
       self.m_SpvEvId  = SpvEvId;
       self.m_BornInd     = ThreadBornIn;#the SPV index of thread the first is 0.
       self.m_ThreadEv    = threading.Event();#Event object to use for call wait clear set .....
       self.m_Name        = name;        
   
class    SPSchedulerCenter():
   def __init__(self):
       self.numThread = 0;#this line and all lines with this variable should be deleted, don't need them.
#self.Events = [];
       #self.OneCEventList     = [];#all  event that are  in live (Wait event  to occure, or occure now)
       #self.EvMng            = EventSHolder();#Manager to all  event 
       self.ThreadParamList= [];#list of new thread to init.
       self.MainEvent        = threading.Event();
       self.IsDebug        = True;
       self.OneCEventInfoLst=[];
       self.CurrCEvInfo    = None;
       #self.TimeCEvInfo    = None;
       self.ThreadCounter  = 0;
       self.m_ToStop        = False;
       self.IsThreadKilled = False; 

    
       
   def  WaitEv(self,SPSigEvent):        
       prevSpvId = self.CurrCEvInfo.m_SpvEvId;
       self.CurrCEvInfo.m_SpvEvId = SPSigEvent.m_SpvEvId;#update any way
       SPV.AddPyCb(prevSpvId,SPSigEvent.m_SpvEvId);
                       
       if self.MainEvent.is_set() is False:
           self.MainEvent.set();#release main thread
       self.CurrCEvInfo.m_ThreadEv.clear();#because if scheduler release it, it is true I must clear it to be false.
       self.CurrCEvInfo.m_ThreadEv.wait();
       #if self.IsDebug is True:
           #PR(threading.current_thread().name + " Wake up at time " + str(SPV.SimTime()));        
       return;

    def    WaitTime(self,timeDelay : int):        
       #PR("WaitTime : " + str(timeDelay));
       #SetTime2EvCb(timeDelay,self.TimeCEvInfo.m_SpvEvId);
       prevSpvId = self.CurrCEvInfo.m_SpvEvId;
       self.CurrCEvInfo.m_SpvEvId = self.CurrCEvInfo.m_TSpvEvId;#update any way
       SPV.AddTimePyCb(prevSpvId,self.CurrCEvInfo.m_SpvEvId,timeDelay);

        if self.MainEvent.is_set() is False:
           self.MainEvent.set();#release main thread
       self.CurrCEvInfo.m_ThreadEv.clear();#because if scheduler release it, it is true I must clear it to be false.
       self.CurrCEvInfo.m_ThreadEv.wait();
       #if self.IsDebug is True:
           #PR(threading.current_thread().name + " Wake up at time " + str(SPV.SimTime()));        
       return;
   
   def    SetDebug(self,isDebug : bool):
       self.IsDebug = isDebug;

    def    DoStop(self):
       self.m_ToStop = True;    
   #SHMUEL: I want to start thread,  that will call wait  how much time it need.
   #when it finish, I must know about because I want to release the main thread. the solution
   #is to start any thread with this function, which will call  immediatelly to the user function.
   #when user function finish it will return to this function, which release the main thread.
   def ThreadStartPoint(self,threadParam : ThreadParam):
       if self.IsDebug is True:
           PR(threadParam.m_ThreadName + " StartUp with id " + str(threadParam.ThreadId));
       #create thread info which will be alive all the thread live.
       self.CurrCEvInfo = OneCEventInfo(888888888,999999999,self.ThreadCounter,threadParam.m_ThreadName);        
       self.OneCEventInfoLst.append(self.CurrCEvInfo);
       currThreadInd        = self.ThreadCounter;
       self.ThreadCounter    = self.ThreadCounter + 1;
       threadParam.ThreadFunction(threadParam.ThreadId);
       #when it come to this line, this is the time that thread dead.
       #so I  want  to del  the OneCEventInfo from list.
       listLen   = len(self.OneCEventInfoLst);
       
       indexToPop=0;
       for i in range(0,listLen):
           if self.OneCEventInfoLst[i].m_BornInd  is currThreadInd:
               indexToPop = i;
               break;

        if self.IsDebug is True:
           PR("Kill thread " + str(indexToPop) + "  " + self.CurrCEvInfo.m_Name + " at time " + str(SPV.SimTime()));
                       
       SPV.PyLastCbKill(self.OneCEventInfoLst[indexToPop].m_SpvEvId);
       self.OneCEventInfoLst.pop(indexToPop);    
       self.IsThreadKilled = True;
       
       self.MainEvent.set();

    def Run(self):
#run first all the new thread.    
       self.m_ToStop = False;#to enable continue after error or any user stop
       while True:
           #if self.IsDebug is True:
           #    PR("Len of self.ThreadParamList == " + str(len(self.ThreadParamList)));
           for ThP in  self.ThreadParamList:
               #theThread = threading.Thread(name=newThread.m_ThreadName, target=newThread.ThreadFunction);
               theThread = threading.Thread(name=ThP.m_ThreadName, target=self.ThreadStartPoint,args=(ThP,));
               theThread.start();
               self.MainEvent.wait();
               self.MainEvent.clear();#for next time set it false.
           self.ThreadParamList.clear();
           #loop on CB from SPV to run now.
           if (len(self.OneCEventInfoLst) == 0):
               return;

            listCEVLen = len(self.OneCEventInfoLst);
           currCEVInd = 0;
           #for currCEV in range(0,listCEVLen):
           #for ThCEvInfo in self.OneCEventInfoLst:
           advance = 1;
           while currCEVInd < listCEVLen:
               #if self.IsDebug is True:
               #    PR(" listCEVLen == " + str(listCEVLen) +  " at time " + str(SPV.SimTime()));    
               #if (self.OneCEventInfoLst[currCEVInd].m_CEvent.m_IsEvOccure == 1):
               if (SPV.IsEvOccure(self.OneCEventInfoLst[currCEVInd].m_SpvEvId)):
                   self.CurrCEvInfo = self.OneCEventInfoLst[currCEVInd];#ThCEvInfo;
                   #if self.IsDebug is True:
                   #    PR(" m_IsEvOccure == True for thread " + self.CurrCEvInfo.m_Name +  " at time " + str(SPV.SimTime()));
                   self.OneCEventInfoLst[currCEVInd].m_ThreadEv.set();
                   #self.CurrCEvInfo.m_ThreadEv.set();#ThCEvInfo.m_ThreadEv.set();#release running thread.
                   self.MainEvent.wait();
                   self.MainEvent.clear();#for next time set it false.
                   if self.IsThreadKilled is  True:
                       listCEVLen = listCEVLen -1;                        
                       self.IsThreadKilled  = False;
                       advance = 0;#because one CEvent deleted from list, so i not advance the counter because the next become the current                        
                       PR("  currCEVInd " +  str(currCEVInd));    
               currCEVInd = currCEVInd + advance;
               advance    = 1;
           #if all thread finish, return to prompt.            
           if (len(self.OneCEventInfoLst) == 0):
               return;
           if  self.m_ToStop is True:
               return;
           SPV.PSRun();        

    def    AddThread(self,functionPointer,name: str,id : int):
       self.ThreadParamList.append(ThreadParam(functionPointer,name,id));

        
GlobSched = SPSchedulerCenter();#global scheduler.