detach_proc() - Code Metrics - Inspection of "add detach and config file support" - neovim/python-gui - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#21)

unknown

created 2016-04-27 08:37 UTC

detach_proc() D

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
cc	9
dl	0
loc	53
rs	4.9852

How to fix Long Method

"""CLI for accessing the gtk/tickit UIs implemented by this package."""
import os
import resource
import sys
import shlex

import click
import yaml

from .ui_bridge import UIBridge
from neovim import attach
from neovim.compat import IS_PYTHON3


CONFIG_FILES = (
    '.pynvim.yaml',
    '~/.pynvim.yaml',
    '~/.config/pynvim/config.yaml'
)


def load_config(config_file):
    """Load config values from yaml."""

    if config_file:
        with open(config_file) as f:
            return yaml.load(f)

    else:
        for config_file in CONFIG_FILES:
            try:
                with open(os.path.expanduser(config_file)) as f:
                    return yaml.load(f)

            except IOError:
                pass

    return {}


# http://code.activestate.com/recipes/278731-creating-a-daemon-the-python-way/
def detach_proc(workdir='.', umask=0):
    """Detach a process from the controlling terminal and run it in the
    background as a daemon.
    """

    # Default maximum for the number of available file descriptors.
    MAXFD = 1024

    # The standard I/O file descriptors are redirected to /dev/null by default.
    if (hasattr(os, "devnull")):
        REDIRECT_TO = os.devnull
    else:
        REDIRECT_TO = "/dev/null"

    try:
        pid = os.fork()
    except OSError, e:
        raise Exception, "%s [%d]" % (e.strerror, e.errno)

    if (pid == 0):
        os.setsid()

        try:
            pid = os.fork()

        except OSError, e:
            raise Exception, "%s [%d]" % (e.strerror, e.errno)

        if (pid == 0):
            os.chdir(workdir)
            os.umask(umask)
        else:
            os._exit(0)
    else:
        os._exit(0)

        maxfd = resource.getrlimit(resource.RLIMIT_NOFILE)[1]
        if (maxfd == resource.RLIM_INFINITY):
            maxfd = MAXFD

            # Iterate through and close all file descriptors.
            for fd in range(0, maxfd):
                try:
                    os.close(fd)
                except OSError:
                    pass

    os.open(REDIRECT_TO, os.O_RDWR)

    os.dup2(0, 1)
    os.dup2(0, 2)

    return(0)


@click.command(context_settings=dict(allow_extra_args=True))
@click.option('--prog')
@click.option('--notify', '-n', default=False, is_flag=True)
@click.option('--listen', '-l')
@click.option('--connect', '-c')
@click.option('--profile',
              default='disable',
              type=click.Choice(['ncalls', 'tottime', 'percall', 'cumtime',
                                 'name', 'disable']))
@click.option('config_file', '--config', type=click.Path(exists=True))
@click.option('--detach/--no-detach', default=True, is_flag=True)
@click.pass_context
def main(ctx, prog, notify, listen, connect, profile, config_file, detach):
    """Entry point."""

    if detach:
        exit_code = detach_proc()

    address = connect or listen

    if address:
        import re
        p = re.compile(r'^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}(?:\:\d{1,5})?$')

        if p.match(address):
            args = ('tcp',)
            kwargs = {'address': address}
        else:
            args = ('socket',)
            kwargs = {'path': address}

    if connect:
        # connect to existing instance listening on address
        nvim = attach(*args, **kwargs)
    elif listen:
        # spawn detached instance listening on address and connect to it
        import os
        import time
        from subprocess import Popen
        os.environ['NVIM_LISTEN_ADDRESS'] = address
        nvim_argv = shlex.split(prog or 'nvim --headless') + ctx.args
        # spawn the nvim with stdio redirected to /dev/null.
        dnull = open(os.devnull)
        p = Popen(nvim_argv, stdin=dnull, stdout=dnull, stderr=dnull)
        dnull.close()
        while p.poll() or p.returncode is None:
            try:
                nvim = attach(*args, **kwargs)
                break
            except IOError:
                # socket not ready yet
                time.sleep(0.050)
    else:
        # spawn embedded instance
        nvim_argv = shlex.split(prog or 'nvim --embed') + ctx.args
        nvim = attach('child', argv=nvim_argv)

    from .gtk_ui import GtkUI
    config = load_config(config_file)
    ui = GtkUI(config)
    bridge = UIBridge()
    bridge.connect(nvim, ui, profile if profile != 'disable' else None, notify)

    if detach:
        sys.exit(exit_code)


if __name__ == '__main__':
    main()


