Piping and Forking in Python

   
   
 coding  conky  perl  
 code  
Table of Content

This article is one part of a collection. All integrated, one related to another, featured with summary. So we can compare each other quickly.

Tutorial/ Guidance/ Article:

[ Pipe Overview ]  [ BASH ]  [ Perl ]  [ Python ]  [ Ruby ]  [ PHP ]  [ Lua ]  [ Haskell ]

Source Code Directory:

[ BASH ]  [ Perl ]  [ Python ]  [ Ruby ]  [ PHP ]  [ Lua ]  [ Haskell ]

Python Plumber.

Goal: A script that continuously show date and time, with Dzen2, and Conky..

Before you dip your toe to scripting, you might desire to know the reason by reading this overview.

Reading

Table of Content

Start Simple

Welcome to n00berland. Begin with simple script. We will use this loop as a source feed to pipe. This step won’t introduce Pipe nor Fork.

This script only show an infinite loop showing local time. Each updated in one second interval. We manage this interval by delaying, using sleep code.

There at least two choices to show current time in Python. Using time library, or using datetime library.

Source:

Using time library.

#!/usr/bin/env python3

import time

timeformat = '%Y-%m-%d %H:%M:%S'

while True:
    timestr = time.strftime(timeformat)
    print(timestr)
    time.sleep(1)

Using datetime library.

#!/usr/bin/env python3

import datetime
from time import sleep

timeformat = '{0:%Y-%m-%d %H:%M:%S}'

while True:
    datestr = timeformat.format(datetime.datetime.now())
    print(datestr)
    sleep(1)

Call to this simple code would produce time marching, one after another, below the command line prompt.

Pipe: Basic

Similar Code
[ BASH basic ]  [ Perl basic ]  [ Python basic datetime ]  [ Python basictime ]  [ Ruby basic ]  [ PHP basic ]  [ Lua basic ]  [ Haskell basic ]

External Command as Source Feed

Beside previous simple loop that is used as Internal Command, this tutorial also provide Conky as External Command in asset directory. I made it as simple as possible.

Source:

conky.config = {
    out_to_x = false,
    out_to_console = true,
    short_units = true,
    update_interval = 1
}

conky.text = [[\
${time %a %b %d %H:%M:%S}\
]]

Spawning Using System Shell

Using os.system shell is simple and straightforward. But it does not have any ability, to stream internal function process, that required later on this article.

I haven’t explore os.popen, since we are going to use subprocess.Popen anyway.

Source:

Using os.system:

#!/usr/bin/env python3

import os

dirname = os.path.dirname(os.path.abspath(__file__))
path    = dirname + "/../assets"
cmdin   = 'conky -c ' + path + '/conky.lua'

cmdout  = 'less' # or 'dzen2'

cmd     = cmdin + ' | ' + cmdout

os.system(cmd)

Using os.popen:

#!/usr/bin/env python3

import os
import subprocess

dirname = os.path.dirname(os.path.abspath(__file__))
path    = dirname + "/../assets"
cmdin   = 'conky -c ' + path + '/conky.lua'

cmdout  = 'dzen2' # no less

cmd     = cmdin + ' | ' + cmdout

process = os.popen(cmd, 'r')
Similar Code
[ Perl system ]  [ Python system ]  [ Python popen ]  [ Ruby system ]  [ Ruby spawn ]  [ Ruby shell ]  [ Haskell system ]

A Unidirectional Pipe Between External Command

This step is overview of Pipe between two external command. This script is using conky as pipe source feed and less as pipe target. Showing time and date forever in the console.

This infinite pipe run in time-less fashioned.

I add dirname, relative to the Python source, to locate the conky script assets.

We use this awesomely cool Python’s subprocess.Popen mechanism.

Source:

#!/usr/bin/env python3

# https://pymotw.com/2/subprocess/

import os
import subprocess

dirname = os.path.dirname(os.path.abspath(__file__))
path    = dirname + "/../assets"
cmdin   = 'conky -c ' + path + '/conky.lua'

cmdout  = 'less -K' # or 'dzen2'

pipein = subprocess.Popen(
        [cmdin], 
        stdout = subprocess.PIPE, 
        stderr = subprocess.STDOUT,
        shell  = True,
        universal_newlines = True
    )

