DZone Snippets: pipe code

Thread Process //Pascal class

jonasraoni (Jonas Raoni Soares Silva) — Mon, 07 Jul 2008 07:25:12 GMT

A thread class to open processes on windows and retrieve its output (input isn't supported but it's easy to add).



unit Process;



interface



uses

  SysUtils, Windows, Classes, TLHelp32;



const

  INITIALIZATION_TIMEOUT = 10000;



type

  TStringArray = array of string;

  TProcessArray = array of Cardinal;

  TExceptionEvent = procedure(Sender: TObject; Exception: Exception) of object;

  EProcessError = class(Exception);



  TProcessThread = class(TThread)

  private

    FException: Exception;

    FWatching, FStarted, FSuspended: Boolean;

    FDirectory, FPath, FCommandLine, FEnvironment: PChar;

    FData: string;

    FOnProcessTerminated, FOnDataAvailable: TNotifyEvent;

    InputRead, InputWrite, OutputRead, OutputWrite: THandle;

    FMainProcess: PROCESS_INFORMATION;

    FOnException: TExceptionEvent;

    FOnProcessStarted: TNotifyEvent;

    function GetPriority: TThreadPriority;

    procedure SetPriority(const Value: TThreadPriority);

    procedure FreeResources;

  protected

    procedure CallDataAvailable; virtual;

    procedure CallProcessTerminated; virtual;

    procedure CallProcessOpened; virtual;

    procedure CallException; virtual;

    procedure Execute; override;

  public

    constructor Create(Path, CommandLine, Directory: string; Environment: TStrings = nil; Watch: Boolean = True);

    destructor Destroy; override;



    function IsProcessAlive: Boolean;

    procedure Resume;

    procedure Suspend;

    property OnDataAvailable: TNotifyEvent read FOnDataAvailable write FOnDataAvailable;

    property OnProcessTerminated: TNotifyEvent read FOnProcessTerminated write FOnProcessTerminated;

    property OnProcessStarted: TNotifyEvent read FOnProcessStarted write FOnProcessStarted;

    property OnException: TExceptionEvent read FOnException write FOnException;

    property Data: string read FData;

    property Process: PROCESS_INFORMATION read FMainProcess;

    property Priority: TThreadPriority read GetPriority write SetPriority;



  end;



  TProcessLineThread = class;

  TOnNewLineEvent = procedure(ProcessLine:  TProcessLineThread; const Line: string) of object;

  TProcessLineThread = class(TProcessThread)

  private

    FCurrentLine: string;

    FOnNewLine: TOnNewLineEvent;

    procedure DataAvailable(Sender: TObject);

    procedure Finished(Sender: TObject);

  public

    constructor Create(Path, CommandLine, Directory: string; Environment: TStrings = nil);

    property OnNewLine: TOnNewLineEvent read FOnNewLine write FOnNewLine;

  end;



function KillProcess(const Process: Cardinal): Boolean;

function GetChildrenProcesses(const Process: Cardinal; const IncludeParent: Boolean = True): TProcessArray;



implementation



const

  Priorities: array [TThreadPriority] of Integer =

   (THREAD_PRIORITY_IDLE, THREAD_PRIORITY_LOWEST, THREAD_PRIORITY_BELOW_NORMAL,

    THREAD_PRIORITY_NORMAL, THREAD_PRIORITY_ABOVE_NORMAL,

    THREAD_PRIORITY_HIGHEST, THREAD_PRIORITY_TIME_CRITICAL);



function GetChildrenProcesses(const Process: Cardinal; const IncludeParent: Boolean): TProcessArray;

var

  Snapshot: Cardinal;

  ProcessList: PROCESSENTRY32;

  Current: Integer;

