KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use terminate_p method in fMBT

Best Python code snippet using fMBT_python

analysis.py

Source:analysis.py

1'''Analysis datatype class'''2# General Import3import sys4# App Import5from basepair.helpers import eprint6from bin.common_parser import add_common_args, add_single_uid_parser, add_uid_parser, add_json_parser, \7  add_tags_parser, add_outdir_parser, valid_uid, validate_analysis_yaml8class Analysis:9  '''Analysis action methods'''10  @staticmethod11  def create_analysis(bp_api, args):12    '''Create and submit an analysis'''13    params = {'node': {}, 'info': {}}14    if args.instance:15      try:16        instance_choices = bp_api.get_instances()17      except KeyError:18        sys.exit('ERROR: Failed to get instance data.')19      if args.instance not in instance_choices:20        sys.exit('ERROR: invalid instance_type available instances - {}'.format(' '.join(instance_choices)))21      params['info']['instance_type'] = args.instance22    if args.custom_pipeline or args.custom_modules:23      validate_analysis_yaml(args.custom_pipeline if args.custom_pipeline else args.custom_modules)24    if args.params:25      for param in args.params:26        try:27          node_id, arg, val = param.split(':')28          if node_id not in params['node']:29            params['node'][node_id] = {}30          params['node'][node_id][arg] = val31        except ValueError:32          sys.exit('ERROR: Missing : in params values.')33    else:34      eprint('You specified no parameters, submitting with default ones.')35    return bp_api.create_analysis(36        control_ids=args.control or [],37        ignore_validation_warnings=args.ignore_warning,38        params=params,39        project_id=args.project,40        sample_ids=args.sample,41        workflow_id=args.pipeline,42        pipeline_yaml=args.custom_pipeline,43        module_yaml=args.custom_modules44      )45  @staticmethod46  def delete_analysis(bp_api, args):47    '''Delete analysis'''48    uids = args.uid49    all_fail = True50    for uid in uids:51      answer = bp_api.yes_or_no('Are you sure you want to delete {}?'.format(uid))52      if answer:53        all_fail = not bool(bp_api.delete_analysis(uid)) and all_fail54    if all_fail:55      sys.exit('ERROR: Deleting analysis failed.')56  @staticmethod57  def download_analysis(bp_api, args):58    '''Download analysis'''59    # download a file from an analysis by tags60    all_fail = True61    for each_uid in args.uid:62      analysis = bp_api.get_analysis(each_uid)63      all_fail = not (bool(analysis.get('id')) \64        and bool(bp_api.download_analysis(each_uid, analysis=analysis, outdir=args.outdir, tagkind=args.tagkind, tags=args.tags))) \65        and all_fail66    if all_fail:67      sys.exit('ERROR: Downloading analysis failed.')68    eprint('All analysis files have been downloaded successfully.')69  @staticmethod70  def download_log_analysis(bp_api, args):71    '''Download analysis log'''72    all_fail = True73    for uid in args.uid:74      analysis = bp_api.get_analysis(uid)  # check analysis id is valid75      all_fail = not (bool(analysis.get('id')) and bool(bp_api.get_log(uid, args.outdir))) and all_fail76    if all_fail:77      sys.exit('ERROR: Downloading logs failed.')78  @staticmethod79  def get_analysis(bp_api, args):80    '''Get analysis'''81    all_fail = True82    for uid in args.uid:83      all_fail = not bool(bp_api.print_data(data_type='analysis', uid=uid, is_json=args.json)) and all_fail84    if all_fail:85      sys.exit('ERROR: Analyses data not found.')86  @staticmethod87  def list_analysis(bp_api, args):88    '''List analyses'''89    bp_api.print_data(data_type='analyses', is_json=args.json, project=args.project)90  @staticmethod91  def reanalyze_analysis(bp_api, args):92    '''Restart analysis'''93    if args.instance:94      try:95        instance_choices = bp_api.get_instances()96      except KeyError:97        sys.exit('ERROR: Failed to get instance data.')98      if args.instance not in instance_choices:99        sys.exit('ERROR: invalid instance_type available instances - {}'.format(' '.join(instance_choices)))100    all_fail = True101    for each_uid in args.uid:102      all_fail = not bool(bp_api.restart_analysis(each_uid, args.instance)) and all_fail103    if all_fail:104      sys.exit('ERROR: while re-analyze the analysis data.')105  @staticmethod106  def terminate_analysis(bp_api, args):107    '''Terminate analysis'''108    for each_uid in args.uid:109      bp_api.terminate_analysis(each_uid)110  @staticmethod111  def update_analysis(bp_api, args):112    '''Update analysis'''113    data = {}114    if args.key and args.val:115      for key, val in zip(args.key, args.val):116        data[key] = val117    bp_api.update_analysis(args.uid, data)118  @staticmethod119  def analysis_action_parser(action_parser):120    '''Analysis parser'''121    # create analysis parser122    create_analysis_p = action_parser.add_parser(123      'create',124      help='Create and run an analysis.'125    )126    create_analysis_p = add_common_args(create_analysis_p)127    create_analysis_p.add_argument(128      '--control', nargs='+', help='Control id', type=valid_uid129    )130    create_analysis_p.add_argument(131      '--custom-modules', dest='custom_modules', help='Path to custom module yaml files', nargs='+'132    )133    create_analysis_p.add_argument(134      '--custom-pipeline', dest='custom_pipeline', help='Path to custom pipeline yaml file',135    )136    create_analysis_p.add_argument(137      '--ignore-warning',138      action='store_true',139      default=False,140      help='Ignore validation warnings',141    )142    create_analysis_p.add_argument(143      '--instance', help='instance_type for analysis'144    )145    create_analysis_p.add_argument('--params', nargs='+')146    create_analysis_p.add_argument('--project', help='Project id', type=valid_uid)147    create_analysis_p.add_argument(148      '--pipeline', help='Pipeline id', type=valid_uid149    )150    create_analysis_p.add_argument(151      '--sample', nargs='+', help='Sample id', required=True, type=valid_uid152    )153    # delete analysis parser154    delete_analysis_p = action_parser.add_parser(155      'delete',156      help='delete an analysis.'157    )158    delete_analysis_p = add_common_args(delete_analysis_p)159    delete_analysis_p = add_uid_parser(delete_analysis_p, 'analysis')160    # download analysis parser161    download_analysis_p = action_parser.add_parser(162      'download',163      help='Download files for one or more analyses. Can filter by file tags.'164    )165    download_analysis_p = add_uid_parser(download_analysis_p, 'analysis')166    download_analysis_p = add_tags_parser(download_analysis_p)167    download_analysis_p = add_outdir_parser(download_analysis_p)168    download_analysis_p = add_common_args(download_analysis_p)169    # download analysis log parser170    download_log_p = action_parser.add_parser(171      'download-log',172      help='Download analysis logs for one or more analyses.'173    )174    download_log_p = add_uid_parser(download_log_p, 'analysis')175    download_log_p = add_outdir_parser(download_log_p)176    download_log_p = add_common_args(download_log_p)177    # get analysis parser178    get_analysis_p = action_parser.add_parser(179      'get',180      help='Get details of an analysis.'181    )182    get_analysis_p = add_common_args(get_analysis_p)183    get_analysis_p = add_uid_parser(get_analysis_p, 'analysis')184    get_analysis_p = add_json_parser(get_analysis_p)185    # list analysis parser186    list_analyses_p = action_parser.add_parser(187      'list',188      help='List basic info about your analyses.'189    )190    list_analyses_p.add_argument(191      '--project',192      help='List analyses of a project.',193      required=True,194      type=valid_uid195    )196    list_analyses_p = add_common_args(list_analyses_p)197    list_analyses_p = add_json_parser(list_analyses_p)198    # reanalyze parser199    reanalyze_p = action_parser.add_parser(200      'reanalyze',201      help='Reanalyze analyses.'202    )203    reanalyze_p.add_argument(204      '--instance', help='instance_type for analysis'205    )206    reanalyze_p = add_common_args(reanalyze_p)207    reanalyze_p = add_uid_parser(reanalyze_p, 'analysis')208    # terminate parser209    terminate_p = action_parser.add_parser(210      'terminate',211      help='Terminate analyses.'212    )213    terminate_p = add_common_args(terminate_p)214    terminate_p = add_uid_parser(terminate_p, 'analysis')215    # update an analysis parser216    update_analysis_parser = action_parser.add_parser(217      'update',218      help='Update information associated with an analysis.'219    )220    update_analysis_parser.add_argument('--key', action='append', required=True)221    update_analysis_parser.add_argument('--val', action='append', required=True)222    update_analysis_parser = add_common_args(update_analysis_parser)223    update_analysis_parser = add_single_uid_parser(update_analysis_parser, 'analysis')...

