How to use get_monitor_cmd method in autotest

Best Python code snippet using autotest_python

autotest.py

Source:autotest.py

...401               monitor_dir, '-H autoserv']402        cmd += self.get_base_cmd_args(section)403        cmd += ['>/dev/null', '2>/dev/null', '&']404        return ' '.join(cmd)405    def get_monitor_cmd(self, monitor_dir, stdout_read, stderr_read):406        cmd = [os.path.join(self.autodir, 'bin', 'autotestd_monitor'),407               monitor_dir, str(stdout_read), str(stderr_read)]408        return ' '.join(cmd)409    def get_client_log(self):410        """Find what the "next" client.* prefix should be411        @returns A string of the form client.INTEGER that should be prefixed412            to all client debug log files.413        """414        max_digit = -1415        debug_dir = os.path.join(self.results_dir, 'debug')416        client_logs = glob.glob(os.path.join(debug_dir, 'client.*.*'))417        for log in client_logs:418            _, number, _ = log.split('.', 2)419            if number.isdigit():420                max_digit = max(max_digit, int(number))421        return 'client.%d' % (max_digit + 1)422    def copy_client_config_file(self, client_log_prefix=None):423        """424        Create and copy the client config file based on the server config.425        @param client_log_prefix: Optional prefix to prepend to log files.426        """427        client_config_file = self._create_client_config_file(client_log_prefix)428        self.host.send_file(client_config_file, self.config_file)429        os.remove(client_config_file)430    def _create_client_config_file(self, client_log_prefix=None):431        """432        Create a temporary file with the [CLIENT] section configuration values433        taken from the server global_config.ini.434        @param client_log_prefix: Optional prefix to prepend to log files.435        @return: Path of the temporary file generated.436        """437        config = global_config.global_config.get_section_values('CLIENT')438        if client_log_prefix:439            config.set('CLIENT', 'default_logging_name', client_log_prefix)440        return self._create_aux_file(config.write)441    def _create_aux_file(self, func, *args):442        """443        Creates a temporary file and writes content to it according to a444        content creation function. The file object is appended to *args, which445        is then passed to the content creation function446        @param func: Function that will be used to write content to the447                temporary file.448        @param *args: List of parameters that func takes.449        @return: Path to the temporary file that was created.450        """451        fd, path = tempfile.mkstemp(dir=self.host.job.tmpdir)452        aux_file = os.fdopen(fd, "w")453        try:454            list_args = list(args)455            list_args.append(aux_file)456            func(*list_args)457        finally:458            aux_file.close()459        return path460    @staticmethod461    def is_client_job_finished(last_line):462        return bool(re.match(r'^END .*\t----\t----\t.*$', last_line))463    @staticmethod464    def is_client_job_rebooting(last_line):465        return bool(re.match(r'^\t*GOOD\t----\treboot\.start.*$', last_line))466    def log_unexpected_abort(self, stderr_redirector):467        stderr_redirector.flush_all_buffers()468        msg = "Autotest client terminated unexpectedly"469        self.host.job.record("END ABORT", None, None, msg)470    def _execute_in_background(self, section, timeout):471        full_cmd = self.get_background_cmd(section)472        devnull = open(os.devnull, "w")473        self.copy_client_config_file(self.get_client_log())474        self.host.job.push_execution_context(self.results_dir)475        try:476            result = self.host.run(full_cmd, ignore_status=True,477                                   timeout=timeout,478                                   stdout_tee=devnull,479                                   stderr_tee=devnull)480        finally:481            self.host.job.pop_execution_context()482        return result483    @staticmethod484    def _strip_stderr_prologue(stderr):485        """Strips the 'standard' prologue that get pre-pended to every486        remote command and returns the text that was actually written to487        stderr by the remote command."""488        stderr_lines = stderr.split("\n")[1:]489        if not stderr_lines:490            return ""491        elif stderr_lines[0].startswith("NOTE: autotestd_monitor"):492            del stderr_lines[0]493        return "\n".join(stderr_lines)494    def _execute_daemon(self, section, timeout, stderr_redirector,495                        client_disconnect_timeout):496        monitor_dir = self.host.get_tmp_dir()497        daemon_cmd = self.get_daemon_cmd(section, monitor_dir)498        # grab the location for the server-side client log file499        client_log_prefix = self.get_client_log()500        client_log_path = os.path.join(self.results_dir, 'debug',501                                       client_log_prefix + '.log')502        client_log = open(client_log_path, 'w', 0)503        self.copy_client_config_file(client_log_prefix)504        stdout_read = stderr_read = 0505        self.host.job.push_execution_context(self.results_dir)506        try:507            self.host.run(daemon_cmd, ignore_status=True, timeout=timeout)508            disconnect_warnings = []509            while True:510                monitor_cmd = self.get_monitor_cmd(monitor_dir, stdout_read,511                                                   stderr_read)512                try:513                    result = self.host.run(monitor_cmd, ignore_status=True,514                                           timeout=timeout,515                                           stdout_tee=client_log,516                                           stderr_tee=stderr_redirector)517                except error.AutoservRunError, e:518                    result = e.result_obj519                    result.exit_status = None520                    disconnect_warnings.append(e.description)521                    stderr_redirector.log_warning(522                        "Autotest client was disconnected: %s" % e.description,523                        "NETWORK")524                except error.AutoservSSHTimeout:...

