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

How to use exec_on_disconnect method in fMBT

Best Python code snippet using fMBT_python

client.py

Source:client.py

...36    Results of executed code are kept on the server after the37    connection is closed.38    Example: Register code that is executed in namespace "goodbye" after39    closing the connection:40    c.exec_in("goodbye", 'pythonshare_ns.exec_on_disconnect("code")')41    """42    def __init__(self, host_or_from_server, port_or_to_server,43                 password=None, namespace="default"):44        """Connect to a pythonshare server45        The server is listening to connections at host:port, or it can be46        communicated via file-like objects from_server and to_server.47        Parameters:48          host_or_from_server (string or file-like object)49                  string: host50                  file: file for receiving messages from the server51          port_or_to_server (int or file-like object)52                  int: port number53                  file: file for sending messages to the server54          password (string, optional)...

__init__.py

Source:__init__.py

1#!/usr/bin/env python32# -*- coding: utf-8 -*-3#4# See LICENSE for copyright and license details5import toml6import os7import sys8import subprocess9import functools10version = '2018.10'11CONNECTED = 'connected'12DISCONNECTED = 'disconnected'13CONFIG_FILE = os.path.expandvars(os.path.join(14        os.environ.get('XDG_CONFIG_HOME',15                       os.path.join('$HOME', '.config')),16        'screenconfig',17        'screenconfig.toml'))18DEFAULT_MONITOR = 'default'19class Event():20    def __init__(self, output=None, event=None, edid=None, screenid=None):21        self.event = event22        self.output = output23        self.edid = edid24        self.screenid = screenid25class MonitorConfiguration():26    def __init__(self, monitor_config):27        """28        @param :monitor_config: monitor configuration29        """30        self.monitor_config = monitor_config31    def __str__(self):32        return str(self.monitor_config)33    def __getattr__(self, name):34        if self.monitor_config:35            return self.monitor_config.get(name, None)36    def get(self, name, default=None):37        if name in self.monitor_config:38            return self.monitor_config[name]39        else:40            return default41class GlobalConfiguration():42    def __init__(self, outputs, monitors_config, default_monitor):43        """44        @param :outputs: dict of outputs and edids of the connected monitors45        @param :monitors_config: monitor configuration46        @param :default_monitor: default monitor in the configuration47        """48        self.outputs = outputs49        self.monitors_config = monitors_config50        self.default_monitor = default_monitor51    def __str__(self):52        return str(self.outputs, self.monitor_config, self.default_monitor)53    def __getattr__(self, name):54        if self.monitors:55            return self.monitors.get(name, None)56    @property57    def monitors(self):58        return self.monitors_config['monitors']59    @property60    def default(self):61        if self.monitors:62            return self.monitors[self.default_monitor]63def first(l):64    if l:65        for e in l:66            return e67def compose(*funcs):68    """Return a new function s.t.69       compose(f,g,...)(x) == f(g(...(x)))"""70    def inner(*args):71        result = args72        for f in reversed(funcs):73            if type(result) in (list, tuple):74                result = f(*result)75            else:76                result = f(result)77        return result78    return inner79def jam(*funcs):80    """Return a new function s.t.81       jam(f,g,...)(x) == g(x), f(x)"""82    def inner(*args):83        for f in reversed(funcs):84                f(*args)85    return inner86def _execute(f, cmd, *args, **kwargs):87    # print('_execute: ', ' '.join(cmd))88    return f(cmd, *args, **kwargs)89def execute(cmd, *args, **kwargs):90    return _execute(subprocess.call, cmd, *args, **kwargs)91def check_output(cmd, *args, **kwargs):92    return _execute(subprocess.check_output, cmd, *args, **kwargs)93def get_connected_outputs():94    """95    @returns dict {output: edid}96    """97    return {output_edid[0]: output_edid[1] if len(output_edid) == 2 else None98            for line in check_output(99                ['srandrd', 'list']).decode('utf-8').strip().split(os.linesep)100            for output_edid in (line.split(' ', 1), )101            }102def merge_monitor_config(default_config, monitor_config, output=None):103    merged_config = default_config.copy()104    if monitor_config:105        merged_config.update(monitor_config)106    if output:107        merged_config.update(108            monitor_config.get('outputs', {})109            .get(output, {}))110    if 'outputs' in merged_config:111        del merged_config['outputs']112    return merged_config113def get_monitor_configuration(config, monitor, output=None):114    assert config and monitor115    return MonitorConfiguration(116        merge_monitor_config(config.default,117                             first(map(118                                 config.monitors.get,119                                 filter(lambda m: m == monitor,120                                        config.monitors.keys())121                             )),122                             output))123def get_mon_configuration_for_edid(config, edid=None, output=None):124    if not config:125        return None126    # find the right configuration, might be multiple if the configuration127    # specifies the edid multiple times128    monitor_config = first(filter(None, [129        c if edid in c.get('edids', [])130        or edid in c.get('outputs', {}).get(output, {}).get('edids', [])131        else None132        for c in config.monitors.values()133    ]))134    if not monitor_config:135        return MonitorConfiguration(config.default)136    # use the first configuration merge configuration137    return MonitorConfiguration(138        merge_monitor_config(config.default,139                             monitor_config, output))140def get_wallpaper(monitor_config):141    if monitor_config and monitor_config.wallpaper:142        return os.path.expandvars(os.path.expanduser(monitor_config.wallpaper))143def set_wallpapers(config, event, commands):144    wallpapers = list(filter(None, map(145        lambda output, edid: compose(get_wallpaper,146                                     get_mon_configuration_for_edid)(config,147                                                                     edid,148                                                                     output),149        config.outputs.keys(),150        config.outputs.values()151    )))152    if wallpapers:153        commands.append(['feh', '--bg-fill', '--no-fehbg'] + wallpapers)154    return config, event, commands155def find_reference_output(config, reference_monitor, _reference_output=None):156    if _reference_output:157        return _reference_output158    reference_monitor_config = get_monitor_configuration(config,159                                                         reference_monitor,160                                                         _reference_output)161    reference_output = first(filter(None, map(162        lambda output, edid: output163        if edid in reference_monitor_config.edids else None,164        config.outputs.keys(),165        config.outputs.values()166    )))167    if reference_output:168        return reference_output169def format_cmd(event, cmd):170    return [arg.format(event) for arg in cmd]171def get_commands(event, cmdlist):172    if cmdlist:173        return [format_cmd(event, cmd) for cmd in cmdlist]174    return []175def activate_crtc(config, event, commands):176    monitor_config = get_mon_configuration_for_edid(config,177                                                    event.edid,178                                                    event.output)179    if not monitor_config:180        return None181    xrandr_cmd = ['xrandr', '--output', event.output]182    resolution = monitor_config.resolution183    if not resolution or resolution == 'auto':184        xrandr_cmd.append('--auto')185    else:186        xrandr_cmd.extend(('--mode', resolution))187    if monitor_config.xrandr_args:188        xrandr_cmd.extend(monitor_config.xrandr_args)189    if monitor_config.position and len(monitor_config.position) >= 2:190        reference_output = find_reference_output(config,191                                                 *monitor_config.position[1:3])192        if reference_output and reference_output != event.output:193            xrandr_cmd.extend((monitor_config.position[0], reference_output))194    commands.append(xrandr_cmd)195    commands.extend(get_commands(event, monitor_config.exec_on_disconnect))196    return config, event, commands197def deactivate_crtc(config, event, commands):198    monitor_config = get_mon_configuration_for_edid(config,199                                                    event.edid,200                                                    event.output)201    commands.append(['xrandr', '--output', event.output, '--off'])202    if monitor_config:203        commands.extend(get_commands(event, monitor_config.exec_on_disconnect))204    return config, event, commands205def process(config, event):206    assert event and event.event and config207    commands = []208    if event.event == CONNECTED:209        _, _, commands = compose(set_wallpapers,210                                 activate_crtc)(config, event, [])211    elif event.event == DISCONNECTED:212        _, _, commands = deactivate_crtc(config, event, [])213    return functools.reduce(214        lambda r1, r2: r1 if r1 and r1 > 0 else r2,215        map(execute, commands))216        # map(jam(execute, print), commands))217def main():218    event = Event(219        output=os.environ.get('SRANDRD_OUTPUT', None),220        event=os.environ.get('SRANDRD_EVENT', None),221        edid=os.environ.get('SRANDRD_EDID', None),222        screenid=os.environ.get('SRANDRD_SCREENID', None)223    )224    global_config = GlobalConfiguration(225        outputs=get_connected_outputs(),226        monitors_config=toml.load(CONFIG_FILE),227        default_monitor=os.environ.get('SCREENCONFIG_DEFAULT',228                                       DEFAULT_MONITOR),229    )230    return process(global_config,  event)231if __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.