pipeout = subprocess.Popen(
        [cmdout],
        stdin  = pipein.stdout,
        shell  = True,
        universal_newlines = True
    )

# http://kendriu.com/how-to-use-pipes-in-python-subprocesspopen-objects

pipein.stdout.close()
outputs, errors = pipeout.communicate()

# avoid zombie apocalypse
pipeout.wait()

You can see, how simple it is. This would have less output similar to this below.

Pipe: to Less

Your wallpaper might be different than mine.

Similar Code
[ BASH native ]  [ Perl uni IO ]  [ Perl uni open ]  [ Perl IPC open2 ]  [ Python subProcess ]  [ Ruby popen ]  [ PHP popen ]  [ PHP proc open ]  [ Lua popen ]  [ Haskell createProcess ]

How does it works ?

First process create a new stdout handle pipein.stdout. And the second process use it as feed to pipeout.stdin. pipeout.communicate() manage both.

pipein = subprocess.Popen(
...
        stdout = subprocess.PIPE, 
    )

pipeout = subprocess.Popen(
...
        stdin  = pipein.stdout,
    )

A Unidirectional Pipe from Internal Function

Using internal function as source feed to external command is straight forward. This should be self explanatory.

Do not forget to flush.

As previous example, we are using two mechanism, open and subprocess.Popen.

A more complex try-except block also provided in link below. But I would not make thos tutorial too complex.

Source:

#!/usr/bin/env python3

# https://gist.github.com/waylan/2353749

import datetime
import time
import subprocess

timeformat = '{0:%Y-%m-%d %H:%M:%S}'

cmdout  = 'less -K' # or 'dzen2'

process = subprocess.Popen(
        [cmdout], 
        stdin  = subprocess.PIPE,
        shell  = True,
        universal_newlines=True
    )

while True:
    datestr = timeformat.format(datetime.datetime.now())
    
    process.stdin.write(datestr + '\n')
    process.stdin.flush()
    
    time.sleep(1)

process.stdin.close()
process.wait()
Similar Code
[ BASH native ]  [ Perl uni IO ]  [ Perl uni open ]  [ Perl IPC open2 ]  [ Python subProcess ]  [ Ruby open3 ]  [ Ruby pipe IO ]  [ Ruby pty ]  [ Ruby popen ]  [ PHP popen ]  [ PHP popen ]  [ Lua popen ]  [ Haskell createProcess ]

How does it works ?

The same as previous. But instead of reading from pipein.stdout, it is managed by internal process using process.stdin.write().

    process.stdin.write(datestr + '\n')
    process.stdin.flush()

Fork Overview

This step use internal function as source feed, as continuation of previous step. To avoid complexity of longer script, most code is written inside function

This step use dzen2, with complete parameters. This dzen2 is forked, running in the background. Detached from the script, no need to wait for dzen2 to finish the script.

Source:

#!/usr/bin/env python3

# https://gist.github.com/waylan/2353749

import datetime
import time
import subprocess
import os
import signal

def get_dzen2_parameters():
    xpos    = '0'
    ypos    = '0'
    width   = '640'
    height  = '24'
    fgcolor = '#000000'
    bgcolor = '#ffffff'
    font    = '-*-fixed-medium-*-*-*-12-*-*-*-*-*-*-*'

    parameters  = '  -x '+xpos+' -y '+ypos+' -w '+width+' -h '+ height
    parameters += " -fn '"+font+"'"
    parameters += " -ta c -bg '"+bgcolor+"' -fg '"+fgcolor+"'"
    parameters += ' -title-name dzentop'

    return parameters;

def generated_output(process):
    timeformat = '{0:%Y-%m-%d %H:%M:%S}'

    while True:
        datestr = timeformat.format(datetime.datetime.now())
    
        process.stdin.write(datestr + '\n')
        process.stdin.flush()
    
        time.sleep(1)

def run_dzen2():
    cmdout  = 'dzen2 '+get_dzen2_parameters()

    pipeout = subprocess.Popen(
            [cmdout], 
            stdin  = subprocess.PIPE,
            shell  = True,
            universal_newlines=True
        )

    generated_output(pipeout)

    pipeout.stdin.close()
    
    # avoid zombie apocalypse
    pipeout.wait()

