The correct answer is:

p.communicate(b"insert into egg values ('egg')")

Note the leading b, telling you that it's a string of bytes, not a string of unicode characters. Also, if you are reading this from a file:

value = open('thefile', 'rt').read()
p.communicate(value)

The change that to:

value = open('thefile', 'rb').read()
p.communicate(value)

Again, note the 'b'. Now if your value is a string you get from an API that only returns strings no matter what, then you need to encode it.

p.communicate(value.encode('latin-1'))

Latin-1, because unlike ASCII it supports all 256 bytes. But that said, having binary data in unicode is asking for trouble. It's better if you can make it binary from the start.

In vimscript, just as an alternative:

:1,$-1substitute/$/:
:global/^/join

In one line with abbreviated commands:

:1,$-1s/$/:/ | g/^/j

In my day job, I wrote the software for an embedded system and a PC communicating with each other by a USB cable, using the UART protocol at 115,200 baud.

I see that you tagged your post with PySerial, so you already know about Python's most popular package for serial port communication. I will add that if you are using PyQt, there's a serial module included in that package as well.

115,200 baud is not fast for a modern desktop PC. I doubt that any parsing you do on the PC side will fail to keep up. I parse data streams and plot graphs of my data in real time using PyQt.

What I have noticed in my work with communication between an embedded system and a PC over a UART is that some data gets corrupted occasionally. A byte can be garbled, repeated, or dropped. Also, even if no bytes are added or dropped, you can occasionally perform a read while only part of a packet is in the buffer, and the read will terminate early. If you use a fixed read length of 40 bytes and trust that each read will always line up exactly with a data packet as you show above, you will frequently be wrong.

To solve these kinds of problems, I wrote a FIFO class in Python which consumes serial port data at the head of the FIFO, yields valid data packets at the tail, and discards invalid data. My FIFO holds 3 times as many bytes as my data packets, so if I am looking for packet boundaries using specific sequences, I have plenty of signposts.

A few more recommendations: work in Python 3 if you have the choice, it's cleaner. Use bytes and bytearray objects. Don't use str, because you will find yourself converting back and forth between Unicode and ASCII.

Here's a slightly better way. Cast the pointer to (single) wchar to a pointer to an array of wchar. Then you can dereference and write the value directly to the array:

def callback(pt: POINTER(c_wchar), ptsize: c_size_t) -> None:
    pa = cast(pt,POINTER(c_wchar * ptsize))
    pa.contents.value = 'python string'

This also has the advantage that if you write more that ptsize characters to the array, Python will throw an exception. For example:

def callback(pt: POINTER(c_wchar), ptsize: c_size_t) -> None:
    s = 'a' * (ptsize + 1)  # one too big
    pa = cast(pt,POINTER(c_wchar * ptsize))
    pa.contents.value = s

This results in:

Exception ignored on calling ctypes callback function: <function callback at 0x0000020B1C7F6310>
Traceback (most recent call last):
  File "C:\test.py", line 13, in callback
    pa.contents.value = s
ValueError: string too long

Be careful of the null termination, though. Python only writes it via .value if it has room in the array. Here's a demo:

test.cpp

#include <iostream>
#include <stdio.h>

using namespace std;

typedef void(__stdcall* Mycallback2)(wchar_t* buff, size_t buffsize);

extern "C" __declspec(dllexport)
void TestMethod5(Mycallback2 callback)
{
    wchar_t* buff = new wchar_t[7];        // length 7 buffer
    swprintf_s(buff, 7, L"%s", L"123456"); // init buffer
    if (callback) callback(buff, 5);       // callback alters only 5
    wcout << L'|' << buff << L'|' << endl;
    delete[] buff;
}

test.py

from ctypes import *

dll = CDLL('./test')

MYCALLBACKTYPE = CFUNCTYPE(None, POINTER(c_wchar), c_size_t)
dll.TestMethod5.restype = None
dll.TestMethod5.argtypes = [MYCALLBACKTYPE]

def callback(pt: POINTER(c_wchar), ptsize: c_size_t) -> None:
    s = 'a' * (ptsize if demo else ptsize-1) # 4 or 5 'a' string, no explicit null
    pa = cast(pt,POINTER(c_wchar * ptsize))
    print('pt =',pa.contents.value)
    pa.contents.value = s

mycallback = MYCALLBACKTYPE(callback)
demo = 0
dll.TestMethod5(mycallback)
demo = 1
dll.TestMethod5(mycallback)

Output:

pt = 12345         # only "sees" the size reported
|aaaa|             # null was written since there was room
pt = 12345
|aaaaa6|           # null wasn't written when length 5 value written,
                   # So wcout "saw" the ending 6.