begin

  Current := 0;

  SetLength(Result, 1);

  Result[0] := Process;

  repeat

    ProcessList.dwSize := SizeOf(PROCESSENTRY32);

    Snapshot := CreateToolhelp32Snapshot(TH32CS_SNAPPROCESS, 0);

    if (Snapshot = INVALID_HANDLE_VALUE) or not Process32First(Snapshot, ProcessList) then

      Continue;

    repeat

      if ProcessList.th32ParentProcessID = Result[Current] then

      begin

        SetLength(Result, Length(Result) + 1);

        Result[Length(Result) - 1] := ProcessList.th32ProcessID;

      end;

    until Process32Next(Snapshot, ProcessList) = False;

    Inc(Current);

  until Current >= Length(Result);

  if not IncludeParent then

    Result := Copy(Result, 2, Length(Result));

end;



function KillProcess(const Process: Cardinal): Boolean;

var

  Handle: Cardinal;

  List: TProcessArray;

  I: Integer;

begin

  Result := True;

  List := GetChildrenProcesses(Process);

  for I := Length(List) - 1 downto 0 do

    if Result then

    begin

      Handle := OpenProcess(PROCESS_TERMINATE, false, List[I]);

      Result := (Handle <> 0) and TerminateProcess(Handle, 0) and CloseHandle(Handle);

    end;

end;



{ TProcessThread }



procedure TProcessThread.CallDataAvailable;

begin

  if Assigned(FOnDataAvailable) then

    FOnDataAvailable(Self);

end;



procedure TProcessThread.Resume;

var

  SuspendCount: Integer;

begin

  if FStarted then

  begin

    FStarted := True;

    SuspendCount := ResumeThread(FMainProcess.hThread);

    CheckThreadError(SuspendCount >= 0);

    if SuspendCount = 1 then

      FSuspended := False;

  end;

  inherited Resume;

end;



function TProcessThread.GetPriority: TThreadPriority;

begin

  if FStarted then

  begin

    CheckThreadError(GetThreadPriority(FMainProcess.hThread) <> THREAD_PRIORITY_ERROR_RETURN);

  end;

  Result := inherited Priority;

end;



function TProcessThread.IsProcessAlive: Boolean;

var

  Status: Cardinal;

begin

  GetExitCodeProcess(FMainProcess.hProcess, Status);

  Result := Status = STILL_ACTIVE;

end;



procedure TProcessThread.SetPriority(const Value: TThreadPriority);

begin

  if FStarted then

    CheckThreadError(SetThreadPriority(FMainProcess.hThread, Priorities[Value]));

  inherited Priority := Value;

end;



procedure TProcessThread.Suspend;

var

  OldSuspend: Boolean;

begin

  if FStarted then

  begin

    OldSuspend := FSuspended;

    try

      FSuspended := True;

      CheckThreadError(Integer(SuspendThread(FMainProcess.hThread)) >= 0);

    except

      FSuspended := OldSuspend;

      raise;

    end;

  end;

  inherited Suspend;

end;





procedure TProcessThread.CallException;

begin

  if Assigned(FOnException) then

    FOnException(Self, FException);

end;



procedure TProcessThread.CallProcessOpened;

begin

  if Assigned(FOnProcessStarted) then

    FOnProcessStarted(Self);

end;



procedure TProcessThread.CallProcessTerminated;

begin

  if Assigned(FOnProcessTerminated) then

    FOnProcessTerminated(Self);

end;



constructor TProcessThread.Create(Path, CommandLine, Directory: string; Environment: TStrings; Watch: Boolean);

var

  Len, I: Integer;

begin

  inherited Create(True);



  if (Length(CommandLine) > 0) and (Length(Path) > 0) then

    CommandLine := ' ' + CommandLine;



  if Length(Path) > 0 then

  begin

    GetMem(FPath, Length(Path) + 1);

    StrCopy(FPath, PChar(Path));

  end;

  if Length(CommandLine) > 0 then

  begin

    GetMem(FCommandLine, Length(CommandLine) + 1);

    StrCopy(FCommandLine, PChar(CommandLine));

  end;



  if Length(Directory) > 0 then

  begin

    GetMem(FDirectory, Length(Directory) + 1);

    StrCopy(FDirectory, PChar(Directory));

  end;



  FWatching := Watch;



  if Assigned(Environment) then

  begin

    GetMem(FEnvironment, 1);

    Len := 1;

    for I := 0 to Environment.Count - 1 do

    begin

      Inc(Len, Length(Environment[I]) + 1);

      ReallocMem(FEnvironment, Len);

      SetEnvironmentVariable(PChar(Environment.Names[I]), PChar(Environment.ValueFromIndex[I]));

      StrCopy(FEnvironment + Len - Length(Environment[I]) - 2, PChar(Environment[I]));

    end;

    (FEnvironment + Len - 1)^ := #0;

    FreeMem(FEnvironment);

    FEnvironment := nil;

  end;