recordingtools.py

Source:recordingtools.py

1""" Python equivalent of akai section recordingtools2Methods to facilitate recording3"""4__author__ =  'Walco van Loon'5__version__ =  '0.2'6from aksy.devices.akai.sysex import Command7import aksy.devices.akai.sysex_types8class Recordingtools:9    def __init__(self, z48):10        self.sampler = z4811        self.get_status_cmd = Command('_', '\x30\x01', 'recordingtools', 'get_status', (), None)12        self.get_progress_cmd = Command('_', '\x30\x02', 'recordingtools', 'get_progress', (), None)13        self.get_max_rec_time_cmd = Command('_', '\x30\x03', 'recordingtools', 'get_max_rec_time', (), None)14        self.arm_cmd = Command('_', '\x30\x10', 'recordingtools', 'arm', (), None)15        self.start_cmd = Command('_', '\x30\x11', 'recordingtools', 'start', (), None)16        self.stop_cmd = Command('_', '\x30\x12', 'recordingtools', 'stop', (), None)17        self.cancel_cmd = Command('_', '\x30\x13', 'recordingtools', 'cancel', (), None)18        self.start_playing_cmd = Command('_', '\x30\x20', 'recordingtools', 'start_playing', (), None)19        self.stop_playing_cmd = Command('_', '\x30\x21', 'recordingtools', 'stop_playing', (), None)20        self.keep_cmd = Command('_', '\x30\x22', 'recordingtools', 'keep', (), None)21        self.delete_cmd = Command('_', '\x30\x23', 'recordingtools', 'delete', (), None)22        self.set_input_cmd = Command('_', '\x32\x01', 'recordingtools', 'set_input', (aksy.devices.akai.sysex_types.BYTE,), None)23        self.set_mode_cmd = Command('_', '\x32\x02', 'recordingtools', 'set_mode', (aksy.devices.akai.sysex_types.BYTE,), None)24        self.enable_monitor_cmd = Command('_', '\x32\x03', 'recordingtools', 'enable_monitor', (aksy.devices.akai.sysex_types.BOOL,), None)25        self.set_rec_time_cmd = Command('_', '\x32\x04', 'recordingtools', 'set_rec_time', (aksy.devices.akai.sysex_types.DWORD,), None)26        self.set_orig_pitch_cmd = Command('_', '\x32\x05', 'recordingtools', 'set_orig_pitch', (aksy.devices.akai.sysex_types.BYTE,), None)27        self.set_threshold_cmd = Command('_', '\x32\x06', 'recordingtools', 'set_threshold', (aksy.devices.akai.sysex_types.SBYTE,), None)28        self.set_trigger_src_cmd = Command('_', '\x32\x07', 'recordingtools', 'set_trigger_src', (aksy.devices.akai.sysex_types.BYTE,), None)29        self.set_bit_depth_cmd = Command('_', '\x32\x08', 'recordingtools', 'set_bit_depth', (aksy.devices.akai.sysex_types.BYTE,), None)30        self.set_prerec_time_cmd = Command('_', '\x32\x09', 'recordingtools', 'set_prerec_time', (aksy.devices.akai.sysex_types.WORD,), None)31        self.set_dest_cmd = Command('_', '\x32\x0A', 'recordingtools', 'set_dest', (aksy.devices.akai.sysex_types.BYTE,), None)32        self.set_name_cmd = Command('_', '\x32\x10', 'recordingtools', 'set_name', (aksy.devices.akai.sysex_types.STRING,), None)33        self.set_name_seed_cmd = Command('_', '\x32\x11', 'recordingtools', 'set_name_seed', (aksy.devices.akai.sysex_types.STRING,), None)34        self.set_autorec_mode_cmd = Command('_', '\x32\x12', 'recordingtools', 'set_autorec_mode', (aksy.devices.akai.sysex_types.BOOL,), None)35        self.set_autonormalize_cmd = Command('_', '\x32\x13', 'recordingtools', 'set_autonormalize', (aksy.devices.akai.sysex_types.BOOL,), None)36        self.get_input_cmd = Command('_', '\x33\x01', 'recordingtools', 'get_input', (), None)37        self.get_mode_cmd = Command('_', '\x33\x02', 'recordingtools', 'get_mode', (), None)38        self.get_monitor_cmd = Command('_', '\x33\x03', 'recordingtools', 'get_monitor', (), None)39        self.get_rec_time_cmd = Command('_', '\x33\x04', 'recordingtools', 'get_rec_time', (), None)40        self.get_pitch_cmd = Command('_', '\x33\x05', 'recordingtools', 'get_pitch', (), None)41        self.get_threshold_cmd = Command('_', '\x33\x06', 'recordingtools', 'get_threshold', (), None)42        self.get_trigger_src_cmd = Command('_', '\x33\x07', 'recordingtools', 'get_trigger_src', (), None)43        self.get_bit_depth_cmd = Command('_', '\x33\x08', 'recordingtools', 'get_bit_depth', (), None)44        self.get_prerec_time_cmd = Command('_', '\x33\x09', 'recordingtools', 'get_prerec_time', (), None)45        self.get_dest_cmd = Command('_', '\x33\x0A', 'recordingtools', 'get_dest', (), None)46        self.get_name_cmd = Command('_', '\x33\x10', 'recordingtools', 'get_name', (), None)47        self.get_name_seed_cmd = Command('_', '\x33\x11', 'recordingtools', 'get_name_seed', (), None)48        self.get_autorec_mode_cmd = Command('_', '\x33\x12', 'recordingtools', 'get_autorec_mode', (), None)49        self.get_autonormalize_cmd = Command('_', '\x33\x13', 'recordingtools', 'get_autonormalize', (), None)50    def get_status(self):51        """Get Record Status52        Returns:53            BYTE54        """55        return self.sampler.execute(self.get_status_cmd, ())56    def get_progress(self):57        """Get Record Progress58        Returns:59            DWORD60        """61        return self.sampler.execute(self.get_progress_cmd, ())62    def get_max_rec_time(self):63        """Get Maximum Record Time64        Returns:65            DWORD66        """67        return self.sampler.execute(self.get_max_rec_time_cmd, ())68    def arm(self):69        """Arm Recording70        """71        return self.sampler.execute(self.arm_cmd, ())72    def start(self):73        """Start Recording74        """75        return self.sampler.execute(self.start_cmd, ())76    def stop(self):77        """Stop Recording78        """79        return self.sampler.execute(self.stop_cmd, ())80    def cancel(self):81        """Cancel Recording82        """83        return self.sampler.execute(self.cancel_cmd, ())84    def start_playing(self):85        """Play Recorded Sample Start86        """87        return self.sampler.execute(self.start_playing_cmd, ())88    def stop_playing(self):89        """Play Recorded Sample Stop90        """91        return self.sampler.execute(self.stop_playing_cmd, ())92    def keep(self):93        """Keep Recorded Sample. Sample with name assigned above, is added to the list of available samples.94        """95        return self.sampler.execute(self.keep_cmd, ())96    def delete(self):97        """Delete Recorded Sample98        """99        return self.sampler.execute(self.delete_cmd, ())100    def set_input(self, arg0):101        """Set Input <Data1> = (0=ANALOGUE, 1=DIGITAL, 2=MAIN OUT 3=ADAT1/2, 4=ADAT3/4, 5=ADAT5/6, 6=ADAT7/8)102        """103        return self.sampler.execute(self.set_input_cmd, (arg0, ))104    def set_mode(self, arg0):105        """Set Record Mode <Data1> = (0=STEREO, 1=MONO L, 2=MONO R, 3=L/R MIX)106        """107        return self.sampler.execute(self.set_mode_cmd, (arg0, ))108    def enable_monitor(self, arg0):109        """Set Record Monitor110        """111        return self.sampler.execute(self.enable_monitor_cmd, (arg0, ))112    def set_rec_time(self, arg0):113        """Set Record Time <Data1> = time in seconds. If <Data1> = 0, MANUAL mode is enabled.114        """115        return self.sampler.execute(self.set_rec_time_cmd, (arg0, ))116    def set_orig_pitch(self, arg0):117        """Set Original Pitch118        """119        return self.sampler.execute(self.set_orig_pitch_cmd, (arg0, ))120    def set_threshold(self, arg0):121        """Set Threshold <Data1> = threshold in dB (-63,0)122        """123        return self.sampler.execute(self.set_threshold_cmd, (arg0, ))124    def set_trigger_src(self, arg0):125        """Set Trigger Source (0=OFF, 1=AUDIO, 2=MIDI)126        """127        return self.sampler.execute(self.set_trigger_src_cmd, (arg0, ))128    def set_bit_depth(self, arg0):129        """Set Bit Depth <Data1> = (0=16-bit, 1=24-bit)130        """131        return self.sampler.execute(self.set_bit_depth_cmd, (arg0, ))132    def set_prerec_time(self, arg0):133        """Set Pre-recording Time <Data1> = time in ms134        """135        return self.sampler.execute(self.set_prerec_time_cmd, (arg0, ))136    def set_dest(self, arg0):137        """Set Recording Detination <Data1> = (0=RAM, 1=DISK) Note: seems to be unsupported138        """139        return self.sampler.execute(self.set_dest_cmd, (arg0, ))140    def set_name(self, arg0):141        """Set Record Name142        """143        return self.sampler.execute(self.set_name_cmd, (arg0, ))144    def set_name_seed(self, arg0):145        """Set Record Name Seed146        """147        return self.sampler.execute(self.set_name_seed_cmd, (arg0, ))148    def set_autorec_mode(self, arg0):149        """Set Auto-Record Mode <Data1> = (0=OFF, 1=ON)150        """151        return self.sampler.execute(self.set_autorec_mode_cmd, (arg0, ))152    def set_autonormalize(self, arg0):153        """Set Auto-Normalise Mode <Data1> = (0=OFF, 1=ON)154        """155        return self.sampler.execute(self.set_autonormalize_cmd, (arg0, ))156    def get_input(self):157        """Get Input (0=ANALOGUE, 1=DIGITAL, 2=MAIN OUT, 3=ADAT1/2, 4=ADAT3/4, 5=ADAT5/6, 6=ADAT7/8)158        Returns:159            BYTE160        """161        return self.sampler.execute(self.get_input_cmd, ())162    def get_mode(self):163        """Get Record Mode (0=STEREO, 1=MONO L, 2=MONO R, 3=L/R MIX)164        """165        return self.sampler.execute(self.get_mode_cmd, ())166    def get_monitor(self):167        """Get Record Monitor <Reply> = (0=OFF, 1=ON)168        Returns:169            BOOL170        """171        return self.sampler.execute(self.get_monitor_cmd, ())172    def get_rec_time(self):173        """Get Record Time <Reply> = time in seconds.174        Returns:175            DWORD176        """177        return self.sampler.execute(self.get_rec_time_cmd, ())178    def get_pitch(self):179        """Get Original Pitch180        Returns:181            BYTE182        """183        return self.sampler.execute(self.get_pitch_cmd, ())184    def get_threshold(self):185        """Get Threshold <Reply> = threshold in dB -63,0186        Returns:187            SBYTE188        """189        return self.sampler.execute(self.get_threshold_cmd, ())190    def get_trigger_src(self):191        """Get Trigger Source <Reply> = (0=OFF, 1=AUDIO, 2=MIDI)192        Returns:193            BYTE194        """195        return self.sampler.execute(self.get_trigger_src_cmd, ())196    def get_bit_depth(self):197        """Get Bit Depth <Reply> = (0=16-bit, 1=24-bit)198        Returns:199            BYTE200        """201        return self.sampler.execute(self.get_bit_depth_cmd, ())202    def get_prerec_time(self):203        """Get Pre-recording Time <Reply> = time in ms204        Returns:205            WORD206        """207        return self.sampler.execute(self.get_prerec_time_cmd, ())208    def get_dest(self):209        """Get Recording Destination <Reply> = (0=RAM, 1=DISK) Note: seems to be unsupported210        Returns:211            BYTE212        """213        return self.sampler.execute(self.get_dest_cmd, ())214    def get_name(self):215        """Get Record Name216        Returns:217            STRING218        """219        return self.sampler.execute(self.get_name_cmd, ())220    def get_name_seed(self):221        """Get Record Name Seed222        Returns:223            STRING224        """225        return self.sampler.execute(self.get_name_seed_cmd, ())226    def get_autorec_mode(self):227        """Get Auto-Record Mode <Reply> = (0=OFF, 1=ON)228        Returns:229            BOOL230        """231        return self.sampler.execute(self.get_autorec_mode_cmd, ())232    def get_autonormalize(self):233        """Get Auto-Normalise Mode <Reply> = (0=OFF, 1=ON)234        Returns:235            BOOL236        """...

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.