Python 3

From the docs:

Concatenating immutable sequences always results in a new object. This means that building up a sequence by repeated concatenation will have a quadratic runtime cost in the total sequence length. To get a linear runtime cost, you must switch to one of the alternatives below: if concatenating str objects, you can build a list and use str.join() at the end or else write to an io.StringIO instance and retrieve its value when complete

Experiment to compare runtime of several options:

import sys
import timeit
from io import StringIO
from array import array


def test_concat():
    out_str = ''
    for _ in range(loop_count):
        out_str += 'abc'
    return out_str


def test_join_list_loop():
    str_list = []
    for _ in range(loop_count):
        str_list.append('abc')
    return ''.join(str_list)


def test_array():
    char_array = array('b')
    for _ in range(loop_count):
        char_array.frombytes(b'abc')
    return str(char_array.tostring())


def test_string_io():
    file_str = StringIO()
    for _ in range(loop_count):
        file_str.write('abc')
    return file_str.getvalue()


def test_join_list_compr():
    return ''.join(['abc' for _ in range(loop_count)])


def test_join_gen_compr():
    return ''.join('abc' for _ in range(loop_count))


loop_count = 80000

print(sys.version)

res = {}

for k, v in dict(globals()).items():
    if k.startswith('test_'):
        res[k] = timeit.timeit(v, number=10)

for k, v in sorted(res.items(), key=lambda x: x[1]):
    print('{:.5f} {}'.format(v, k))

results

3.7.5 (default, Nov  1 2019, 02:16:32) 
[Clang 11.0.0 (clang-1100.0.33.8)]
0.03738 test_join_list_compr
0.05681 test_join_gen_compr
0.09425 test_string_io
0.09636 test_join_list_loop
0.11976 test_concat
0.19267 test_array

Python 2

Efficient String Concatenation in Python is a rather old article and its main statement that the naive concatenation is far slower than joining is not valid anymore, because this part has been optimized in CPython since then. From the docs:

CPython implementation detail: If s and t are both strings, some Python implementations such as CPython can usually perform an in-place optimization for assignments of the form s = s + t or s += t. When applicable, this optimization makes quadratic run-time much less likely. This optimization is both version and implementation dependent. For performance sensitive code, it is preferable to use the str.join() method which assures consistent linear concatenation performance across versions and implementations.

I've adapted their code a bit and got the following results on my machine:

from cStringIO import StringIO
from UserString import MutableString
from array import array

import sys, timeit

def method1():
    out_str = ''
    for num in xrange(loop_count):
        out_str += `num`
    return out_str

def method2():
    out_str = MutableString()
    for num in xrange(loop_count):
        out_str += `num`
    return out_str

def method3():
    char_array = array('c')
    for num in xrange(loop_count):
        char_array.fromstring(`num`)
    return char_array.tostring()

def method4():
    str_list = []
    for num in xrange(loop_count):
        str_list.append(`num`)
    out_str = ''.join(str_list)
    return out_str

def method5():
    file_str = StringIO()
    for num in xrange(loop_count):
        file_str.write(`num`)
    out_str = file_str.getvalue()
    return out_str

def method6():
    out_str = ''.join([`num` for num in xrange(loop_count)])
    return out_str

def method7():
    out_str = ''.join(`num` for num in xrange(loop_count))
    return out_str


loop_count = 80000

print sys.version

print 'method1=', timeit.timeit(method1, number=10)
print 'method2=', timeit.timeit(method2, number=10)
print 'method3=', timeit.timeit(method3, number=10)
print 'method4=', timeit.timeit(method4, number=10)
print 'method5=', timeit.timeit(method5, number=10)
print 'method6=', timeit.timeit(method6, number=10)
print 'method7=', timeit.timeit(method7, number=10)

Results:

2.7.1 (r271:86832, Jul 31 2011, 19:30:53) 
[GCC 4.2.1 (Based on Apple Inc. build 5658) (LLVM build 2335.15.00)]
method1= 0.171155929565
method2= 16.7158739567
method3= 0.420584917068
method4= 0.231794118881
method5= 0.323612928391
method6= 0.120429992676
method7= 0.145267963409

Conclusions:

• join still wins over concat, but marginally
• list comprehensions are faster than loops (when building a list)
• joining generators is slower than joining lists
• other methods are of no use (unless you're doing something special)

Decode the bytes object to produce a string:

>>> b"abcde".decode("utf-8")
'abcde'

The above example assumes that the bytes object is in UTF-8, because it is a common encoding. However, you should use the encoding your data is actually in!