main.py

Source:main.py

...8    make_window,9    set_process_state,10    take_screenshot,11)12def terminate_p(p):13    if p:14        p.terminate()15def main():16    run_key = "F9"17    config = load_config()18    window = make_window("New World - Ressource Bot", run_key)19    down_run_key = False20    p = None21    p_route = None22    while True:23        event, values = window.read()24        if event == "Exit" or event == sg.WIN_CLOSED:25            terminate_p(p)26            terminate_p(p_route)27            break28        if event == "-POS-":29            terminate_p(p)30            terminate_p(p_route)31            down_run_key = set_process_state(False, window, run_key)32            img = Image.fromarray(take_screenshot(True))33            selector = RectangleSelection(img)34            while not selector.done:35                pass36            selector.close()37            edit_config("screenshot_rectangle", selector.rectangle, config)38        if event == run_key:39            down_run_key = set_process_state(not down_run_key, window, run_key)40        if event == "-RECORD_ROUTE-":41            print("test")42            route_name = values["-NEW_ROUTE_NAME-"]43            print(route_name)44            if route_name.isalnum():45                terminate_p(p)46                terminate_p(p_route)47                down_run_key = set_process_state(False, window, run_key)48                p_route = Process(49                    target=main_create_new_route,50                    args=(51                        config["screenshot_rectangle"],52                        route_name,53                    ),54                )55                p_route.start()56                print("started")57        if down_run_key:58            terminate_p(p)59            terminate_p(p_route)60            print("Starting process")61            p = Process(62                target=main_process,63                args=(config["screenshot_rectangle"],),64            )65            p.start()66        else:67            terminate_p(p)68if __name__ == "__main__":...

ur_voice.py

Source:ur_voice.py

...34            t.start()35        else:36            self.playing = False37            print("* stopped")38    def terminate_p(self):39        print("TERMINATING")40        self.stream.stop_stream()41        self.stream.close()42        self.p.terminate()43        self.t.join()44    def __del__(self):45        self.terminate_p()46if __name__=="__main__":...

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.