def CopyFileToCe(src_name, dest_name, progress = None):
sh = win32file.CreateFile(src_name, win32con.GENERIC_READ, 0, None, win32con.OPEN_EXISTING, 0, None)
bytes=0
try:
dh = wincerapi.CeCreateFile(dest_name, win32con.GENERIC_WRITE, 0, None, win32con.OPEN_ALWAYS, 0, None)
try:
while 1:
hr, data = win32file.ReadFile(sh, 2048)
if not data:
break
wincerapi.CeWriteFile(dh, data)
bytes = bytes + len(data)
if progress is not None: progress(bytes)
finally:
pass
dh.Close()
finally:
sh.Close()
return bytes
python类GENERIC_WRITE的实例源码
def testTransactNamedPipeBlocking(self):
event = threading.Event()
self.startPipeServer(event)
open_mode = win32con.GENERIC_READ | win32con.GENERIC_WRITE
hpipe = win32file.CreateFile(self.pipename,
open_mode,
0, # no sharing
None, # default security
win32con.OPEN_EXISTING,
0, # win32con.FILE_FLAG_OVERLAPPED,
None)
# set to message mode.
win32pipe.SetNamedPipeHandleState(
hpipe, win32pipe.PIPE_READMODE_MESSAGE, None, None)
hr, got = win32pipe.TransactNamedPipe(hpipe, str2bytes("foo\0bar"), 1024, None)
self.failUnlessEqual(got, str2bytes("bar\0foo"))
event.wait(5)
self.failUnless(event.isSet(), "Pipe server thread didn't terminate")
def testTransactNamedPipeBlockingBuffer(self):
# Like testTransactNamedPipeBlocking, but a pre-allocated buffer is
# passed (not really that useful, but it exercises the code path)
event = threading.Event()
self.startPipeServer(event)
open_mode = win32con.GENERIC_READ | win32con.GENERIC_WRITE
hpipe = win32file.CreateFile(self.pipename,
open_mode,
0, # no sharing
None, # default security
win32con.OPEN_EXISTING,
0, # win32con.FILE_FLAG_OVERLAPPED,
None)
# set to message mode.
win32pipe.SetNamedPipeHandleState(
hpipe, win32pipe.PIPE_READMODE_MESSAGE, None, None)
buffer = win32file.AllocateReadBuffer(1024)
hr, got = win32pipe.TransactNamedPipe(hpipe, str2bytes("foo\0bar"), buffer, None)
self.failUnlessEqual(got, str2bytes("bar\0foo"))
event.wait(5)
self.failUnless(event.isSet(), "Pipe server thread didn't terminate")
def CopyFileToCe(src_name, dest_name, progress = None):
sh = win32file.CreateFile(src_name, win32con.GENERIC_READ, 0, None, win32con.OPEN_EXISTING, 0, None)
bytes=0
try:
dh = wincerapi.CeCreateFile(dest_name, win32con.GENERIC_WRITE, 0, None, win32con.OPEN_ALWAYS, 0, None)
try:
while 1:
hr, data = win32file.ReadFile(sh, 2048)
if not data:
break
wincerapi.CeWriteFile(dh, data)
bytes = bytes + len(data)
if progress is not None: progress(bytes)
finally:
pass
dh.Close()
finally:
sh.Close()
return bytes
def testTransactNamedPipeBlocking(self):
event = threading.Event()
self.startPipeServer(event)
open_mode = win32con.GENERIC_READ | win32con.GENERIC_WRITE
hpipe = win32file.CreateFile(self.pipename,
open_mode,
0, # no sharing
None, # default security
win32con.OPEN_EXISTING,
0, # win32con.FILE_FLAG_OVERLAPPED,
None)
# set to message mode.
win32pipe.SetNamedPipeHandleState(
hpipe, win32pipe.PIPE_READMODE_MESSAGE, None, None)
hr, got = win32pipe.TransactNamedPipe(hpipe, str2bytes("foo\0bar"), 1024, None)
self.failUnlessEqual(got, str2bytes("bar\0foo"))
event.wait(5)
self.failUnless(event.isSet(), "Pipe server thread didn't terminate")
def testTransactNamedPipeBlockingBuffer(self):
# Like testTransactNamedPipeBlocking, but a pre-allocated buffer is
# passed (not really that useful, but it exercises the code path)
event = threading.Event()
self.startPipeServer(event)
open_mode = win32con.GENERIC_READ | win32con.GENERIC_WRITE
hpipe = win32file.CreateFile(self.pipename,
open_mode,
0, # no sharing
None, # default security
win32con.OPEN_EXISTING,
0, # win32con.FILE_FLAG_OVERLAPPED,
None)
# set to message mode.
win32pipe.SetNamedPipeHandleState(
hpipe, win32pipe.PIPE_READMODE_MESSAGE, None, None)
buffer = win32file.AllocateReadBuffer(1024)
hr, got = win32pipe.TransactNamedPipe(hpipe, str2bytes("foo\0bar"), buffer, None)
self.failUnlessEqual(got, str2bytes("bar\0foo"))
event.wait(5)
self.failUnless(event.isSet(), "Pipe server thread didn't terminate")
def __init__(self):
self.pty = Pty()
self.ready_f = Future()
self._input_ready_callbacks = []
self.loop = get_event_loop()
self.stdout_handle = win32file.CreateFile(
self.pty.conout_name(),
win32con.GENERIC_READ,
0,
win32security.SECURITY_ATTRIBUTES(),
win32con.OPEN_EXISTING,
win32con.FILE_FLAG_OVERLAPPED,
0)
self.stdin_handle = win32file.CreateFile(
self.pty.conin_name(),
win32con.GENERIC_WRITE,
0,
win32security.SECURITY_ATTRIBUTES(),
win32con.OPEN_EXISTING,
win32con.FILE_FLAG_OVERLAPPED,
0)
self._buffer = []
def w32CreateFile(name, access=GENERIC_READ|GENERIC_WRITE,
flags=OPEN_BY_SERIAL_NUMBER):
return FTD2XX(_ft.FT_W32_CreateFile(_ft.STRING(name),
_ft.DWORD(access),
_ft.DWORD(0),
None,
_ft.DWORD(OPEN_EXISTING),
_ft.DWORD(flags),
_ft.HANDLE(0)))
def DemoCopyFile():
# Create a file on the device, and write a string.
cefile = wincerapi.CeCreateFile("TestPython", win32con.GENERIC_WRITE, 0, None, win32con.OPEN_ALWAYS, 0, None)
wincerapi.CeWriteFile(cefile, "Hello from Python")
cefile.Close()
# reopen the file and check the data.
cefile = wincerapi.CeCreateFile("TestPython", win32con.GENERIC_READ, 0, None, win32con.OPEN_EXISTING, 0, None)
if wincerapi.CeReadFile(cefile, 100) != "Hello from Python":
print "Couldnt read the data from the device!"
cefile.Close()
# Delete the test file
wincerapi.CeDeleteFile("TestPython")
print "Created, wrote to, read from and deleted a test file!"
def testTransactNamedPipeAsync(self):
event = threading.Event()
overlapped = pywintypes.OVERLAPPED()
overlapped.hEvent = win32event.CreateEvent(None, 0, 0, None)
self.startPipeServer(event, 0.5)
open_mode = win32con.GENERIC_READ | win32con.GENERIC_WRITE
hpipe = win32file.CreateFile(self.pipename,
open_mode,
0, # no sharing
None, # default security
win32con.OPEN_EXISTING,
win32con.FILE_FLAG_OVERLAPPED,
None)
# set to message mode.
win32pipe.SetNamedPipeHandleState(
hpipe, win32pipe.PIPE_READMODE_MESSAGE, None, None)
buffer = win32file.AllocateReadBuffer(1024)
hr, got = win32pipe.TransactNamedPipe(hpipe, str2bytes("foo\0bar"), buffer, overlapped)
self.failUnlessEqual(hr, winerror.ERROR_IO_PENDING)
nbytes = win32file.GetOverlappedResult(hpipe, overlapped, True)
got = buffer[:nbytes]
self.failUnlessEqual(got, str2bytes("bar\0foo"))
event.wait(5)
self.failUnless(event.isSet(), "Pipe server thread didn't terminate")
def DemoCopyFile():
# Create a file on the device, and write a string.
cefile = wincerapi.CeCreateFile("TestPython", win32con.GENERIC_WRITE, 0, None, win32con.OPEN_ALWAYS, 0, None)
wincerapi.CeWriteFile(cefile, "Hello from Python")
cefile.Close()
# reopen the file and check the data.
cefile = wincerapi.CeCreateFile("TestPython", win32con.GENERIC_READ, 0, None, win32con.OPEN_EXISTING, 0, None)
if wincerapi.CeReadFile(cefile, 100) != "Hello from Python":
print("Couldnt read the data from the device!")
cefile.Close()
# Delete the test file
wincerapi.CeDeleteFile("TestPython")
print("Created, wrote to, read from and deleted a test file!")
def testTransactNamedPipeAsync(self):
event = threading.Event()
overlapped = pywintypes.OVERLAPPED()
overlapped.hEvent = win32event.CreateEvent(None, 0, 0, None)
self.startPipeServer(event, 0.5)
open_mode = win32con.GENERIC_READ | win32con.GENERIC_WRITE
hpipe = win32file.CreateFile(self.pipename,
open_mode,
0, # no sharing
None, # default security
win32con.OPEN_EXISTING,
win32con.FILE_FLAG_OVERLAPPED,
None)
# set to message mode.
win32pipe.SetNamedPipeHandleState(
hpipe, win32pipe.PIPE_READMODE_MESSAGE, None, None)
buffer = win32file.AllocateReadBuffer(1024)
hr, got = win32pipe.TransactNamedPipe(hpipe, str2bytes("foo\0bar"), buffer, overlapped)
self.failUnlessEqual(hr, winerror.ERROR_IO_PENDING)
nbytes = win32file.GetOverlappedResult(hpipe, overlapped, True)
got = buffer[:nbytes]
self.failUnlessEqual(got, str2bytes("bar\0foo"))
event.wait(5)
self.failUnless(event.isSet(), "Pipe server thread didn't terminate")
def win32_comports_bruteforce():
import win32file
import win32con
ports = []
for i in range(1, 257):
portname = "\\\\.\\COM%i" % i
try:
mode = win32con.GENERIC_READ | win32con.GENERIC_WRITE
port = \
win32file.CreateFile(portname,
mode,
win32con.FILE_SHARE_READ,
None,
win32con.OPEN_EXISTING,
0,
None)
if portname.startswith("\\"):
portname = portname[4:]
ports.append((portname, "Unknown", "Serial"))
win32file.CloseHandle(port)
port = None
except Exception, e:
pass
return ports
def write_mbr_winapi( _file ):
print 'Are you SURE you want to overwrite the MBR?? This will possibly make the volume unbootable.'
response = raw_input( 'Type \"YES\" then Return to continue, anything else then Return to not continue:' )
if response != 'YES':
return
h = None
handles = []
try:
for x in range( num_mbr_handles ):
h = win32file.CreateFile( '\\\\.\\PhysicalDrive0',
win32con.GENERIC_WRITE,
win32file.FILE_SHARE_WRITE,
None,
win32file.OPEN_EXISTING,
win32file.FILE_ATTRIBUTE_NORMAL,
None )
if ( h != win32file.INVALID_HANDLE_VALUE ):
handles.append( h )
f = open( _file, 'rb' )
if f <> None:
fsize = os.path.getsize( _file )
wsize = 512
if fsize > 512:
print 'WARNING: File being written is > 512 bytes, will only write 512...'
wsize = 512
contents = f.read( fsize )
if fsize < 512:
print 'WARNING: Padding file up to 512 bytes, may not have expected results...'
## pad it out to 512 bytes
diff = 512 - 512
for num in xrange( diff ):
contents += 'A'
win32file.WriteFile( h, contents, None )
f.close()
except Exception, e:
print str( e )
print '\tAre you running as Administrator?'
for handle in handles:
win32file.CloseHandle( handle )
#############
