Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use write_pid method in autotest

Best Python code snippet using autotest_python

GHA.py

Source:GHA.py

1#!/usr/bin/env python32###############################################################################3#                                                                             #4#                                    GHA.py                                   #5#                                   by Niols                                  #6#                                                                             #7#  BEERWARE License:                                                          #8#  <niols@niols.net> wrote this file. As long as you retain this notice you   #9#  can do whatever you want with this stuff. If we meet some day, and you     #10#  think this stuff is worth it, you can buy me a beer in return.             #11#                                                       ââ Poul-Henning Kamp  #12#                                                                             #13###############################################################################14# Regular python imports15import argparse16import logging17import json18from sys import argv19from os import getpid20from multiprocessing import Process, Queue21from traceback import format_exc22# GHA specific imports23from parsing.gitlab import parse as gitlab_parse24from parsing.github import parse as github_parse25from entrypoints.web import HooksHandlerThread26from outputs.irc import FrontBot, FrontBotThread27class GHA(Process):28    def __init__(self, config):29        super().__init__()30        self.config = config31        self.hooks_queue = Queue()32        self.text_queue = Queue()33    def start_webserver(self):34        hht = HooksHandlerThread(35            self.hooks_queue,36            self.config.listen_host,37            self.config.listen_port38        )39        hht.start()40        logging.info("HooksHandlerThread's pid: {:d}".format(hht.pid))41        if self.config.write_pid:42            logging.debug(43                "Writing HooksHandlerThread's pid in `{}`."44                .format(self.config.write_pid)45            )46            with open(self.config.write_pid, 'a') as file:47                file.write("{:d}\n".format(hht.pid))48    def start_ircbot(self):49        bot = FrontBot(50            self.text_queue,51            self.config.irc_host,52            self.config.irc_port,53            self.config.irc_chans,54            self.config.irc_name55        )56        bot_thread = FrontBotThread(bot)57        bot_thread.start()58        logging.info("FrontBotThread's pid: {:d}".format(bot_thread.pid))59        if self.config.write_pid:60            logging.debug(61                "Writing FrontBotThread's pid in `{}`."62                .format(self.config.write_pid)63            )64            with open(self.config.write_pid, 'a') as file:65                file.write("{:d}\n".format(bot_thread))66    def run(self):67        self.start_webserver()68        self.start_ircbot()69        while True:70            (headers, body) = self.hooks_queue.get()71            hook = json.loads(body)72            try:73                git_obj = None74                if 'X-GitHub-Event' in headers.keys():75                    git_obj = github_parse(headers, hook)76                else:77                    git_obj = gitlab_parse(hook)78                if git_obj is not None:79                    self.text_queue.put((80                        "prnt",81                        {"message": git_obj.render_irccolors()}82                    ))83            except:84                if self.config.report_errors:85                    for line in format_exc().split('\n'):86                        if line:87                            self.text_queue.put((88                                "prnt",89                                {90                                    "message": line,91                                    "chans": [self.config.report_errors]92                                }93                            ))94if __name__ == "__main__":95    # ---96    # Parsing command line arguments97    # ---98    DESCRIPTION = """Github Announcer"""99    parser = argparse.ArgumentParser(description=DESCRIPTION, prog=argv[0])100    parser.add_argument('-lh', '--listen-host',101                        type=str,102                        help='the address where GHA will be listening')103    parser.add_argument('-lp', '--listen-port',104                        type=int,105                        help='the port where GHA will be listening')106    parser.add_argument('-ih', '--irc-host',107                        type=str,108                        help="the irc server\'s address")109    parser.add_argument('-ip', '--irc-port',110                        type=int,111                        help="the irc server's port")112    parser.add_argument('-ic', '--irc-chans',113                        nargs='*',114                        help='the irc channels')115    parser.add_argument('-in', '--irc-name',116                        type=str,117                        help="the bot's name")118    parser.add_argument('-ea', '--export-arguments',119                        metavar='FILE',120                        type=str,121                        help='export arguments in the given file')122    parser.add_argument('-ia', '--import-arguments',123                        metavar='FILE',124                        type=str,  # Ã  rendre plus prÃ©cis125                        help='import arguments from the given file')126    parser.add_argument('--write-pid',127                        metavar='FILE',128                        type=str,129                        help='write all threads pids in given file')130    parser.add_argument('-re', '--report-errors',131                        metavar='NICK',132                        type=str,133                        help='Report errors to the given person')134    parser.add_argument("-v", "--verbosity",135                        type=int,136                        default=1,137                        help="verbosity level")138    config = parser.parse_args()139    # ---140    # Setting up the logger configuration141    # ---142    log_level = logging.WARN143    if 0 <= config.verbosity <= 3:144        levels = [logging.CRITICAL, logging.WARN, logging.INFO, logging.DEBUG]145        log_levels = levels[config.verbosity]146    else:147        logging.error("Invalid verbosity level: {:d}.\nMaximum verbosity: 3"148                      .format(config.verbosity))149        exit(1)150    logging.basicConfig(format='%(asctime)s | %(levelname)s | %(filename)s '151                               'line %(lineno)s | %(message)s',152                        datefmt='%m/%d/%Y %H:%M:%S', level=log_levels)153    if config.import_arguments:154        try:155            file_config = None156            with open(config.import_arguments, "r") as file:157                file_config = json.load(file)158            # The command line arguments have the priority:159            # If an argument is specified both in the command line and in a160            # config file, we pick the command line's one161            for arg in file_config:162                if hasattr(config, arg):163                    if not getattr(config, arg):164                        setattr(config, arg, file_config[arg])165                else:166                    setattr(config, arg, file_config[arg])167        except FileNotFoundError:168            logging.error('File %s not found', config.import_arguments)169            exit(1)170        except:171            logging.error('Error while importing arguments from file.')172            for line in format_exc().split('\n'):173                if line:174                    logging.error(line)175            exit(1)176    if not config.listen_host:177        logging.info('No listen host given. Using 0.0.0.0.')178        config.listen_host = '0.0.0.0'179    if not config.listen_port:180        logging.info('No listen port given. Using 80.')181        config.listen_port = 80182    if not config.irc_host:183        logging.info('No IRC host given. Using localhost.')184        config.irc_host = 'localhost'185    if not config.irc_port:186        logging.info('No IRC port given. Using 6667.')187        config.irc_port = 6667188    if not config.irc_chans:189        logging.info('No IRC chans given.')190        config.irc_chans = []191    if not config.irc_name:192        logging.info('No IRC name given. Using GHA.')193        config.irc_name = 'GHA'194    logging.info("Main thread's pid: {:d}".format(getpid()))195    if config.write_pid:196        open(config.write_pid, 'w').close()  # Shrink size to 0197        logging.debug(198            "Writing main thread's pid in `{:s}`."199            .format(config.write_pid)200        )201        with open(config.write_pid, 'a') as file:202            file.write("{:d\n".format(getpid()))203    if config.export_arguments:204        args = {205            arg: value206            for arg, value in vars(config).items()207            if arg not in ['import_arguments', 'export_arguments']208        }209        with open(config.export_arguments, 'w+') as file:210            json.dump(args, file, indent=4)211        exit(0)212    gha = GHA(config)213    gha.start()...

