KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use stoplog method in pyatom
Best Python code snippet using pyatom_python
process.py
Source:process.py 
1#2# Copyright 2018 justworx3# This file is part of the trix project, distributed under4# the terms of the GNU Affero General Public License.5#6from .. import * # trix, sys7from ..fmt import JCompact8from .runner import Runner9from .xqueue import *10import subprocess11class ProcessError(Exception): pass12class LaunchError(ProcessError): pass13try:14 Timeout = subprocess.Timeout15except:16 Timeout = Exception17PROC_IO_TIMEOUT = 0.0118#19# CPORT CAN'T BE AN ARGUMENT20# - All Process instances try to get the remote process to create21# a connection, then timeout if they don't cooperate.22# - However... if a "ctime" kwarg is provided, it overrides the 23# default timeout (5 minutes), and if it's set to zero, no attempt24# to receive a connection.25#26class Process(Runner):27 """28 1 - Create a process given object creation args.29 2 - Launch the process (giving `run` args as needed)30 3 - Start this Process object...31 - loop waiting for input/error; errors that fall through (from the32 remote process) will come via stderr and are fatal. Errors from33 the control handler may be handled as appropriate.34 4 - Read and handle data received via qread (for stdin) and/or cread35 (for data received via the process server).36 """37 38 CLINE = "%s -m %s launch %s"39 CTIME = 300 # timeout for socket connect-back: default 5m40 CTOUT = 3.0 # timeout for cport communication response 3s41 DEBUG = {}42 HAND = trix.innerpath('net.handler.hlines.HandleLines')43 WRAP = trix.innerpath('util.sock.sockwrap.sockwrap')44 ENCODE = DEF_ENCODE45 46 #47 # INIT48 #49 def __init__(self, cpath, *a, **k):50 """51 Pass 'package.module.Class' as `cpath`, plus any args/kwargs for 52 creation of the object to run in a remote process.53 54 Encoding-related kwargs ('encoding' and 'errors') are shared with55 both the remote process and this controlling Process object, which56 must communicate by bytes encoded with the specified encoding (or 57 DEF_ENCODE).58 59 Note that the remote process Runner will also react to the same 60 kwargs as this Process object (in this local process), so their 61 "sleep" time will be in synch.62 """63 64 rk = trix.kpop(k, 'encoding errors sleep')65 66 # DEBUG INFO67 ID = "%s.%s:%i" % (__name__, type(self).__name__, trix.pid())68 #trix.log("%s.%s:%i" % (__name__, type(self).__name__, trix.pid()))69 processconfig = dict(ID=ID, cpath=cpath, a=a, k=k)70 71 #72 # DEFAULT ENCODING73 # - An encoding is necessary, and defaults to DEF_ENCODE. This74 # encoding's used to convert bytes sent from the remote process75 # via pipes (using the EncodingHelper's `encode` and `decode`76 # methods).77 # - If your system is producing bytes in an encoding other than78 # DEF_ENCODE, set Process.ENCODE to that encoding.79 # - If encoding-related errors arise, it may be necessary to80 # add some (more) complication to this class or to __main__. 81 # For now, keep fingers crossed.82 #83 k = dict(encoding=self.ENCODE)84 85 #86 # RUNNER KWARGS87 # - The 'encoding', 'errorse, and 'sleep' kwargs, extracted to88 # `rk`, above, are shared in common with the remote process.89 #90 Runner.__init__(self, processconfig=processconfig, **rk)91 92 #93 # LAUNCH PARAMS94 # - These (cpath,a,k) must be sent to the launch method95 #96 self.__cpath = cpath # object path (package.module.class/function)97 self.__args = a # object constructor args98 self.__krgs = k # object constructor kwargs99 100 #101 # PIPES102 # - Pipes are used after launch to receive any launch errors;103 # - The stdin pipe is never needed by the trix package, but is104 # available in case needed for the launching of some externally105 # defined object.106 #107 self.__p = None108 self.__pk = {} # kwargs for Popen constructor109 #self.__pk.setdefault('stdin', subprocess.PIPE)110 self.__pk.setdefault('stdout', subprocess.PIPE) 111 self.__pk.setdefault('stderr', subprocess.PIPE) 112 113 #114 # STDOUT QUEUE115 # - Data recieved from stdin is queued. Use self.reado to pop116 # one item (or `None` if no item exists).117 # - Anything received from stderr is converted to a ProcessError118 # or LaunchError exception and raised immediately.119 #120 self.__stdrecv = Queue()121 122 #123 # CONTROL SERVER124 # - Pass a cport kwarg if you want communication between this125 # object and the remote process.126 # - Launch a server and wait (until timeout) for the remote 127 # process to connect.128 # 129 self.__cserv = None130 self.__csock = None131 self.__chand = None132 self.__cport = None133 self.__ctime = k.get('ctime', self.CTIME)134 self.__qsend = Queue()135 136 # This should only be set after `launch()` is called.137 #self.__ctimeout = None138 139 # keep an eye on self.stop() for a while140 self.__stoplog = []141 self.__exitcode = None142 143 #144 # DEBUGGING145 # - store process-id and constructor params146 #147 #self.dbg = dict(pid=trix.pid(), dak=dict(cpath=cpath, a=a, k=k))148 #trix.log("process.dbg", self.dbg)149 150 151 152 #153 # DEL154 #155 def __del__(self):156 if self.active:157 self.stop()158 159 160 # runtime values161 162 @property163 def args(self):164 """Additional Process constructor args."""165 return self.__args166 167 @property168 def kwargs(self):169 """Process constructor kwargs."""170 return self.__krgs171 172 @property173 def cpath(self):174 """module.class path (1st Process constructor arg)."""175 return self.__cpath176 177 @property178 def cport(self):179 """Remote process connection port."""180 return self.__cport181 182 @property183 def csock(self):184 """Remote process connection socket."""185 return self.__csock186 187 @property188 def chand(self):189 """Remote process connection handler."""190 return self.__chand191 192 @property193 def ctime(self):194 """Time limit for remote process to connect. Default: 5 min."""195 return self.__ctime196 197 198 199 # runtime values200 201 @property202 def active(self):203 """True if the process exists and has not exited."""204 try:205 return self.__p and (self.poll() == None)206 except:207 return False208 209 @property210 def pid(self):211 """Return the remote process id."""212 return self.__p.pid if self.__p else None213 214 @property215 def p(self):216 """Direct access to the `subprocess.Popen` object."""217 return self.__p218 219 220 221 222 223 def poll(self):224 """225 Poll for an exit code; Remains None until the process terminates.226 Retains the value even after the process is terminated - even 227 after self.__p is deleted.228 """229 try:230 try:231 return self.__poll232 except:233 poll = self.__p.poll()234 if poll != None:235 self.__poll = poll236 return self.__poll237 else:238 return None239 except:240 return None241 242 243 244 245 246 #247 # STATUS/DISPLAY248 #249 def status(self):250 """Return the status of this object."""251 runner = Runner.status(self)252 process = dict(253 active = self.active,254 cpath = self.__cpath,255 cport = self.__cport,256 chand = self.__chand,257 poll = self.poll(),258 pid = self.pid259 )260 261 # add the exit code, if one exists262 if self.__stoplog != None:263 process['stoplog'] = self.__stoplog264 265 return dict(266 runner = runner,267 process = process268 )269 270 271 # DISPLAY272 def display(self, data=None, **k):273 """Print formatted display of this object's status."""274 trix.display(data or self.status())275 276 277 # REMOTE STATUS278 def rstatus(self):279 """Return the remote process status as a dict."""280 self.write('status\r\n')281 tstart = time.time()282 tmout = tstart+self.CTOUT283 while time.time() < tmout:284 status = self.read().strip()285 if status:286 return trix.jparse(status)287 288 289 # REMOTE DISPLAY290 def rdisplay(self):291 """Display the remote process status in JSON format."""292 trix.display(self.rstatus())293 294 295 296 #297 # IO - main loop handler (see `runner.Runner`)298 #299 def io(self):300 """Manage process activity, including i/o from pipes and cserv."""301 302 # if the process dies, stop running303 if (not self.__p) or self.poll():304 self.stop()305 306 else:307 # 1 - HANDLE STDERR/STDOUT FROM REMOTE PROCESS308 self.__stdeo()309 310 # 2 - DEAL WITH CONNECTION SERVER OPERATION311 if self.__chand:312 #313 # UNDER CONSTRUCTION314 # - Nothing to do at the moment - io is performed using315 # the read() and write() methods.316 # - Future releases may add maintenaince features (eg, a317 # PING/PONG feature) here.318 # - Either way, this `if` is still needed - if only to 319 # prevent checking the conditions below.320 #321 pass322 323 elif self.__cserv:324 325 # Check for timeout.326 if (time.time() < self.__ctimeout):327 328 # Accept, if incoming connection is attempted.329 h = self.__cserv.acceptif()330 331 # got a connection; wrap sock and set `chand`332 if h:333 self.__csock = trix.create(self.WRAP, h[0])334 self.__chand = trix.create(self.HAND, h[0])335 self.__cserv.shutdown()336 self.__cserv = None337 338 # read pid339 time.sleep(0.01)340 rdata = self.readline()341 if rdata:342 self.receivedPid = rdata.strip()343 if not int(self.receivedPid) == self.pid:344 raise Exception("err-process-fail", xdata(345 reason="err-pid-mismatch"346 ))347 348 #349 # 1 - Set self.__write to the real `_write` method.350 # 2 - Send any data that was written (and queued) before 351 # csock and chand came online...352 #353 self.__write = self._write354 try:355 dk = self.__qsend.get_nowait()356 while dk:357 self.writeline(dk[0], **dk[1])358 dk = self.__qsend.get_nowait()359 except Empty:360 pass361 362 else:363 #364 # The timeout has been reached; stop waiting for connection365 # from the remote process.366 #367 self.__cserv.shutdown()368 self.__cserv = None369 370 371 372 373 374 #375 #376 # STOP 377 #378 #379 def stop(self):380 """Shutdown, terminate, or kill the process."""381 382 ##383 ## Send 'shutdown' command and wait for remote process to exit.384 ##385 self.__stoplog.extend(['stoplog', time.time()])#386 if self.active:387 388 self.__chand.write('shutdown\r\n')389 self.__stoplog.append('shutdown sent')#390 391 # wait for process to exit392 if self.__p:393 tstart = time.time()394 tmout = tstart+1.1395 while time.time() < tmout:396 self.__exitcode = self.poll()397 if self.__exitcode == None:398 time.sleep(0.05)399 else:400 self.__exitcode = poll = self.poll()401 time.sleep(0.05)402 self.__stoplog.extend([403 'shutdown success', time.time(), "poll=%s"%str(poll),404 'active="%s"' % str(self.active)405 ])#406 break407 408 #409 # CHECK PROCESS ENDED410 # - If remote process failed to exit naturally (by 'shutdown'411 # command), try to terminate it, or kill it if necessary.412 #413 if self.active:414 self.__stoplog.extend(['still active; terminate',415 "poll=%s"%str(self.poll())416 ])#417 try:418 # TERMINATE419 self.__p.terminate()420 self.__stoplog.append("process-terminated")#421 except Exception as ex:422 self.__stoplog.append("terminate failed", type(ex), ex.args)423 try:424 # KILL425 self.__p.kill()426 self.__stoplog.append("process killed")#427 except Exception as ex:428 # TERMINATE/KILL FAIL429 self.__stoplog.append("kill failed", type(ex), ex.args)#430 431 # Clean up; 432 try:433 # record any exit error434 self.__exitcode = self.poll()435 436 # make sure p is garbage-collected437 self.__p = None438 439 except Exception as ex:440 self.__stoplog.append("cleanup error", type(ex), ex.args)441 442 finally:443 # stop the runner444 Runner.stop(self)445 446 447 448 449 450 #451 # CONTROL-SOCKET I/0452 # - copy the socket reading/writing features453 #454 def read(self, **k):455 """Read data from control socket."""456 if self.__csock and self.__chand:457 data = self.__csock.read()458 if data:459 self.__chand.handledata(data)460 return self.__chand.read()461 462 463 def readline(self):464 """Read a line from control socket."""465 if self.__csock and self.__chand:466 data = self.__csock.read()467 if data:468 self.__chand.handledata(data)469 return self.__chand.readline()470 471 472 def readlines(self):473 """Read all complete lines from control socket."""474 if self.__csock and self.__chand:475 data = self.__csock.read()476 if data:477 self.__chand.handledata(data)478 return self.__chand.readlines()479 480 481 482 483 484 @property485 def write(self):486 """487 Returns the `write` method suitable to current circumstances.488 Call this as though it were the actual `write` method.489 >>> p.write(someData)490 """491 try:492 return self.__write493 except AttributeError:494 return self.__writeq495 496 497 def writeline(self, data, **k):498 """Append CRLF to data and write."""499 self.write(data+"\r\n", **k)500 501 502 def _write(self, data, **k):503 """Write data to socket."""504 if self.__csock and self.__chand:505 self.__csock.write(data, **k)506 507 508 def __writeq(self, data, **k):509 """Queue data/k for writing when csock and chand are set."""510 self.__qsend.put([data, k])511 512 513 514 515 516 # direct send/recv methods517 518 def recv(self, sz):519 """Standard, old-fashioned `recv` direct from socket."""520 return self.__csock.recv(sz)521 522 def send(self, data):523 """Standard, old-fashioned `send` direct to socket."""524 return self.__csock.send(data)525 526 527 # query528 def query(self, cmd):529 """530 Query the remote process; Return result as a dict.531 """532 self.writeline(cmd)533 tstart = time.time()534 tmout = tstart+self.CTOUT535 while time.time() < tmout:536 reply = self.read().strip()537 if reply:538 return trix.jparse(reply)539 540 541 542 #543 # READ-O (Standard Output Queue)544 #545 def reado(self):546 """Read queued standard input from remote process."""547 #548 # This has actually never happened. I'm not sure it even *can* 549 # happen.550 #551 # It may be best to remove this method.552 #553 try:554 return self.__stdrecv.get_nowait()555 except Empty:556 return None557 558 559 560 561 562 #563 # LAUNCH564 #565 def launch(self, run=None, *a, **k):566 """567 The `run` argument may be specified if the object created (as 568 specified by the constructor args/kwargs) contains a function or569 method that must be called after the process starts. In such a 570 case, any required args/kwargs may be passed, too. If the object571 is a class with a constructor that starts the process, then `run`572 should be left as None and no args/kwargs may be specified.573 574 Returns `self`.575 """576 577 if self.__p:578 raise Exception('err-launch-fail', xdata(579 detail="process-already-launched", p=self.__p,580 DEBUG="Launch may be called only once per Process"581 ))582 583 #584 # LAUNCH PROCESS585 # - PROCESS : create control server, add cport to kwargs586 # - PROCESS : pack args and popen()587 # - __MAIN__ : unpack object args and load module588 # - __MAIN__ : unpack method args589 # - __MAIN__ : call method (eg, "run")590 # - PROCESS : self.start(), and return self591 #592 593 #594 # CONTROL SERVER595 #596 if self.__ctime:597 # always use the basic socserv class here598 self.__cserv = trix.ncreate('util.sock.sockserv.sockserv', 0)599 self.__cport = self.__cserv.port600 601 602 # timeout interval starts when the process launches603 self.__ctimeout = self.__ctime + time.time()604 605 606 #607 # PACK ARGS608 #609 610 # set connect-back cport611 if self.__cport:612 self.__krgs['CPORT'] = self.__cport613 614 # create a compact (json -> zlib -> base64) argument spec615 cargs = [self.__cpath, self.__args, self.__krgs, run, a, k]616 617 # compress (zlib) the json and b64 it618 clarg = JCompact().compact(cargs)619 620 # launch args621 pyexe = sys.executable622 trixp = trix.innerpath()623 fsenc = sys.getfilesystemencoding()624 cline = self.CLINE % (pyexe, trixp, clarg.decode(fsenc))625 626 627 # DEBUG628 #dbg = dict(pyexe=pyexe, trixp=trixp, fsenc=fsenc, cline=cline )629 #self.dbg['launch']=dbg630 631 632 #633 # LAUNCH REMOTE PROCESS634 # - Popen through trix.__main__.py635 # - The 'launch' command (in __main__) uses trix.popen(), so 636 # stderr/stdout are set to PIPE.637 # - Only the command line argument and kwargs are sent to popen -638 # the `run`, `a`, and `k` values are used (in __main__) to get639 # the external object going (eg, by calling its `run` method.)640 #641 self.__p = trix.popen(cline.split(), **self.__pk)642 643 # Give the remote process a chance to get going.644 time.sleep(0.05)645 646 # QUICK CHECK FOR LAUNCH ERRORS647 if (not self.__p) or self.poll():648 self.__stdeo(LaunchError)649 650 #651 # START PROCESS652 # - start a thread that loops through this object's `io` method 653 #654 if not self.running:655 self.start()656 657 # return `self` so that `p=trix.process(x).launch(y)` will work.658 return self659 660 661 662 663 664 def __stdeo(self, TException=None):665 #666 # STDOUT/STDERR667 # - queue any stdout/stderr data; raise any errors.668 #669 e=None670 o=None671 try:672 o,e = self.p.communicate(timeout=PROC_IO_TIMEOUT)673 except Timeout:674 pass675 676 # check for errors677 if e:678 # there's been an error679 TException = TException or ProcessError680 681 o = self.decode(o)682 e = self.decode(e).strip()683 684 # defaults685 XDATA = dict(stderr=e, stdout=o)686 687 if o and e:688 #689 # O AND E690 # - If both out and err exist, an exception was raised; out691 # is the exception message and err is the traceback.692 # - split out into lines and parse all but the first (which693 # is the <Exception type>).694 #695 lines = o.strip().splitlines()696 xtype = lines[0]697 xtext = "\n".join(lines[1:-1]) # last line is "Traceback:"698 699 700 # JPARSE STDOUT (o)701 try:702 _err = trix.jparse(xtext)703 XDATA['stdout'] = _err704 except Exception as ex:705 #706 # J-PARSE EXCEPTION707 #708 XDATA['xinfo'] = dict(709 message = 'stdout-parse-fail',710 xtype = type(ex),711 xargs = ex.args,712 )713 714 # TRACEBACK IS STDERR (e)715 XDATA['stderr'] = e.splitlines()716 # raise the stderr error717 raise TException('process-error', dict(718 xtype = xtype,719 xdata = XDATA['stdout'],720 tracebk = XDATA['stderr']721 ))722 723 else:724 #725 # STDERR ONLY726 # - otherwise, err is probably an error message string.727 #728 729 # parse the error730 try:731 _err = trix.jparse(e)732 XDATA['stderr'] = _err733 except Exception as ex:734 #735 # J-PARSE EXCEPTION736 # - The `e` value couldn't be parsed, so let it remain as a737 # string. Include the parse-error type and args for738 # debugging parse errors.739 #740 XDATA['xinfo'] = dict(741 message = 'stderr-parse-fail',742 xtype = type(ex),743 xargs = ex.args744 )745 746 # raise the stderr error747 raise TException('process-error', XDATA)748 749 elif o:750 #751 # STDOUT ONLY752 # - Some data was written. Store it in the stdrecv queue.753 # - Read it by calling the `reado()` method.754 #755 self.__stdrecv.put(self.decode(o))756 #print ("\n#\n# RECV'D: %s\n#\n", self.decode(o))#debug...
obd_logger.py
Source:obd_logger.py
1#!/usr/bin/python32import obd3import time4import getpass5#import subprocess6import datetime7import sys8from threading import Thread9import os10#sys.path.insert(0, '../')11#from config import cmds12# ELM327 v1.513#subprocess.Popen("sudo rfcomm connect 0 00:0D:18:3A:67:89 1", shell=True)14## Logger management15#HOME_FOLDER = '/home/chip'#+getpass.getuser()16LOG_FOLDER = "/var/www/owncloud/data/owncloud/files/logs/obd_logs/"#HOME_FOLDER+"/Documents/logs/obd_logs/"17LOGDATA_FOLDER = "/var/www/owncloud/data/owncloud/files/logs/obddata_logs/"#HOME_FOLDER+"/Documents/logs/obddata_logs/"18LOG_FILENAME = "_obd.log"19LOGDATA_FILENAME = "_obdData.log"20LOG_TOKEN = "[O]"#"[USB_SHARE]"21LOG_SEP = ";"22PORT = '/dev/rfcomm0'23RECONNECTION_DELAY_SEC = 1024RECONNECTION_MAX_TRIALS = 2025LOGDATA_FILE = None26LOG_FILE = None27VERBOSE = False28TS_S = 0.1 # sampling time29TIMEOUT_FOR_STOPLOG = 60*5# s30n_reconnection_trials = 131cmds = [32 obd.commands.THROTTLE_POS,#33 obd.commands.SPEED, #34 obd.commands.RPM, #35 obd.commands.ENGINE_LOAD,# %36 obd.commands.FUEL_LEVEL,#37 obd.commands.COOLANT_TEMP,#38 #obd.commands.AMBIANT_AIR_TEMP,39 #obd.commands.INTAKE_PRESSURE,40 #obd.commands.INTAKE_TEMP,41 obd.commands.MAF,# gps42 #obd.commands.FUEL_RAIL_PRESSURE_DIRECT,43 #obd.commands.BAROMETRIC_PRESSURE,44 #obd.commands.CONTROL_MODULE_VOLTAGE,45 obd.commands.ACCELERATOR_POS_D,# % throttle46 #obd.commands.ACCELERATOR_POS_E,# % throttle47 #obd.commands.THROTTLE_ACTUATOR, # %48 obd.commands.OIL_TEMP,49 #'v': obd.commands.RELATIVE_THROTTLE_POS,#50 #obd.commands.ACCELERATOR_POS_F,51 #obd.commands.FUEL_RATE,52 #obd.commands.ABSOLUTE_LOAD,53 #obd.commands.COMMANDED_EQUIV_RATIO,54 #obd.commands.THROTTLE_POS_B,55 #obd.commands.THROTTLE_POS_C,56 #obd.commands.COMMANDED_EGR,57 #obd.commands.FUEL_RAIL_PRESSURE_VAC,58 #obd.commands.FUEL_PRESSURE,59 #obd.commands.TIMING_ADVANCE,60]61 62 63def write_to_log(msg):64 global LOG_FILENAME65 now = datetime.datetime.now()66 hms = str(now.hour)+':'+str(now.minute)+':'+str(now.second)67 log_file = open(LOG_FOLDER+LOG_FILENAME, 'a')68 log_file.write(hms + " " + LOG_TOKEN + " " + msg + '\n')69 log_file.close() 70def write_to_logData(msg, logdata_file, printTime = True):71 #now = datetime.datetime.now()72 # hms = str(now.hour)+':'+str(now.minute)+':'+str(now.second)73 millis = str(time.time())74 if printTime:75 logdata_file.write(millis + msg + '\n')76 else:77 logdata_file.write(msg + '\n')78 79def init_LOG_FILEs():80 global LOG_FILENAME81 global LOGDATA_FILE82 now = datetime.datetime.now()83 timenow = str(now)84 timenow = timenow.replace(' ', 'h')85 timenow = timenow.replace(':', '_')86 #print(timenow)87 LOGDATA_FILE = open(LOGDATA_FOLDER+timenow+LOGDATA_FILENAME, 'a')88 LOG_FILENAME = timenow+LOG_FILENAME89 90 91def OBDconnect(port, cmds):92 #OBDconnection = obd.OBD(port)93 OBDconnection = obd.Async(port)#, delay_cmds=0.05)94 for cmd in cmds:95 OBDconnection.watch(cmd) # keep track of the RPM96 OBDconnection.start()97 return OBDconnection98 99 100def OBDreconnect(port, cmds):101 OBDconnection = OBDconnect(port, cmds) 102 103 n_reconnection_trials = 0104 105 while not OBDconnection.is_connected():106 107 n_reconnection_trials += 1108 msg = str(n_reconnection_trials)+" Not connected, reconn. in "+str(RECONNECTION_DELAY_SEC)+" sec"109 write_to_log(msg)110 time.sleep(RECONNECTION_DELAY_SEC)111 112 try:113 OBDconnection = OBDconnect(port, cmds) 114 except:115 OBDconnection.stop()116 write_to_log("Unexpected error: "+str(sys.exc_info()[0]) )117 118 if n_reconnection_trials > RECONNECTION_MAX_TRIALS:119 OBDconnection.stop()120 write_to_log("Impossible to connect, quit application")121 quit()122 return OBDconnection123 124def threadSamplingTime(ts_s):125 time.sleep(ts_s)126#time.sleep(1)127#response = OBDconnection.query(obd.commands.ELM_VERSION); print(response)128write_to_log("Refresh OwnCloud index")129os.system("sudo -u www-data php /var/www/owncloud/occ files:scan --all")130OBDconnection = OBDconnect(PORT, cmds)131t_init_for_stoplog = time.time()132t_since_stop = 0133nDatalines = 0134engine_rpm = 0135init_LOG_FILEs()136header = "Time_s"137for cmd in cmds: 138 header += LOG_SEP + cmd.name139write_to_logData(header, LOGDATA_FILE, printTime = False)140write_to_log("Connected to "+PORT)141while True:142 logged_values = ""143 error_while_logging = False144 logged_all_data = True145 146 ts_thread = Thread(target=threadSamplingTime, args=(TS_S,))147 ts_thread.start()148 149 for cmd in cmds: 150 try:151 response = OBDconnection.query(cmd)152 logged_values += LOG_SEP + str(response.value.magnitude)153 154 if cmd.name == 'RPM':155 engine_rpm = response.value.magnitude156 157 except:158 write_to_log("Error when logging "+cmd.name)159 logged_all_data = False160 engine_rpm = 0161 break162 163 #try:164 # logged_values += LOG_SEP + str(response.value.magnitude)165 #except:166 # error_while_logging = True167 # write_to_log("Error in connection, reconnecting")168 # OBDconnection.stop()169 # OBDconnection = OBDreconnect(PORT, cmds)170 171 172 #print(logged_values)173 174 if logged_all_data: #not error_while_logging:175 if VERBOSE:176 print(logged_values)177 write_to_logData(logged_values, LOGDATA_FILE)178 nDatalines += 1179 if (nDatalines % 1000) == 0 :180 write_to_log("Logged "+str(nDatalines)+" lines")181 else:182 if VERBOSE:183 print("Error in logging")184 logged_values = ""185 186 #time.sleep(0.01)187 ts_thread.join()188 189 190 # if vehicle still, proceed to monitor whether stop the log191 if engine_rpm == 0:192 t_since_stop = time.time()-t_init_for_stoplog193 else:194 t_since_stop = 0195 t_init_for_stoplog = time.time()196 197 # stop the log198 if t_since_stop > TIMEOUT_FOR_STOPLOG:199 write_to_log("Engine RPM zero for "+str(TIMEOUT_FOR_STOPLOG)+" seconds, shutting down!")200 break201 202write_to_log("Closing OBD connection") 203OBDconnection.close()204write_to_log("Closing log data file") 205LOGDATA_FILE.close() 206write_to_log("Refresh OwnCloud index")207os.system("sudo -u www-data php /var/www/owncloud/occ files:scan --all")...
shutdown_button.py
Source:shutdown_button.py
...9from gpiozero import Button10from signal import pause11from subprocess import check_call12held_for=0.013def stoplog():14 os.system("sudo kill $(pidof str2str)")15 time.sleep(5)16 17def rls():18 global held_for19 if (held_for > 5.0):20 stoplog() 21 check_call(['/sbin/poweroff'])22 elif (held_for > 2.0):23 stoplog()24 check_call(['/sbin/reboot'])25 else:26 held_for = 0.027def hld():28 # callback for when button is held29 # is called every hold_time seconds30 global held_for31 # need to use max() as held_time resets to zero on last callback32 held_for = max(held_for, button.held_time + button.hold_time)33button=Button(use_button, hold_time=1.0, hold_repeat=True)34button.when_held = hld35button.when_released = rls...
Automation Testing Tutorials
Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run pyatom automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