1			"""CLI for accessing the gtk/tickit UIs implemented by this package."""
2			import os
3			import resource
4			import sys
5			import shlex
6
7			import click
8			import yaml
9
10			from .ui_bridge import UIBridge
11			from neovim import attach
12			from neovim.compat import IS_PYTHON3
13
14
15			CONFIG_FILES = (
16			'.pynvim.yaml',
17			'~/.pynvim.yaml',
18			'~/.config/pynvim/config.yaml'
19			)
20
21
22			def load_config(config_file):
23			"""Load config values from yaml."""
24
25			if config_file:
26			with open(config_file) as f:
27			return yaml.load(f)
28
29			else:
30			for config_file in CONFIG_FILES:
31			try:
32			with open(os.path.expanduser(config_file)) as f:
33			return yaml.load(f)
34
35			except IOError:
36			pass
37
38			return {}
39
40
41			# http://code.activestate.com/recipes/278731-creating-a-daemon-the-python-way/
42			def detach_proc(workdir='.', umask=0):
43			"""Detach a process from the controlling terminal and run it in the
44			background as a daemon.
45			"""
46
47			# Default maximum for the number of available file descriptors.
48			MAXFD = 1024
49
50			# The standard I/O file descriptors are redirected to /dev/null by default.
51			if (hasattr(os, "devnull")):
52			REDIRECT_TO = os.devnull
53			else:
54			REDIRECT_TO = "/dev/null"
55
56			try:
57			pid = os.fork()
58			except OSError, e:
59			raise Exception, "%s [%d]" % (e.strerror, e.errno)
60
61			if (pid == 0):
62			os.setsid()
63
64			try:
65			pid = os.fork()
66
67			except OSError, e:
68			raise Exception, "%s [%d]" % (e.strerror, e.errno)
69
70			if (pid == 0):
71			os.chdir(workdir)
72			os.umask(umask)
73			else:
74			os._exit(0)
75			else:
76			os._exit(0)
77
78			maxfd = resource.getrlimit(resource.RLIMIT_NOFILE)[1]
79			if (maxfd == resource.RLIM_INFINITY):
80			maxfd = MAXFD
81
82			# Iterate through and close all file descriptors.
83			for fd in range(0, maxfd):
84			try:
85			os.close(fd)
86			except OSError:
87			pass
88
89			os.open(REDIRECT_TO, os.O_RDWR)
90
91			os.dup2(0, 1)
92			os.dup2(0, 2)
93
94			return(0)
95
96
97			@click.command(context_settings=dict(allow_extra_args=True))
98			@click.option('--prog')
99			@click.option('--notify', '-n', default=False, is_flag=True)
100			@click.option('--listen', '-l')
101			@click.option('--connect', '-c')
102			@click.option('--profile',
103			default='disable',
104			type=click.Choice(['ncalls', 'tottime', 'percall', 'cumtime',
105			'name', 'disable']))
106			@click.option('config_file', '--config', type=click.Path(exists=True))
107			@click.option('--detach/--no-detach', default=True, is_flag=True)
108			@click.pass_context
109			def main(ctx, prog, notify, listen, connect, profile, config_file, detach):
110			"""Entry point."""
111
112			if detach:
113			exit_code = detach_proc()
114
115			address = connect or listen
116
117			if address:
118			import re
119			p = re.compile(r'^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}(?:\:\d{1,5})?$')
120
121			if p.match(address):
122			args = ('tcp',)
123			kwargs = {'address': address}
124			else:
125			args = ('socket',)
126			kwargs = {'path': address}
127
128			if connect:
129			# connect to existing instance listening on address
130			nvim = attach(args, *kwargs)
131			elif listen:
132			# spawn detached instance listening on address and connect to it
133			import os
134			import time
135			from subprocess import Popen
136			os.environ['NVIM_LISTEN_ADDRESS'] = address
137			nvim_argv = shlex.split(prog or 'nvim --headless') + ctx.args
138			# spawn the nvim with stdio redirected to /dev/null.
139			dnull = open(os.devnull)
140			p = Popen(nvim_argv, stdin=dnull, stdout=dnull, stderr=dnull)
141			dnull.close()
142			while p.poll() or p.returncode is None:
143			try:
144			nvim = attach(args, *kwargs)
145			break
146			except IOError:
147			# socket not ready yet
148			time.sleep(0.050)
149			else:
150			# spawn embedded instance
151			nvim_argv = shlex.split(prog or 'nvim --embed') + ctx.args
152			nvim = attach('child', argv=nvim_argv)
153
154			from .gtk_ui import GtkUI
155			config = load_config(config_file)
156			ui = GtkUI(config)
157			bridge = UIBridge()
158			bridge.connect(nvim, ui, profile if profile != 'disable' else None, notify)
159
160			if detach:
161			sys.exit(exit_code)
162
163
164			if __name__ == '__main__':
165			main()
166

neovim / python-gui

Pull Request — master (#21)

detach_proc() D

Complexity

Size

Duplication

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like