def detach_dzen2():
    pid = os.fork()
    
    if pid == 0:
        try:
            run_dzen2()
        finally:
            os.kill(pid, signal.SIGTERM)

# ----- ----- ----- ----- ----- ----- ----- ----- ----- ----- ----- ----
# main

# remove all dzen2 instance
os.system('pkill dzen2')

detach_dzen2()

This step also add system command that kill any previous dzen2 instance. So it will be guaranteed, that the dzen2 shown is coming from the latest script.

Similar Code
[ BASH fork ]  [ Perl fork ]  [ Python fork ]  [ Ruby fork ]  [ PHP fork ]  [ Lua fork ]  [ Haskell fork ]

How does it works ?

Any code after the os.fork executed in both parent and child. The child process has been detached from parent process. The only different is the $pid.

def detach_dzen2():
    pid = os.fork()
    
    if pid == 0:
        run_dzen2()

Polishing The Script

This step, we use conky again, as a source feed. And also parameterized dzen2 as continuation of previous step.

This step add optional transset transparency, detached from script. So we two forks, dzen and transset.

Source:

#!/usr/bin/env python3

import datetime
import time
import subprocess
import os
import signal

def get_dzen2_parameters():
    xpos    = '0'
    ypos    = '0'
    width   = '640'
    height  = '24'
    fgcolor = '#000000'
    bgcolor = '#ffffff'
    font    = '-*-fixed-medium-*-*-*-12-*-*-*-*-*-*-*'

    parameters  = '  -x '+xpos+' -y '+ypos+' -w '+width+' -h '+ height
    parameters += " -fn '"+font+"'"
    parameters += " -ta c -bg '"+bgcolor+"' -fg '"+fgcolor+"'"
    parameters += ' -title-name dzentop'

    return parameters;

def generated_output(pipeout):
    dirname = os.path.dirname(os.path.abspath(__file__))
    path    = dirname + "/../assets"
    cmdin   = 'conky -c ' + path + '/conky.lua'

    pipein = subprocess.Popen(
            [cmdin], 
            stdout = pipeout.stdin,
            stderr = subprocess.STDOUT,
            shell  = True,
            universal_newlines = True
        )

def run_dzen2():
    cmdout  = 'dzen2 '+get_dzen2_parameters()

    pipeout = subprocess.Popen(
            [cmdout], 
            stdin  = subprocess.PIPE,
            shell  = True,
            universal_newlines=True
        )

    generated_output(pipeout)

    pipeout.stdin.close()
    outputs, errors = pipeout.communicate()
    
    # avoid zombie apocalypse
    pipeout.wait()

def detach_dzen2():
    pid = os.fork()
    
    if pid == 0:
        try:
            run_dzen2()
            os._exit(1)
        finally:
            os.kill(pid, signal.SIGTERM)

def detach_transset():
    pid = os.fork()
    
    if pid == 0:
        try:
            time.sleep(1)
            os.system('transset .8 -n dzentop >/dev/null')
            os._exit(1)
        finally:
            os.kill(pid, signal.SIGTERM)

# ----- ----- ----- ----- ----- ----- ----- ----- ----- ----- ----- ----
# main

# remove all dzen2 instance
os.system('pkill dzen2')

# run process in the background
detach_dzen2()

# optional transparency
detach_transset()

This would have dzen2 output similar to this below.

Pipe: to Dzen2

You may use transset-df instead of transset.

Similar Code
[ BASH conky ]  [ Perl conky ]  [ Python conky ]  [ Ruby conky ]  [ PHP conky ]  [ Lua conky ]  [ Haskell conky ] [ BASH simple conky ] 

How does it works ?

Nothing new here.

Lemonbar

I also provide Lemonbar, instead of Dzen2. The code is very similar.

Source:

Similar Code
[ BASH lemon ]  [ Perl lemon ]  [ Python lemon ]  [ Ruby lemon ]  [ PHP lemon ]  [ Lua lemon ]  [ Haskell lemon ]

Coming up Next

There already an advance case of Pipe and Fork. Multitier, and bidirectional.

There above are some simple codes I put together. I’m mostly posting codes so I won’t have any problems finding it in the future.

Thank you for reading.