transfer_heater_PID.py

Source:transfer_heater_PID.py

...34        self.templog += [self.reading]35    def write(self):36        if self.setpoint != 0:37            arduino_set(arduino_name=self.arduino_address, t=self.setpoint)38    def write_pid(self):39        if self.pid_value != '':40            arduino_set_pid(arduino_name=self.arduino_address, pid=self.pid_value)41    def run(self):42        while self.update_flag:43            self.write()44            self.write_pid()45            self.read()46      # #      print(time.asctime(time.localtime(time.time())), "\nTemp =", self.reading, "C", ", Set point =",47      # #            self.setpoint_read, "C")48      #       print(f"Setpoint = {self.setpoint_read}C, Output = {self.Output_read}%, PID value = {self.pid_read}")49      #       plt.title(f"Setpoint = {self.setpoint_read}C, Output = {self.Output_read}%, PID value = {self.pid_read}")50      #       plt.plot(self.timenow - t_now, self.reading, '.r')51      #       plt.xlim([0, self.timenow - t_now + 1])52      #       plt.pause(0.5)53      #       time.sleep(self.time_interval)54      #       np.savetxt(55      #           self.dir +'\\'+ 'temp.txt',56      #           np.column_stack(([a - self.timelog[0] for a in self.timelog], self.templog)), delimiter='\t',57      #           header=f"\Setpoint = {self.setpoint_read}C, Output = {self.Output_read}%, PID value = {self.pid_read}"+ '\n'58      #                  +datetime.now().strftime('%Y%m%d') + '\n' + "time\t\t\ttemp\t\t\t")59      #       print("data saved")60transfer_pid = MyPID()61def transfer_pid_get(arduino_address='COM10'):62    transfer_pid.arduino_address = arduino_address63    transfer_pid.read()64    return transfer_pid.reading65def transfer_pid_set(arduino_address='COM10',kp=1,ki=0,kd=0,set_point=30):66    transfer_pid.arduino_address = arduino_address67    transfer_pid.setpoint = set_point68    transfer_pid.write()69    transfer_pid.pid_value = 'k'+ str(kp)+'p'70    transfer_pid.write_pid()71    transfer_pid.pid_value = 'k' + str(ki) + 'i'72    transfer_pid.write_pid()73    transfer_pid.pid_value = 'k' + str(kd) + 'd'...

daemonize.py

Source:daemonize.py

1# -*- coding: utf-8  -*-2"""Module to daemonize the current process on Unix."""3#4# (C) Pywikibot team, 2007-20155#6# Distributed under the terms of the MIT license.7#8__version__ = '$Id: 918ee974372c7f9744f380a3c0d91bd73ae9e98e $'9#10import os11import sys12import codecs13is_daemon = False14def daemonize(close_fd=True, chdir=True, write_pid=False, redirect_std=None):15    """16    Daemonize the current process.17    Only works on POSIX compatible operating systems.18    The process will fork to the background and return control to terminal.19    @param close_fd: Close the standard streams and replace them by /dev/null20    @type close_fd: bool21    @param chdir: Change the current working directory to /22    @type chdir: bool23    @param write_pid: Write the pid to sys.argv[0] + '.pid'24    @type write_pid: bool25    @param redirect_std: Filename to redirect stdout and stdin to26    @type redirect_std: str27    """28    # Fork away29    if not os.fork():30        # Become session leader31        os.setsid()32        # Fork again to prevent the process from acquiring a33        # controlling terminal34        pid = os.fork()35        if not pid:36            global is_daemon37            is_daemon = True38            if close_fd:39                os.close(0)40                os.close(1)41                os.close(2)42                os.open('/dev/null', os.O_RDWR)43                if redirect_std:44                    os.open(redirect_std,45                            os.O_WRONLY | os.O_APPEND | os.O_CREAT)46                else:47                    os.dup2(0, 1)48                os.dup2(1, 2)49            if chdir:50                os.chdir('/')51            return52        else:53            # Write out the pid54            path = os.path.basename(sys.argv[0]) + '.pid'55            with codecs.open(path, 'w', 'utf-8') as f:56                f.write(str(pid))57            os._exit(0)58    else:59        # Exit to return control to the terminal60        # os._exit to prevent the cleanup to run...

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.