end;



destructor TProcessThread.Destroy;

begin

  FreeMem(FPath);

  FreeMem(FCommandLine);

  FreeMem(FDirectory);

  FreeResources;

  if Assigned(FEnvironment) then

    FreeMem(FEnvironment);

  inherited;

end;



procedure TProcessThread.Execute;

const

  MAX_BUFFER = 512 * 1024;

var

  MaxBytes, Available, BytesRead: Cardinal;

  Buffer: array[0..MAX_BUFFER] of Char;



  function Read: Boolean;

  begin

    Result := True;

    FillChar(Buffer, MAX_BUFFER, #0);

    PeekNamedPipe(OutputRead, @Buffer, MAX_BUFFER, @BytesRead, @Available, nil);



    if BytesRead < MAX_BUFFER then

    begin

      MaxBytes := BytesRead;

    end

    else

      MaxBytes := MAX_BUFFER;



    if MaxBytes > 0 then

      if ReadFile(OutputRead, Buffer, MaxBytes, BytesRead, nil) then

      begin

        if BytesRead > 0 then

        begin

          FData := StrPas(Buffer);

          Synchronize(CallDataAvailable);

        end;

      end

      else

        Result := False;

  end;



var

  Startup: STARTUPINFO;

  SecurityDescriptor: SECURITY_DESCRIPTOR;

  SecurityAttributes: SECURITY_ATTRIBUTES;

begin

  try

    ZeroMemory(@Startup, SizeOf(STARTUPINFO));

    Startup.cb := SizeOf(STARTUPINFO);

    ZeroMemory(@SecurityDescriptor, SizeOf(SECURITY_DESCRIPTOR));

    ZeroMemory(@SecurityAttributes, SizeOf(SECURITY_ATTRIBUTES));

    InputRead := 0;

    InputWrite := 0;

    OutputRead := 0;

    OutputWrite := 0;



    if Win32Platform = VER_PLATFORM_WIN32_NT then

    begin

      InitializeSecurityDescriptor(@SecurityDescriptor, SECURITY_DESCRIPTOR_REVISION);

      SetSecurityDescriptorDacl(@SecurityDescriptor, True, nil, False);

      SecurityAttributes.lpSecurityDescriptor := @SecurityDescriptor;

    end

    else

      SecurityAttributes.lpSecurityDescriptor := nil;

    SecurityAttributes.nLength := SizeOf(SECURITY_ATTRIBUTES);

    SecurityAttributes.bInheritHandle := True;

    if not CreatePipe(OutputRead, OutputWrite, @SecurityAttributes, 0)

    or not CreatePipe(InputRead, InputWrite, @SecurityAttributes, 0) then

      raise EProcessError.Create('Error while opening pipes');



    SetHandleInformation(OutputRead, HANDLE_FLAG_INHERIT, 0);

    SetHandleInformation(InputWrite, HANDLE_FLAG_INHERIT, 0);



    GetStartupInfo(Startup);

    Startup.dwFlags := STARTF_USESHOWWINDOW OR STARTF_USESTDHANDLES;



    Startup.hStdOutput := OutputWrite;

    Startup.hStdError := OutputWrite;

    Startup.hStdInput := InputRead;



    FlushFileBuffers(OutputWrite);

    FlushFileBuffers(OutputRead);

    FlushFileBuffers(InputRead);

    FlushFileBuffers(InputWrite);



    Startup.wShowWindow := SW_HIDE;



    if not CreateProcess(FPath, FCommandLine, nil, nil, True, CREATE_NEW_CONSOLE OR NORMAL_PRIORITY_CLASS, FEnvironment, FDirectory, Startup, FMainProcess) then

      raise EProcessError.Create('Error while starting Process: ' + SysErrorMessage(GetLastError) + ':' + FPath + ':' + FCommandLine + ':' + FDirectory);

    WaitForInputIdle(FMainProcess.hProcess, INITIALIZATION_TIMEOUT);

    FStarted := True;

    SetPriority(GetPriority);

    Synchronize(CallProcessOpened);

    

    if not FWatching then

      Exit;



    repeat

      if not Read then

        Break;

    until not IsProcessAlive or Terminated;

    Read;



    if not IsProcessAlive then

      Synchronize(CallProcessTerminated);

      

  except

    on E: Exception do

    begin

      FException := E;

      Synchronize(CallException);

    end;

  end;

end;



procedure TProcessThread.FreeResources;

begin

  KillProcess(FMainProcess.dwProcessId);



  if OutputRead <> 0 then

  begin

    CloseHandle(OutputRead);

    OutputRead := 0;

  end;

  if OutputWrite <> 0 then

  begin

    CloseHandle(OutputWrite);

    OutputWrite := 0;

  end;

  if InputWrite <> 0 then

  begin

    CloseHandle(InputWrite);

    InputWrite := 0;

  end;

  if InputRead <> 0 then

  begin

    CloseHandle(InputRead);

    InputRead := 0;

  end;

end;



{  TProcessLineThread }



constructor  TProcessLineThread.Create(Path, CommandLine, Directory: string; Environment: TStrings);

begin

  inherited Create(Path, CommandLine, Directory, Environment);

  OnDataAvailable := DataAvailable;

  OnTerminate := Finished;

end;



procedure  TProcessLineThread.DataAvailable(Sender: TObject);

var

  I, L: Integer;

begin

  I := 0;

  L := Length(Data);

  while I < L do

  begin

    Inc(I);

    if Data[I] in [#13, #10] then

    begin

      if (I < L) and (Data[I+1] in [#13, #10]) then

        Inc(I);

      if Assigned(FOnNewLine) then

        FOnNewLine(Self, FCurrentLine);

      FCurrentLine := '';

    end

    else

      FCurrentLine := FCurrentLine + Data[I];

  end;

end;



procedure  TProcessLineThread.Finished(Sender: TObject);

begin

  if (FCurrentLine <> '') and Assigned(FOnNewLine) then

    FOnNewLine(Self, FCurrentLine);

end;



end.

gzip pipe

cdvddt (Raphael Jolivet) — Sun, 15 Jun 2008 08:50:59 GMT

#! /usr/bin/python

from gzip import GzipFile
from StringIO import StringIO

class GZipPipe(StringIO) :
"""This class implements a compression pipe suitable for asynchronous
process.

Only one buffer of data is read/compressed at a time.
The process doesn't read the whole file at once : This improves performance
and prevent hight memory consumption for big files."""

# Size of the internal buffer
CHUNCK_SIZE = 1024

def __init__(self, source = None, name = "data") :
"""Constructor

@param source Source data to compress (as a stream/File/Buffer - anything with a read() method)
@param name Name of the data within the zip file"""

# Source file
self.source = source

# OEF reached for source ?
self.source_eof = False

# Buffer
self.buffer = ""

StringIO.__init__(self)

# Inherited constructor

# Init ZipFile that writes to us (the StringIO buffer)
self.zipfile = GzipFile(name, 'wb', 9, self)

def write(self, data) :
"""The write mzthod shouldn't be called from outside.
A GZipFile was created with this current object as a output buffer anbd it
fills it whenever we write to it (calling the read method of this object will do it for you)
"""

self.buffer += data

def read(self, size = -1) :
"""Calling read() on a zip pipe will suck data from the source stream.

@param size Maximum size to read - Read whole compressed file if not specified.
@return Compressed data"""

# Feed the zipped buffer by writing source data to the zip stream
while ((len(self.buffer) < size) or (size == -1)) and not self.source_eof :

# No source given in input
if self.source == None: break

# Get a chunk of source data
chunk = self.source.read(GZipPipe.CHUNCK_SIZE)

# Feed the source zip file (that fills the compressed buffer)
self.zipfile.write(chunk)

# End of source file ?
if (len(chunk) < GZipPipe.CHUNCK_SIZE) :
self.source_eof = True
self.zipfile.flush()
self.zipfile.close()
break

# We have enough data in the buffer (or source file is EOF): Give it to the output
if size == 0:
result = ""
if size >= 1 :
result = self.buffer[0:size]
self.buffer = self.buffer[size:]
else : # size < 0 : All requested
result = self.buffer
self.buffer = ""

return result

Twitter bot weatherlisbon

griflet (guillaume riflet) — Mon, 18 Jun 2007 21:37:40 GMT

This little bash script shows how to use curl, grep, tail, sed and perl one-liners in order to compose a bleeding-edge twitter bot.
This one returns daily weather forecasts for Lisbon city based on the BBC weather forecast rss feed.



#! /bin/sh



#Goto here

here=/home/guillaume/Personal

cd $here



#BBC Lisbon weather id

id=0048



#BBC weather RSS feed address

feed="http://feeds.bbc.co.uk/weather/feeds/rss/5day/world/${id}.xml"



#City

city=lisbon



#temporary file

file="weather${city}"



#Weather twitter bot

twitbot=weatherlisbon:*******



#Timestamp the log file

echo .>> $file.log

date >> $file.log



#Read the RSS feed and filter it

curl $feed | grep 'title' | tail -n 1 | perl -wlne'm/title>(.*)<\/title/i && print $1' | sed -e "s/°//g" > $file.txt



#Read the forecast into a weather variable

read weather < $file.txt



#Twit the weather variable away

curl --basic --user $twitbot --data status="$weather" http://twitter.com/statuses/update.xml >> $file.log

Making md5sum.exe Work with Paths in Python

minitotoro (Natalie) — Wed, 13 Jun 2007 23:10:02 GMT

On Windows systems, md5sum.exe is a nice little program to generate MD5 sums. However, whoever created this application forgot to add the ability to recognize paths in the file given to md5sum.exe. For example, say you want to md5 a file called test.txt.

C:\test>dir
Directory of C:\test

06/13/2007 03:52 PM .
06/13/2007 03:52 PM ..
06/13/2007 03:52 PM 0 test.txt
1 File(s) 0 bytes

C:\test>md5sum test.txt
d41d8cd98f00b204e9800998ecf8427e *test.txt

Groovy, cool...

However, if you try the same thing from some other directory (where text.txt does not live) you get the following:

C:\>md5sum c:/test/test.txt
md5sum: test.txt: No such file or directory

Sucky, eh?
You could always find a different program to do md5sum. Or, if you don't want to do this, you can spawn a pipe to "cmd", navigate to the correct directory, and then run md5sum.

Here's the code in python to do just that.



# define your file name and path

file_name = "afile.txt"

file_path = "c:\\afolder"



# setup the md5sum command (assuming md5sum.exe location is in PATH)

md5_cmd = "md5sum \"" + file_path "\\" + file_name + "\"\n"



# open the command shell

fromchild, tochild = popen2.popen4("cmd")



#push the directory change and md5sum commands to the shell 

tochild.write("c:\nchdir " + my_path + "\n" + md5_cmd + "exit\n")

tochild.close()



#get the output from the shell session

out = fromchild.read()



#split the output so that we may extract the md5sum

output = string.split(out)



#grab the md5sum  

md5sum_local = ""

for item in output:

    matchmd5local = re.match("([0-9a-fA-F]{32})",item)

    if matchmd5local:

        md5sum_local = matchmd5local.groups()

        md5sum_local = md5sum_local[0]



if md5sum_local:

    print_status("MD5: Local checksum = " + md5sum_local + "\n")

else:

    print_status("ERROR: could not obtain local md5sum for " + my_file + "\n")